Update lock strategy for faster writes

This commit changes the default lock strategy for general read
and write operations, such as SET and INTERSECTS, to speed up
write heavy workloads.

In some cases showing a 35% performance increase for SET
operations, while keeping read heavy workload performance the
same as before.

Author: tidwall

cmd/tile38-server/main.go

@@ -90,6 +90,7 @@ Advanced Options:
   --http-transport yes/no : HTTP transport (default: yes)
   --protected-mode yes/no : protected mode (default: yes)
   --nohup                 : do not exit on SIGHUP
+  --spinlock              : use a spinlock. For very write-heavy workloads
 
 Developer Options:
   --dev                             : enable developer mode
@@ -144,6 +145,7 @@ Developer Options:
 		nohup               bool
 		showEvioDisabled    bool
 		showThreadsDisabled bool
+		spinlock            bool
 	)
 
 	var (
@@ -199,6 +201,9 @@ Developer Options:
 		case "--nohup", "-nohup":
 			nohup = true
 			continue
+		case "--spinlock", "-spinlock":
+			spinlock = true
+			continue
 		case "--appendonly", "-appendonly":
 			i++
 			if i < len(os.Args) {
@@ -414,7 +419,7 @@ Developer Options:
 	}
 
 	c := make(chan os.Signal, 1)
-	shutdown := make (chan bool, 1)
+	shutdown := make(chan bool, 1)
 
 	signal.Notify(c, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
 	go func() {
@@ -484,6 +489,7 @@ Developer Options:
 		AppendFileName:    appendFileName,
 		QueueFileName:     queueFileName,
 		Shutdown:          shutdown,
+		Spinlock:          spinlock,
 	}
 	if err := server.Serve(opts); err != nil {
 		log.Fatal(err)

internal/server/expire.go

@@ -16,7 +16,7 @@ const bgExpireDelay = time.Second / 10
 func (s *Server) backgroundExpiring(wg *sync.WaitGroup) {
 	defer wg.Done()
 	s.loopUntilServerStops(bgExpireDelay, func() {
-		s.mu.Lock()
+		s.mu.LockLowPriority()
 		defer s.mu.Unlock()
 		now := time.Now()
 		s.backgroundExpireObjects(now)

internal/server/server.go

@@ -69,6 +69,88 @@ type commandDetails struct {
 	children  []*commandDetails // for multi actions such as "PDEL"
 }
 
+type rwlocker interface {
+	LockLowPriority()
+	Lock()
+	Unlock()
+	RLock()
+	RUnlock()
+}
+
+// rwspinlock is the same as a RWLock, but uses spinlocks instead of mutexes.
+type rwspinlock struct {
+	state atomic.Int32
+}
+
+func (l *rwspinlock) Lock() {
+	for {
+		state := l.state.Load()
+		if state == 0 && l.state.CompareAndSwap(state, -1) {
+			return
+		}
+		runtime.Gosched()
+	}
+}
+func (l *rwspinlock) LockLowPriority() {
+	// All write locks are low priority and unfair. First come first serve.
+	l.Lock()
+}
+func (l *rwspinlock) Unlock() {
+	if l.state.Add(1) > 0 {
+		panic("Unlock of unlocked rwspinlock")
+	}
+}
+func (l *rwspinlock) RLock() {
+	for {
+		state := l.state.Load()
+		if state >= 0 && l.state.CompareAndSwap(state, state+1) {
+			return
+		}
+		runtime.Gosched()
+	}
+}
+func (l *rwspinlock) RUnlock() {
+	if l.state.Add(-1) < 0 {
+		panic("RUnlock of unlocked rwspinlock")
+	}
+}
+
+// rwmutex is the same as a RWLock but includes a LockLowPriority for
+// performing write locks that do not block the queue for readers.
+type rwmutex struct {
+	mu sync.RWMutex
+}
+
+func (l *rwmutex) Lock() {
+	start := time.Now()
+	for {
+		for range 100 {
+			if l.mu.TryLock() {
+				return
+			}
+			runtime.Gosched()
+		}
+		if time.Since(start) > time.Millisecond*50 {
+			break
+		}
+	}
+	l.mu.Lock()
+}
+func (l *rwmutex) LockLowPriority() {
+	for !l.mu.TryLock() {
+		runtime.Gosched()
+	}
+}
+func (l *rwmutex) Unlock() {
+	l.mu.Unlock()
+}
+func (l *rwmutex) RLock() {
+	l.mu.RLock()
+}
+func (l *rwmutex) RUnlock() {
+	l.mu.RUnlock()
+}
+
 // Server is a tile38 controller
 type Server struct {
 	// user defined options
@@ -107,7 +189,7 @@ type Server struct {
 	connsmu sync.RWMutex
 	conns   map[int]*Client
 
-	mu sync.RWMutex
+	mu rwlocker // sync.RWMutex
 
 	// aof
 	aof       *os.File    // active aof file
@@ -183,6 +265,9 @@ type Options struct {
 
 	// Shutdown allows for shutting down the server.
 	Shutdown <-chan bool
+
+	// Spinlock uses a spinlock instead of a mutex
+	Spinlock bool
 }
 
 // Serve starts a new tile38 server
@@ -208,8 +293,17 @@ func Serve(opts Options) error {
 		}
 	}()
 
+	var lock rwlocker
+
+	if opts.Spinlock {
+		lock = new(rwspinlock)
+	} else {
+		lock = new(rwmutex)
+	}
+
 	// Initialize the s
 	s := &Server{
+		mu:        lock,
 		unix:      opts.UnixSocketPath,
 		host:      opts.Host,
 		port:      opts.Port,
@@ -804,7 +898,7 @@ func (s *Server) watchLuaStatePool(wg *sync.WaitGroup) {
 func (s *Server) backgroundSyncAOF(wg *sync.WaitGroup) {
 	defer wg.Done()
 	s.loopUntilServerStops(time.Second, func() {
-		s.mu.Lock()
+		s.mu.LockLowPriority()
 		defer s.mu.Unlock()
 		s.flushAOF(true)
 	})

scripts/build.sh

@@ -37,5 +37,9 @@ export CGO_ENABLED=0
 # 	go mod vendor
 # fi
 
+if [[ "$GORACE" == "1" ]]; then
+goflags="$goflags -race"
+fi
+
 # Build and store objects into original directory.
-go build -ldflags "$LDFLAGS" -o $1 cmd/$1/*.go
+go build -ldflags "$LDFLAGS" $goflags -o $1 cmd/$1/*.go