Skip to content

Commit

Permalink
Speed up leader/follower replication
Browse files Browse the repository at this point in the history
This commit lowers the response time for a leader to send
updates to a follower. Should now be nearly instant.
  • Loading branch information
tidwall committed Dec 14, 2022
1 parent e60ea70 commit a8c92a0
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 9 deletions.
12 changes: 6 additions & 6 deletions internal/server/aof.go
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,8 @@ func (s *Server) flushAOF(sync bool) {
if err != nil {
panic(err)
}
// send a broadcast to all sleeping followers
s.fcond.Broadcast()
if sync {
if err := s.aof.Sync(); err != nil {
panic(err)
Expand Down Expand Up @@ -165,11 +167,6 @@ func (s *Server) writeAOF(args []string, d *commandDetails) error {
s.aofsz += len(s.aofbuf) - n
}

// notify aof live connections that we have new data
s.fcond.L.Lock()
s.fcond.Broadcast()
s.fcond.L.Unlock()

// process geofences
if d != nil {
// webhook geofences
Expand Down Expand Up @@ -516,6 +513,7 @@ func (s *Server) liveAOF(pos int64, conn net.Conn, rd *PipelineReader, msg *Mess
if err != nil {
return err
}

b := make([]byte, 4096*2)
for {
n, err := f.Read(b)
Expand All @@ -525,7 +523,9 @@ func (s *Server) liveAOF(pos int64, conn net.Conn, rd *PipelineReader, msg *Mess
}
}
if err == io.EOF {
time.Sleep(time.Second / 4)
s.fcond.L.Lock()
s.fcond.Wait()
s.fcond.L.Unlock()
} else if err != nil {
return err
}
Expand Down
15 changes: 12 additions & 3 deletions internal/server/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -135,9 +135,9 @@ type Server struct {
fcond *sync.Cond
lstack []*commandDetails
lives map[*liveBuffer]bool
lcond *sync.Cond
fcup bool // follow caught up
fcuponce bool // follow caught up once
lcond *sync.Cond // live geofence signal
fcup bool // follow caught up
fcuponce bool // follow caught up once
aofconnM map[net.Conn]io.Closer

// lua scripts
Expand Down Expand Up @@ -305,10 +305,19 @@ func Serve(opts Options) error {
nerr <- s.netServe()
}()

var fstop atomic.Bool
go func() {
for !fstop.Load() {
s.fcond.Broadcast()
time.Sleep(time.Second / 4)
}
}()

go func() {
<-opts.Shutdown
s.stopServer.Store(true)
log.Warnf("Shutting down...")
fstop.Store(true)
s.lnmu.Lock()
ln := s.ln
s.ln = nil
Expand Down

0 comments on commit a8c92a0

Please sign in to comment.