re
/
ledisdb
forked from mirror/ledisdb
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

control commit id flush

This commit is contained in:
siddontang 2014-11-10 14:13:29 +08:00
parent a0abd79328
commit 9b4c23353e
1 changed files with 21 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
rpl/rpl.go
View File

@ -24,8 +24,9 @@ type Replication struct {
s LogStore s LogStore
commitID uint64 commitID uint64
commitLog *os.File commitLog *os.File
commitLastTime time.Time
quit chan struct{} quit chan struct{}
@ -92,6 +93,10 @@ func (r *Replication) Close() error {
r.s = nil r.s = nil
} }
if err := r.updateCommitID(r.commitID, true); err != nil {
log.Error("update commit id err %s", err.Error())
}
if r.commitLog != nil { if r.commitLog != nil {
r.commitLog.Close() r.commitLog.Close()
r.commitLog = nil r.commitLog = nil
@ -185,7 +190,7 @@ func (r *Replication) UpdateCommitID(id uint64) error {
r.m.Lock() r.m.Lock()
defer r.m.Unlock() defer r.m.Unlock()
return r.updateCommitID(id) return r.updateCommitID(id, r.cfg.Replication.SyncLog == 2)
} }
func (r *Replication) Stat() (*Stat, error) { func (r *Replication) Stat() (*Stat, error) {
@ -207,17 +212,23 @@ func (r *Replication) Stat() (*Stat, error) {
return s, nil return s, nil
} }
func (r *Replication) updateCommitID(id uint64) error { func (r *Replication) updateCommitID(id uint64, force bool) error {
if _, err := r.commitLog.Seek(0, os.SEEK_SET); err != nil { n := time.Now()
return err
}
if err := binary.Write(r.commitLog, binary.BigEndian, id); err != nil { if force || n.Sub(r.commitLastTime) > time.Second {
return err if _, err := r.commitLog.Seek(0, os.SEEK_SET); err != nil {
return err
}
if err := binary.Write(r.commitLog, binary.BigEndian, id); err != nil {
return err
}
} }
r.commitID = id r.commitID = id
r.commitLastTime = n
return nil return nil
} }
@ -266,7 +277,7 @@ func (r *Replication) ClearWithCommitID(id uint64) error {
return err return err
} }
return r.updateCommitID(id) return r.updateCommitID(id, true)
} }
func (r *Replication) onPurgeExpired() { func (r *Replication) onPurgeExpired() {