vlog.go: Fix wrongly maintain the offset of file in memory

go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/Shopify/toxiproxy v2.1.3+incompatible // indirect
 	github.com/codahale/hdrhistogram v0.0.0-20161010025455-3a0bb77429bd // indirect
 	github.com/coreos/etcd v3.3.0-rc.0.0.20180530235116-2b3aa7e1d49d+incompatible
+	github.com/docker/docker v1.13.1
 	github.com/dustin/go-humanize v1.0.0 // indirect
 	github.com/eapache/go-resiliency v1.1.0 // indirect
 	github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 // indirect

go.sum

@@ -35,6 +35,8 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
+github.com/docker/docker v1.13.1 h1:IkZjBSIc8hBjLpqeAbeE5mca5mNgeatLHBy3GO78BWo=
+github.com/docker/docker v1.13.1/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/dustin/go-humanize v0.0.0-20180421182945-02af3965c54e/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=
 github.com/dustin/go-humanize v1.0.0 h1:VSnTsYCnlFHaM2/igO1h6X3HA71jcobQuxemgkq4zYo=
 github.com/dustin/go-humanize v1.0.0/go.mod h1:HtrtbFcZ19U5GC7JDqmcUSB87Iq5E25KnS6fMYU6eOk=

pump/storage/vlog.go

@@ -329,6 +329,8 @@ func (vlog *valueLog) write(reqs []*request) error {
 
 	toDisk := func() error {
 		n, err := curFile.fd.Write(vlog.buf.Bytes())
+		atomic.AddInt64(&vlog.writableLogOffset, int64(n))
+
 		if err != nil {
 			return errors.Annotatef(err, "unable to write to log file: %s", curFile.path)
 		}
@@ -339,7 +341,6 @@ func (vlog *valueLog) write(reqs []*request) error {
 			}
 		}
 
-		atomic.AddInt64(&vlog.writableLogOffset, int64(n))
 		for _, req := range bufReqs {
 			curFile.updateMaxTS(req.ts())
 		}

pump/storage/vlog_linux_test.go

@@ -0,0 +1,86 @@
+// Copyright 2019 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// +build linux
+
+package storage
+
+import (
+	"fmt"
+	"io/ioutil"
+	"os"
+	"path"
+	"strings"
+
+	"github.com/docker/docker/pkg/mount"
+	"github.com/pingcap/check"
+)
+
+// Need root privilege for using mount to get a specify tmpfs file system
+func (vs *VlogSuit) TestNoSpace(c *check.C) {
+	dir := c.MkDir()
+	c.Log("use dir: ", dir)
+
+	size := "40k"
+	err := mount.ForceMount("tmpfs", dir, "tmpfs", fmt.Sprintf("size=%s", size))
+	if strings.Contains(err.Error(), "operation not permitted") {
+		c.Skip("operation not permitted to using mount")
+	}
+
+	c.Assert(err, check.IsNil)
+	defer func() {
+		err := mount.Unmount(dir)
+		if err != nil {
+			c.Log(err)
+		}
+	}()
+
+	// occupy 20k, leaving 20k available
+	occupyFile := path.Join(dir, "dummmy")
+	err = ioutil.WriteFile(occupyFile, make([]byte, 20*1024), 0644)
+	c.Assert(err, check.IsNil)
+
+	vlog := new(valueLog)
+	err = vlog.open(dir, DefaultOptions())
+	c.Assert(err, check.IsNil)
+
+	// 1k payload per record
+	payload := make([]byte, 1<<10)
+	req := &request{
+		payload: payload,
+	}
+
+	// should be enough space to write 19 records
+	for i := 0; i < 19; i++ {
+		err = vlog.write([]*request{req})
+		c.Assert(err, check.IsNil)
+	}
+
+	// failed because only 20k space available and may write a incomplete record
+	err = vlog.write([]*request{req})
+	c.Assert(err, check.NotNil)
+
+	// free space and write again
+	err = os.Remove(occupyFile)
+	c.Assert(err, check.IsNil)
+
+	err = vlog.write([]*request{req})
+	c.Assert(err, check.IsNil)
+
+	// read back normally
+	_, err = vlog.readValue(req.valuePointer)
+	c.Assert(err, check.IsNil)
+
+	err = vlog.close()
+	c.Assert(err, check.IsNil)
+}