1
0
mirror of https://github.com/astaxie/beego.git synced 2024-12-24 16:00:49 +00:00
Beego/vendor/github.com/siddontang/ledisdb/rpl/rpl.go
2018-07-30 12:05:51 +08:00

337 lines
5.6 KiB
Go

package rpl
import (
"encoding/binary"
"os"
"path"
"sync"
"time"
"github.com/siddontang/go/log"
"github.com/siddontang/go/snappy"
"github.com/siddontang/ledisdb/config"
)
type Stat struct {
FirstID uint64
LastID uint64
CommitID uint64
}
type Replication struct {
m sync.Mutex
cfg *config.Config
s LogStore
commitID uint64
commitLog *os.File
quit chan struct{}
wg sync.WaitGroup
nc chan struct{}
ncm sync.Mutex
}
func NewReplication(cfg *config.Config) (*Replication, error) {
if len(cfg.Replication.Path) == 0 {
cfg.Replication.Path = path.Join(cfg.DataDir, "rpl")
}
base := cfg.Replication.Path
r := new(Replication)
r.quit = make(chan struct{})
r.nc = make(chan struct{})
r.cfg = cfg
var err error
switch cfg.Replication.StoreName {
case "goleveldb":
if r.s, err = NewGoLevelDBStore(path.Join(base, "wal"), cfg.Replication.SyncLog); err != nil {
return nil, err
}
default:
if r.s, err = NewFileStore(path.Join(base, "ldb"), cfg); err != nil {
return nil, err
}
}
if r.commitLog, err = os.OpenFile(path.Join(base, "commit.log"), os.O_RDWR|os.O_CREATE, 0644); err != nil {
return nil, err
}
if s, _ := r.commitLog.Stat(); s.Size() == 0 {
r.commitID = 0
} else if err = binary.Read(r.commitLog, binary.BigEndian, &r.commitID); err != nil {
return nil, err
}
log.Infof("staring replication with commit ID %d", r.commitID)
r.wg.Add(1)
go r.run()
return r, nil
}
func (r *Replication) Close() error {
close(r.quit)
r.wg.Wait()
r.m.Lock()
defer r.m.Unlock()
log.Infof("closing replication with commit ID %d", r.commitID)
if r.s != nil {
r.s.Close()
r.s = nil
}
if err := r.updateCommitID(r.commitID, true); err != nil {
log.Errorf("update commit id err %s", err.Error())
}
if r.commitLog != nil {
r.commitLog.Close()
r.commitLog = nil
}
return nil
}
func (r *Replication) Log(data []byte) (*Log, error) {
if r.cfg.Replication.Compression {
//todo optimize
var err error
if data, err = snappy.Encode(nil, data); err != nil {
return nil, err
}
}
r.m.Lock()
lastID, err := r.s.LastID()
if err != nil {
r.m.Unlock()
return nil, err
}
commitId := r.commitID
if lastID < commitId {
lastID = commitId
} else if lastID > commitId {
r.m.Unlock()
return nil, ErrCommitIDBehind
}
l := new(Log)
l.ID = lastID + 1
l.CreateTime = uint32(time.Now().Unix())
if r.cfg.Replication.Compression {
l.Compression = 1
} else {
l.Compression = 0
}
l.Data = data
if err = r.s.StoreLog(l); err != nil {
r.m.Unlock()
return nil, err
}
r.m.Unlock()
r.ncm.Lock()
close(r.nc)
r.nc = make(chan struct{})
r.ncm.Unlock()
return l, nil
}
func (r *Replication) WaitLog() <-chan struct{} {
r.ncm.Lock()
ch := r.nc
r.ncm.Unlock()
return ch
}
func (r *Replication) StoreLog(log *Log) error {
r.m.Lock()
err := r.s.StoreLog(log)
r.m.Unlock()
return err
}
func (r *Replication) FirstLogID() (uint64, error) {
r.m.Lock()
id, err := r.s.FirstID()
r.m.Unlock()
return id, err
}
func (r *Replication) LastLogID() (uint64, error) {
r.m.Lock()
id, err := r.s.LastID()
r.m.Unlock()
return id, err
}
func (r *Replication) LastCommitID() (uint64, error) {
r.m.Lock()
id := r.commitID
r.m.Unlock()
return id, nil
}
func (r *Replication) UpdateCommitID(id uint64) error {
r.m.Lock()
err := r.updateCommitID(id, r.cfg.Replication.SyncLog == 2)
r.m.Unlock()
return err
}
func (r *Replication) Stat() (*Stat, error) {
r.m.Lock()
defer r.m.Unlock()
s := &Stat{}
var err error
if s.FirstID, err = r.s.FirstID(); err != nil {
return nil, err
}
if s.LastID, err = r.s.LastID(); err != nil {
return nil, err
}
s.CommitID = r.commitID
return s, nil
}
func (r *Replication) updateCommitID(id uint64, force bool) error {
if force {
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
return nil
}
func (r *Replication) CommitIDBehind() (bool, error) {
r.m.Lock()
id, err := r.s.LastID()
if err != nil {
r.m.Unlock()
return false, err
}
behind := id > r.commitID
r.m.Unlock()
return behind, nil
}
func (r *Replication) GetLog(id uint64, log *Log) error {
return r.s.GetLog(id, log)
}
func (r *Replication) NextNeedCommitLog(log *Log) error {
r.m.Lock()
defer r.m.Unlock()
id, err := r.s.LastID()
if err != nil {
return err
}
if id <= r.commitID {
return ErrNoBehindLog
}
return r.s.GetLog(r.commitID+1, log)
}
func (r *Replication) Clear() error {
return r.ClearWithCommitID(0)
}
func (r *Replication) ClearWithCommitID(id uint64) error {
r.m.Lock()
defer r.m.Unlock()
if err := r.s.Clear(); err != nil {
return err
}
return r.updateCommitID(id, true)
}
func (r *Replication) run() {
defer r.wg.Done()
syncTc := time.NewTicker(1 * time.Second)
purgeTc := time.NewTicker(1 * time.Hour)
for {
select {
case <-purgeTc.C:
n := (r.cfg.Replication.ExpiredLogDays * 24 * 3600)
r.m.Lock()
err := r.s.PurgeExpired(int64(n))
r.m.Unlock()
if err != nil {
log.Errorf("purge expired log error %s", err.Error())
}
case <-syncTc.C:
if r.cfg.Replication.SyncLog == 1 {
r.m.Lock()
err := r.s.Sync()
r.m.Unlock()
if err != nil {
log.Errorf("sync store error %s", err.Error())
}
}
if r.cfg.Replication.SyncLog != 2 {
//we will sync commit id every 1 second
r.m.Lock()
err := r.updateCommitID(r.commitID, true)
r.m.Unlock()
if err != nil {
log.Errorf("sync commitid error %s", err.Error())
}
}
case <-r.quit:
syncTc.Stop()
purgeTc.Stop()
return
}
}
}