From dbb6d53f9d75868329f40d1ff545e26605da6a94 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Tue, 13 Dec 2022 16:07:41 -0600 Subject: [PATCH] implement closing on dax, remove all locks when shutting down --- api_directive.go | 38 +++++++++++--- dax/server/server.go | 3 +- dax/service_manager.go | 12 +++++ dax/storage/storage.go | 92 +++++++++++++++++++++++++++++++--- dax/writelogger/writelogger.go | 8 +-- server.go | 6 ++- 6 files changed, 139 insertions(+), 20 deletions(-) diff --git a/api_directive.go b/api_directive.go index fad852fb8..c61e9d9fc 100644 --- a/api_directive.go +++ b/api_directive.go @@ -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{ diff --git a/dax/server/server.go b/dax/server/server.go index 5359ac7bf..63e006043 100644 --- a/dax/server/server.go +++ b/dax/server/server.go @@ -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) diff --git a/dax/service_manager.go b/dax/service_manager.go index b19ba695a..10c2d27f5 100644 --- a/dax/service_manager.go +++ b/dax/service_manager.go @@ -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 { diff --git a/dax/storage/storage.go b/dax/storage/storage.go index b2e1df826..5ee55ade0 100644 --- a/dax/storage/storage.go +++ b/dax/storage/storage.go @@ -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") } diff --git a/dax/writelogger/writelogger.go b/dax/writelogger/writelogger.go index 6b213ab08..2c63c6a78 100644 --- a/dax/writelogger/writelogger.go +++ b/dax/writelogger/writelogger.go @@ -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 diff --git a/server.go b/server.go index 588a870e8..30e5473ba 100644 --- a/server.go +++ b/server.go @@ -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") } }