Background fsync and aofbuf flushing

internal/collection/item/item.go

@@ -72,11 +72,7 @@ func New(id string, obj geojson.Object) *Item {
 		item = (*Item)(unsafe.Pointer(oitem))
 	}
 	item.idLen = uint32(len(id))
-	if len(id) > 0 {
-		data := make([]byte, len(id))
-		copy(data, id)
-		item.data = unsafe.Pointer(&data[0])
-	}
+	item.data = unsafe.Pointer((*reflect.SliceHeader)(unsafe.Pointer(&id)).Data)
 	return item
 }
 

internal/server/aof.go

@@ -115,12 +115,17 @@ func commandErrIsFatal(err error) bool {
 	return true
 }
 
-func (server *Server) flushAOF() {
+func (server *Server) flushAOF(sync bool) {
 	if len(server.aofbuf) > 0 {
 		_, err := server.aof.Write(server.aofbuf)
 		if err != nil {
 			panic(err)
 		}
+		if sync {
+			if err := server.aof.Sync(); err != nil {
+				panic(err)
+			}
+		}
 		server.aofbuf = server.aofbuf[:0]
 	}
 }

internal/server/aofshrink.go

@@ -240,7 +240,7 @@ func (server *Server) aofshrink() {
 			defer server.mu.Unlock()
 
 			// flush the aof buffer
-			server.flushAOF()
+			server.flushAOF(false)
 
 			aofbuf = aofbuf[:0]
 			for _, values := range server.shrinklog {

internal/server/follow.go

@@ -165,7 +165,7 @@ func (c *Server) followHandleCommand(args []string, followc int, w io.Writer) (i
 		return c.aofsz, err
 	}
 	if len(c.aofbuf) > 10240 {
-		c.flushAOF()
+		c.flushAOF(false)
 	}
 	return c.aofsz, nil
 }
@@ -301,7 +301,7 @@ func (c *Server) followStep(host string, port int, followc int) error {
 			if aofsz >= int(aofSize) {
 				caughtUp = true
 				c.mu.Lock()
-				c.flushAOF()
+				c.flushAOF(false)
 				c.fcup = true
 				c.fcuponce = true
 				c.mu.Unlock()

internal/server/server.go

@@ -252,7 +252,7 @@ func Serve(host string, port int, dir string, http bool) error {
 			return err
 		}
 		defer func() {
-			server.flushAOF()
+			server.flushAOF(false)
 			server.aof.Sync()
 		}()
 	}
@@ -268,6 +268,7 @@ func Serve(host string, port int, dir string, http bool) error {
 	go server.watchLuaStatePool()
 	go server.watchAutoGC()
 	go server.backgroundExpiring()
+	go server.backgroundSyncAOF()
 	defer func() {
 		// Stop background routines
 		server.followc.add(1) // this will force any follow communication to die
@@ -489,7 +490,7 @@ func (server *Server) evioServe() error {
 		if atomic.LoadInt32(&server.aofdirty) != 0 {
 			server.mu.Lock()
 			defer server.mu.Unlock()
-			server.flushAOF()
+			server.flushAOF(false)
 			atomic.StoreInt32(&server.aofdirty, 1)
 		}
 	}
@@ -667,7 +668,7 @@ func (server *Server) netServe() error {
 							// prewrite
 							server.mu.Lock()
 							defer server.mu.Unlock()
-							server.flushAOF()
+							server.flushAOF(false)
 						}()
 						atomic.StoreInt32(&server.aofdirty, 0)
 					}
@@ -755,7 +756,6 @@ func (server *Server) watchOutOfMemory() {
 	defer t.Stop()
 	var mem runtime.MemStats
 	for range t.C {
-		func() {
 		if server.stopServer.on() {
 			return
 		}
@@ -771,7 +771,6 @@ func (server *Server) watchOutOfMemory() {
 		}
 		runtime.ReadMemStats(&mem)
 		server.outOfMemory.set(int(mem.HeapAlloc) > server.config.maxMemory())
-		}()
 	}
 }
 
@@ -779,9 +778,26 @@ func (server *Server) watchLuaStatePool() {
 	t := time.NewTicker(time.Second * 10)
 	defer t.Stop()
 	for range t.C {
-		func() {
+		if server.stopServer.on() {
+			return
+		}
 		server.luapool.Prune()
-		}()
+	}
+}
+
+func (server *Server) backgroundSyncAOF() {
+	t := time.NewTicker(time.Second)
+	defer t.Stop()
+	for range t.C {
+		if server.stopServer.on() {
+			return
+		}
+		server.mu.Lock()
+		if len(server.aofbuf) > 0 {
+			server.flushAOF(true)
+		}
+		server.aofbuf = nil
+		server.mu.Unlock()
 	}
 }