scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 17:15:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

implement closing on dax, remove all locks when shutting down

Browse source
This commit is contained in:
Matthew Jaffee 2022-12-13 16:07:41 -06:00 • committed by Matthew Jaffee
parent bbaa7dd0f1
commit dbb6d53f9d
6 changed files with 139 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
api_directive.go
View file

@ -42,6 +42,9 @@ func (api *API) ApplyDirective(ctx context.Context, d *dax.Directive) error {
if err := api.deleteAllIndexes(ctx); err != nil {
return errors.Wrap(err, "deleting all indexes")
}
if err := api.serverlessStorage.RemoveAll(); err != nil {
return errors.Wrap(err, "removing all managers")
}
// Set previousDirective to empty so the diff handles everything as new.
previousDirective = dax.Directive{}
@ -359,9 +362,12 @@ func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey
if err != nil {
return errors.Wrap(err, "loading table key snapshot")
}
defer rc.Close()
if err := api.TranslateIndexDB(ctx, string(tkey), int(partition), rc); err != nil {
return errors.Wrap(err, "restoring table keys")
if rc != nil {
defer rc.Close()
if err := api.TranslateIndexDB(ctx, string(tkey), int(partition), rc); err != nil {
return errors.Wrap(err, "restoring table keys")
}
}
// define write log loading in a function since we have to do it
@ -371,6 +377,9 @@ func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey
if err != nil {
return errors.Wrap(err, "getting write log reader for table keys")
}
if writelog == nil {
return nil
}
reader := storage.NewTableKeyReader(qtid, partition, writelog)
defer reader.Close()
store := idx.TranslateStore(int(partition))
@ -426,9 +435,11 @@ func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax.
if err != nil {
return errors.Wrap(err, "loading field key snapshot")
}
defer rc.Close()
if err := api.TranslateFieldDB(ctx, string(tkey), string(field), rc); err != nil {
return errors.Wrap(err, "restoring field keys")
if rc != nil {
defer rc.Close()
if err := api.TranslateFieldDB(ctx, string(tkey), string(field), rc); err != nil {
return errors.Wrap(err, "restoring field keys")
}
}
// define write log loading in a function since we have to do it
@ -438,6 +449,9 @@ func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax.
if err != nil {
return errors.Wrap(err, "getting write log reader for field keys")
}
if writelog == nil {
return nil
}
reader := storage.NewFieldKeyReader(qtid, field, writelog)
defer reader.Close()
// Get field in order to find the translate store.
@ -503,8 +517,11 @@ func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.Shar
if err != nil {
return errors.Wrap(err, "reading latest snapshot for shard")
}
if err := api.RestoreShard(ctx, string(tkey), uint64(shard), rc); err != nil {
return errors.Wrap(err, "restoring shard data")
if rc != nil {
defer rc.Close()
if err := api.RestoreShard(ctx, string(tkey), uint64(shard), rc); err != nil {
return errors.Wrap(err, "restoring shard data")
}
}
// define write log loading in a func because we do it twice.
@ -513,12 +530,17 @@ func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.Shar
if err != nil {
return errors.Wrap(err, "")
}
if writelog == nil {
return nil
}
reader := storage.NewShardReader(qtid, partition, shard, writelog)
defer reader.Close()
for logMsg, err := reader.Read(); err != io.EOF; logMsg, err = reader.Read() {
if err != nil {
return errors.Wrap(err, "reading from log reader")
}
switch msg := logMsg.(type) {
case *computer.ImportRoaringMessage:
req := &ImportRoaringRequest{

3
dax/server/server.go
View file

@ -163,8 +163,7 @@ func (m *Command) Close() error {
return nil
default:
eg := errgroup.Group{}
//eg.Go(m.Server.Close)
eg.Go(m.svcmgr.StopAll)
err := eg.Wait()
//_ = testhook.Closed(pilosa.NewAuditor(), m, nil)
close(m.done)

12
dax/service_manager.go
View file

@ -82,6 +82,18 @@ func (s *ServiceManager) StartAll() error {
return nil
}
func (s *ServiceManager) StopAll() error {
for key := range s.computers {
if err := s.ComputerStop(key); err != nil {
s.Logger.Printf("stopping computer %s: %v", key, err)
}
}
if err := s.QueryerStop(); err != nil {
s.Logger.Printf("stopping queryer: %v", err)
}
return s.MDSStop()
}
// MDSStart starts the MDS service.
func (s *ServiceManager) MDSStart() error {
if s.MDS == nil {

92
dax/storage/storage.go
View file

@ -1,10 +1,10 @@
package storage
import (
"bytes"
"fmt"
"io"
"path"
"strings"
"sync"
"github.com/molecula/featurebase/v3/dax"
@ -75,6 +75,19 @@ func (mm *ManagerManager) GetShardManager(qtid dax.QualifiedTableID, partition d
return mm.shardManagers[key]
}
func (mm *ManagerManager) RemoveShardManager(qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum) {
mm.mu.Lock()
defer mm.mu.Unlock()
key := shardK{qtid: qtid, partition: partition, shard: shard}
if m, ok := mm.shardManagers[key]; ok {
err := m.Unlock()
if err != nil {
mm.Logger.Printf("unlocking shard manager during removal: %v", err)
}
delete(mm.shardManagers, key)
}
}
func (mm *ManagerManager) GetTableKeyManager(qtid dax.QualifiedTableID, partition dax.PartitionNum) *Manager {
mm.mu.Lock()
defer mm.mu.Unlock()
@ -92,6 +105,19 @@ func (mm *ManagerManager) GetTableKeyManager(qtid dax.QualifiedTableID, partitio
return mm.tableKeyManagers[key]
}
func (mm *ManagerManager) RemoveTableKeyManager(qtid dax.QualifiedTableID, partition dax.PartitionNum) {
mm.mu.Lock()
defer mm.mu.Unlock()
key := tableKeyK{qtid: qtid, partition: partition}
if m, ok := mm.tableKeyManagers[key]; ok {
err := m.Unlock()
if err != nil {
mm.Logger.Printf("unlocking table key manager during removal: %v", err)
}
delete(mm.tableKeyManagers, key)
}
}
func (mm *ManagerManager) GetFieldKeyManager(qtid dax.QualifiedTableID, field dax.FieldName) *Manager {
mm.mu.Lock()
defer mm.mu.Unlock()
@ -109,6 +135,53 @@ func (mm *ManagerManager) GetFieldKeyManager(qtid dax.QualifiedTableID, field da
return mm.fieldKeyManagers[key]
}
func (mm *ManagerManager) RemoveFieldKeyManager(qtid dax.QualifiedTableID, field dax.FieldName) {
mm.mu.Lock()
defer mm.mu.Unlock()
key := fieldKeyK{qtid: qtid, field: field}
if m, ok := mm.fieldKeyManagers[key]; ok {
err := m.Unlock()
if err != nil {
mm.Logger.Printf("unlocking field key manager during removal: %v", err)
}
delete(mm.fieldKeyManagers, key)
}
}
// RemoveAll unlocks and deletes all Managers held within this
// ManagerManager.
func (mm *ManagerManager) RemoveAll() error {
mm.mu.Lock()
defer mm.mu.Unlock()
errList := make([]error, 0)
for k, mgr := range mm.shardManagers {
err := mgr.Unlock()
if err != nil && !strings.Contains(err.Error(), "resource was not locked") {
errList = append(errList, err)
}
delete(mm.shardManagers, k)
}
for k, mgr := range mm.tableKeyManagers {
err := mgr.Unlock()
if err != nil && !strings.Contains(err.Error(), "resource was not locked") {
errList = append(errList, err)
}
delete(mm.tableKeyManagers, k)
}
for k, mgr := range mm.fieldKeyManagers {
err := mgr.Unlock()
if err != nil && !strings.Contains(err.Error(), "resource was not locked") {
errList = append(errList, err)
}
delete(mm.fieldKeyManagers, k)
}
if len(errList) > 0 {
return errors.Errorf("%v", errList)
}
return nil
}
// Manager wraps the snapshotter and writelogger to implement the
// ServerlessStorage interface and maintain messy state between
// calls. Manager is *not* threadsafe, care should be taken that
@ -136,6 +209,9 @@ func (m *Manager) initialize() *Manager {
return m
}
// LoadLatestSnapshot finds the most recent snapshot for this resource
// and returns a ReadCloser for that snapshot data. If there is no
// snapshot for this resource it returns nil, nil.
func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) {
snaps, err := m.snapshotter.List(m.bucket, m.key)
if err != nil {
@ -145,7 +221,7 @@ func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) {
if len(snaps) == 0 {
m.loadWLsPastVersion = -1
return io.NopCloser(bytes.NewReader([]byte{})), nil
return nil, nil
}
// assuming snapshots come back in sorted order
latest := snaps[len(snaps)-1]
@ -165,10 +241,11 @@ func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) {
// ListSnapshots() []SnapInfo
// LoadSnapshot(version int) (data io.ReadCloser, err error)
// LoadWriteLog can be called after LoadLatestSnapshot. It loads
// any writelog data which has been written since the latest
// LoadWriteLog can be called after LoadLatestSnapshot. It loads any
// writelog data which has been written since the latest
// snapshot. Subsequent calls to LoadWriteLog will only return new
// data that hasn't previously been returned from LoadWriteLog.
// data that hasn't previously been returned from LoadWriteLog. If
// there is no writelog, it returns nil, nil.
func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) {
if m.loadWLsPastVersion == -2 {
return nil, errors.New(errors.ErrUncoded, "LoadWriteLog called in inconsistent state, can't tell what version to load from")
@ -193,7 +270,7 @@ func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) {
if len(versions) == 0 {
m.log.Debugf("LoadWriteLog: no logs after snapshot: %d on %s", m.loadWLsPastVersion, path.Join(m.bucket, m.key))
m.latestWLVersion = m.loadWLsPastVersion + 1
return io.NopCloser(bytes.NewReader([]byte{})), nil
return nil, nil
}
if m.locked && m.latestWLVersion != versions[0] {
@ -296,6 +373,9 @@ func (m *Manager) SnapshotTo(wt io.WriterTo) error {
// should have those removed by the operating system when the
// process exits anyway.
func (m *Manager) Unlock() error {
if !m.locked {
return errors.New(errors.ErrUncoded, "resource was not locked")
}
if err := m.writeLogger.Unlock(m.bucket, m.key); err != nil {
return errors.Wrap(err, "unlocking")
}

8
dax/writelogger/writelogger.go
View file

@ -133,13 +133,15 @@ func (w *WriteLogger) Lock(bucket, key string) error {
if err := os.MkdirAll(lockDir, 0777); err != nil {
return errors.Wrapf(err, "lock dir %s", lockDir)
}
f, err := os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL, 0644)
f, err := os.OpenFile(lockFile, os.O_CREATE|os.O_EXCL|syscall.O_NONBLOCK, 0644)
if err != nil {
return errors.Wrapf(err, "opening lock file: %s", lockFile)
}
w.mu.Lock()
defer w.mu.Unlock()
w.lockFiles[lockFile] = f
// fd, err = syscall.Open(lockFile, syscall.O_RDWR|syscall.O_CREAT, 0644)
// if err != nil {
// return 0, errors.Wrapf(err, "syscall opening %s", lockFile)
@ -147,7 +149,7 @@ func (w *WriteLogger) Lock(bucket, key string) error {
// err = syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
// Type: syscall.F_WRLCK,
// })
return errors.Wrap(err, "locking")
return nil
}
@ -177,7 +179,7 @@ func (w *WriteLogger) Unlock(bucket, key string) error {
// err := syscall.FcntlFlock(uintptr(fd), syscall.F_SETLK, &syscall.Flock_t{
// Type: syscall.F_UNLCK,
// })
return errors.Wrap(err, "closing lock file")
return errors.Wrap(err, "removing lock file")
}
// paths takes a key and returns the full file path (including the root data

6
server.go
View file

@ -816,6 +816,7 @@ func (s *Server) Close() error {
if s.holder != nil {
errh = s.holder.Close()
}
errSS := s.serverlessStorage.RemoveAll()
// prefer to return holder error over cluster
// error. This order is somewhat arbitrary. It would be better if we had
@ -833,7 +834,10 @@ func (s *Server) Close() error {
if errd != nil {
return errors.Wrap(errd, "closing disco")
}
return errors.Wrap(errE, "closing executor")
if errE != nil {
return errors.Wrap(errE, "closing executor")
}
return errors.Wrap(errSS, "unlocking all serverless storage")
}
}