package rpl import ( "fmt" "io/ioutil" "os" "sort" "sync" "time" "github.com/ledisdb/ledisdb/config" "github.com/siddontang/go/log" "github.com/siddontang/go/num" ) const ( defaultMaxLogFileSize = int64(256 * 1024 * 1024) maxLogFileSize = int64(1024 * 1024 * 1024) defaultLogNumInFile = int64(1024 * 1024) ) /* File Store: 00000001.data 00000001.meta 00000002.data 00000002.meta data: log1 data | log2 data | magic data if data has no magic data, it means that we don't close replication gracefully. so we must repair the log data log data: id (bigendian uint64), create time (bigendian uint32), compression (byte), data len(bigendian uint32), data split data = log0 data + [padding 0] -> file % pagesize() == 0 meta: log1 offset | log2 offset log offset: bigendian uint32 | bigendian uint32 //sha1 of github.com/ledisdb/ledisdb 20 bytes magic data = "\x1c\x1d\xb8\x88\xff\x9e\x45\x55\x40\xf0\x4c\xda\xe0\xce\x47\xde\x65\x48\x71\x17" we must guarantee that the log id is monotonic increment strictly. if log1's id is 1, log2 must be 2 */ type FileStore struct { LogStore cfg *config.Config base string rm sync.RWMutex wm sync.Mutex rs tableReaders w *tableWriter quit chan struct{} } func NewFileStore(base string, cfg *config.Config) (*FileStore, error) { s := new(FileStore) s.quit = make(chan struct{}) var err error if err = os.MkdirAll(base, 0755); err != nil { return nil, err } s.base = base if cfg.Replication.MaxLogFileSize == 0 { cfg.Replication.MaxLogFileSize = defaultMaxLogFileSize } cfg.Replication.MaxLogFileSize = num.MinInt64(cfg.Replication.MaxLogFileSize, maxLogFileSize) s.cfg = cfg if err = s.load(); err != nil { return nil, err } index := int64(1) if len(s.rs) != 0 { index = s.rs[len(s.rs)-1].index + 1 } s.w = newTableWriter(s.base, index, cfg.Replication.MaxLogFileSize, cfg.Replication.UseMmap) s.w.SetSyncType(cfg.Replication.SyncLog) go s.checkTableReaders() return s, nil } func (s *FileStore) GetLog(id uint64, l *Log) error { //first search in table writer if err := s.w.GetLog(id, l); err == nil { return nil } else if err != ErrLogNotFound { return err } s.rm.RLock() t := s.rs.Search(id) if t == nil { s.rm.RUnlock() return ErrLogNotFound } err := t.GetLog(id, l) s.rm.RUnlock() return err } func (s *FileStore) FirstID() (uint64, error) { id := uint64(0) s.rm.RLock() if len(s.rs) > 0 { id = s.rs[0].first } else { id = 0 } s.rm.RUnlock() if id > 0 { return id, nil } //if id = 0, return s.w.First(), nil } func (s *FileStore) LastID() (uint64, error) { id := s.w.Last() if id > 0 { return id, nil } //if table writer has no last id, we may find in the last table reader s.rm.RLock() if len(s.rs) > 0 { id = s.rs[len(s.rs)-1].last } s.rm.RUnlock() return id, nil } func (s *FileStore) StoreLog(l *Log) error { s.wm.Lock() err := s.storeLog(l) s.wm.Unlock() return err } func (s *FileStore) storeLog(l *Log) error { err := s.w.StoreLog(l) if err == nil { return nil } else if err != errTableNeedFlush { return err } var r *tableReader r, err = s.w.Flush() if err != nil { log.Fatalf("write table flush error %s, can not store!!!", err.Error()) s.w.Close() return err } s.rm.Lock() s.rs = append(s.rs, r) s.rm.Unlock() err = s.w.StoreLog(l) return err } func (s *FileStore) PurgeExpired(n int64) error { s.rm.Lock() var purges []*tableReader t := uint32(time.Now().Unix() - int64(n)) for i, r := range s.rs { if r.lastTime > t { purges = append([]*tableReader{}, s.rs[0:i]...) n := copy(s.rs, s.rs[i:]) s.rs = s.rs[0:n] break } } s.rm.Unlock() s.purgeTableReaders(purges) return nil } func (s *FileStore) Sync() error { return s.w.Sync() } func (s *FileStore) Clear() error { s.wm.Lock() s.rm.Lock() defer func() { s.rm.Unlock() s.wm.Unlock() }() s.w.Close() for i := range s.rs { s.rs[i].Close() } s.rs = tableReaders{} if err := os.RemoveAll(s.base); err != nil { return err } if err := os.MkdirAll(s.base, 0755); err != nil { return err } s.w = newTableWriter(s.base, 1, s.cfg.Replication.MaxLogFileSize, s.cfg.Replication.UseMmap) return nil } func (s *FileStore) Close() error { close(s.quit) s.wm.Lock() s.rm.Lock() if r, err := s.w.Flush(); err != nil { if err != errNilHandler { log.Errorf("close err: %s", err.Error()) } } else { r.Close() s.w.Close() } for i := range s.rs { s.rs[i].Close() } s.rs = tableReaders{} s.rm.Unlock() s.wm.Unlock() return nil } func (s *FileStore) checkTableReaders() { t := time.NewTicker(60 * time.Second) defer t.Stop() for { select { case <-t.C: s.rm.Lock() for _, r := range s.rs { if !r.Keepalived() { r.Close() } } purges := []*tableReader{} maxNum := s.cfg.Replication.MaxLogFileNum num := len(s.rs) if num > maxNum { purges = s.rs[:num-maxNum] s.rs = s.rs[num-maxNum:] } s.rm.Unlock() s.purgeTableReaders(purges) case <-s.quit: return } } } func (s *FileStore) purgeTableReaders(purges []*tableReader) { for _, r := range purges { dataName := fmtTableDataName(r.base, r.index) metaName := fmtTableMetaName(r.base, r.index) r.Close() if err := os.Remove(dataName); err != nil { log.Errorf("purge table data %s err: %s", dataName, err.Error()) } if err := os.Remove(metaName); err != nil { log.Errorf("purge table meta %s err: %s", metaName, err.Error()) } } } func (s *FileStore) load() error { fs, err := ioutil.ReadDir(s.base) if err != nil { return err } s.rs = make(tableReaders, 0, len(fs)) var r *tableReader var index int64 for _, f := range fs { if _, err := fmt.Sscanf(f.Name(), "%08d.data", &index); err == nil { if r, err = newTableReader(s.base, index, s.cfg.Replication.UseMmap); err != nil { log.Errorf("load table %s err: %s", f.Name(), err.Error()) } else { s.rs = append(s.rs, r) } } } if err := s.rs.check(); err != nil { return err } return nil } type tableReaders []*tableReader func (ts tableReaders) Len() int { return len(ts) } func (ts tableReaders) Swap(i, j int) { ts[i], ts[j] = ts[j], ts[i] } func (ts tableReaders) Less(i, j int) bool { return ts[i].first < ts[j].first } func (ts tableReaders) Search(id uint64) *tableReader { i, j := 0, len(ts)-1 for i <= j { h := i + (j-i)/2 if ts[h].first <= id && id <= ts[h].last { return ts[h] } else if ts[h].last < id { i = h + 1 } else { j = h - 1 } } return nil } func (ts tableReaders) check() error { if len(ts) == 0 { return nil } sort.Sort(ts) first := ts[0].first last := ts[0].last index := ts[0].index if first == 0 || first > last { return fmt.Errorf("invalid log in table %s", ts[0]) } for i := 1; i < len(ts); i++ { if ts[i].first <= last { return fmt.Errorf("invalid first log id %d in table %s", ts[i].first, ts[i]) } if ts[i].index <= index { return fmt.Errorf("invalid index %d in table %s", ts[i].index, ts[i]) } first = ts[i].first last = ts[i].last index = ts[i].index } return nil }