Wal flock (#177)

* feat: use flock for WAL

* fix

* save

* save

Author: coanor

diskcache/diskcache.go

@@ -90,8 +90,8 @@ type DiskCache struct {
 	rlock  *InstrumentedMutex // read-lock: used to exclude concurrent Get on the tail file.
 	rwlock *InstrumentedMutex // used to exclude switch/rotate/drop/Close on current disk cache instance.
 
-	flock *flock // disabled multi-Open on same path
-	pos   *pos   // current read fd position info
+	flock *walLock // disabled multi-Open on same path
+	pos   *pos     // current read fd position info
 
 	// specs of current diskcache
 	size atomic.Int64 // current byte size

diskcache/flock.go

@@ -0,0 +1,24 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+package diskcache
+
+import (
+	"os"
+	"path/filepath"
+)
+
+type walLock struct {
+	file string
+	f    *os.File
+}
+
+func newFlock(path string) *walLock {
+	file := filepath.Clean(filepath.Join(path, ".lock"))
+
+	return &walLock{
+		file: file,
+	}
+}

diskcache/lock_test.go diskcache/flock_test.godiskcache/lock_test.go renamed to diskcache/flock_test.go

@@ -7,31 +7,33 @@ package diskcache
 
 import (
 	"os"
-	"runtime"
+	"path/filepath"
 	"sync"
 	T "testing"
 	"time"
 
 	"github.com/stretchr/testify/assert"
 )
 
-func TestPidAlive(t *T.T) {
-	t.Run("pid-1", func(t *T.T) {
-		if runtime.GOOS != "windows" {
-			assert.True(t, pidAlive(1))
-		}
-	})
+func TestLockUnlock(t *T.T) {
+	t.Run("unlock-remove", func(t *T.T) {
+		p := t.TempDir()
+		fl := newFlock(p)
 
-	t.Run("pid-not-exist", func(t *T.T) {
-		assert.False(t, pidAlive(-1))
-	})
+		ok, err := fl.tryLock()
+		assert.True(t, ok)
+		assert.NoError(t, err)
+
+		fi, err := os.Stat(filepath.Join(p, ".lock"))
+		assert.NoError(t, err)
+		t.Logf("fi: %+#v", fi)
 
-	t.Run("cur-pid", func(t *T.T) {
-		assert.True(t, pidAlive(os.Getpid()))
+		fl.unlock()
+
+		_, err = os.Stat(filepath.Join(p, ".lock"))
+		assert.Error(t, err)
 	})
-}
 
-func TestLockUnlock(t *T.T) {
 	t.Run("lock", func(t *T.T) {
 		p := t.TempDir()
 
@@ -42,7 +44,10 @@ func TestLockUnlock(t *T.T) {
 			defer wg.Done()
 			fl := newFlock(p)
 
-			assert.NoError(t, fl.lock())
+			ok, err := fl.tryLock()
+
+			assert.True(t, ok)
+			assert.NoError(t, err)
 			defer fl.unlock()
 
 			time.Sleep(time.Second * 5)
@@ -54,7 +59,8 @@ func TestLockUnlock(t *T.T) {
 			defer wg.Done()
 			fl := newFlock(p)
 
-			err := fl.lock()
+			ok, err := fl.tryLock()
+			assert.False(t, ok)
 			assert.Error(t, err)
 
 			t.Logf("[expect] err: %s", err.Error())
@@ -68,7 +74,7 @@ func TestLockUnlock(t *T.T) {
 
 			// try lock until ok
 			for {
-				if err := fl.lock(); err != nil {
+				if ok, err := fl.tryLock(); !ok {
 					t.Logf("[expect] err: %s", err.Error())
 					time.Sleep(time.Second)
 				} else {

diskcache/flock_unix.go

@@ -0,0 +1,57 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+//go:build !windows
+
+package diskcache
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"syscall"
+)
+
+func (wl *walLock) tryLock() (bool, error) {
+	f, err := os.OpenFile(wl.file, os.O_CREATE|os.O_RDWR, 0o666) // nolint: gosec
+	if err != nil {
+		return false, err
+	}
+	wl.f = f
+
+	// LOCK_EX = Exclusive, LOCK_NB = Non-blocking
+	err = syscall.Flock(int(wl.f.Fd()), syscall.LOCK_EX|syscall.LOCK_NB)
+	if err != nil {
+		// If the error is EWOULDBLOCK, it means someone else has the lock
+		if errors.Is(err, syscall.EWOULDBLOCK) {
+			if err := wl.f.Close(); err != nil {
+				l.Errorf("Close: %s", err.Error())
+			}
+			return false, fmt.Errorf("locked")
+		}
+
+		if err := wl.f.Close(); err != nil {
+			l.Errorf("Close: %s", err.Error())
+		}
+		return false, err
+	}
+	return true, nil
+}
+
+func (wl *walLock) unlock() {
+	if wl.f != nil {
+		if err := syscall.Flock(int(wl.f.Fd()), syscall.LOCK_UN); err != nil {
+			l.Errorf("Flock: %s", err.Error())
+		}
+
+		if err := wl.f.Close(); err != nil {
+			l.Errorf("CLose: %s", err.Error())
+		}
+
+		if err := os.Remove(wl.file); err != nil { // Optional on Unix
+			l.Errorf("Remove: %s", err.Error())
+		}
+	}
+}

diskcache/flock_windows.go

@@ -0,0 +1,61 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+//go:build windows
+// +build windows
+
+package diskcache
+
+import (
+	"os"
+	"syscall"
+	"unsafe"
+)
+
+var (
+	modkernel32    = syscall.NewLazyDLL("kernel32.dll")
+	procLockFileEx = modkernel32.NewProc("LockFileEx")
+)
+
+func (wl *walLock) tryLock() (bool, error) {
+	f, err := os.OpenFile(wl.file, os.O_CREATE|os.O_RDWR, 0o666)
+	if err != nil {
+		return false, err
+	}
+	wl.f = f
+
+	// LOCKFILE_EXCLUSIVE_LOCK = 2, LOCKFILE_FAIL_IMMEDIATELY = 1
+	flags := uint32(2 | 1)
+	var overlapped syscall.Overlapped
+
+	// Call Win32 LockFileEx
+	ret, _, err := procLockFileEx.Call(
+		uintptr(wl.f.Fd()),
+		uintptr(flags),
+		0, // reserved
+		0, // length low
+		1, // length high (lock 1 byte)
+		uintptr(unsafe.Pointer(&overlapped)),
+	)
+
+	if ret == 0 {
+		// ERROR_LOCK_VIOLATION = 33
+		if errno, ok := err.(syscall.Errno); ok && errno == 33 {
+			wl.f.Close()
+			return false, nil
+		}
+		wl.f.Close()
+		return false, err
+	}
+
+	return true, nil
+}
+
+func (wl *walLock) unlock() {
+	if wl.f != nil {
+		wl.f.Close() // Closing the file handle automatically releases the lock in Windows
+		os.Remove(wl.file)
+	}
+}

diskcache/lock.go

@@ -107,9 +107,10 @@ func (c *DiskCache) doOpen() error {
 	// disable open multiple times
 	if !c.noLock {
 		fl := newFlock(c.path)
-		if err := fl.lock(); err != nil {
+		if ok, err := fl.tryLock(); !ok {
 			return WrapLockError(err, c.path, 0).WithDetails("failed_to_acquire_directory_lock")
 		} else {
+			l.Infof("locked file %s ok", fl.file)
 			c.flock = fl
 		}
 	}
@@ -205,9 +206,7 @@ func (c *DiskCache) Close() error {
 
 	if !c.noLock {
 		if c.flock != nil {
-			if err := c.flock.unlock(); err != nil {
-				return WrapLockError(err, c.path, 0).WithDetails("failed_to_release_directory_lock")
-			}
+			c.flock.unlock()
 		}
 	}
 

diskcache/open.go

@@ -22,7 +22,11 @@ func TestOpen(t *T.T) {
 		// lock then no-lock
 		c, err := Open(WithPath(p))
 		assert.NoError(t, err)
+
+		assert.FileExists(t, filepath.Join(p, ".lock"))
+
 		assert.NoError(t, c.Close())
+		assert.NoFileExists(t, filepath.Join(p, ".lock"))
 
 		c2, err := Open(WithPath(p), WithNoPos(true))
 		assert.NoError(t, err)