From 033be817991a25ea33df1f163d97d7a08db44aa8 Mon Sep 17 00:00:00 2001 From: Matthew Jaffee Date: Wed, 14 Dec 2022 12:30:50 -0600 Subject: [PATCH] rename stupid manager names ManagerManager -> ResourceManager Manager -> Resource --- api.go | 38 +++++------ api_directive.go | 36 +++++------ cluster.go | 10 +-- dax/storage/storage.go | 126 ++++++++++++++++++------------------ dax/storage/storage_test.go | 90 +++++++++++++------------- server.go | 4 +- server/server.go | 4 +- 7 files changed, 154 insertions(+), 154 deletions(-) diff --git a/api.go b/api.go index 75f749454..8cb8c0dbf 100644 --- a/api.go +++ b/api.go @@ -56,7 +56,7 @@ type API struct { Serializer Serializer - serverlessStorage *storage.ManagerManager + serverlessStorage *storage.ResourceManager directiveWorkerPoolSize int @@ -86,7 +86,7 @@ func OptAPIServer(s *Server) apiOption { } } -func OptAPIServerlessStorage(mm *storage.ManagerManager) apiOption { +func OptAPIServerlessStorage(mm *storage.ResourceManager) apiOption { return func(a *API) error { a.serverlessStorage = mm return nil @@ -705,8 +705,8 @@ func (api *API) ImportRoaring(ctx context.Context, indexName, fieldName string, return errors.Wrap(err, "marshalling log message") } - mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum) - err = mgr.Append(b) + resource := api.serverlessStorage.GetShardResource(qtid, partitionNum, shardNum) + err = resource.Append(b) if err != nil { return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something? } @@ -1518,8 +1518,8 @@ func (api *API) Import(ctx context.Context, qcx *Qcx, req *ImportRequest, opts . return errors.Wrap(err, "marshalling log message") } - mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum) - err = mgr.Append(b) + resource := api.serverlessStorage.GetShardResource(qtid, partitionNum, shardNum) + err = resource.Append(b) if err != nil { return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something? } @@ -1768,8 +1768,8 @@ func (api *API) ImportRoaringShard(ctx context.Context, indexName string, shard return err1 } - mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum) - err1 = errors.Wrap(mgr.Append(b), "appending shard data") + resource := api.serverlessStorage.GetShardResource(qtid, partitionNum, shardNum) + err1 = errors.Wrap(resource.Append(b), "appending shard data") if err1 != nil { return err1 } @@ -1860,8 +1860,8 @@ func (api *API) ImportValue(ctx context.Context, qcx *Qcx, req *ImportValueReque return errors.Wrap(err, "marshalling log message") } - mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, shardNum) - err = mgr.Append(b) + resource := api.serverlessStorage.GetShardResource(qtid, partitionNum, shardNum) + err = resource.Append(b) if err != nil { return errors.Wrap(err, "appending shard data") // TODO do we need to set err0 or something? } @@ -3114,13 +3114,13 @@ func (api *API) SnapshotShardData(ctx context.Context, req *dax.SnapshotShardDat return errors.Wrap(err, "getting index/shard readcloser") } - mgr := api.serverlessStorage.GetShardManager(qtid, partitionNum, req.ShardNum) + resource := api.serverlessStorage.GetShardResource(qtid, partitionNum, req.ShardNum) // Bump writelog version while write Tx is held. - if err := mgr.IncrementWLVersion(); err != nil { + if err := resource.IncrementWLVersion(); err != nil { return errors.Wrap(err, "incrementing write log version") } // TODO(jaffee) look into downgrading Tx on RBF to read lock here now that WL version is incremented. - err = mgr.Snapshot(rc) + err = resource.Snapshot(rc) return errors.Wrap(err, "snapshotting shard data") } @@ -3144,12 +3144,12 @@ func (api *API) SnapshotTableKeys(ctx context.Context, req *dax.SnapshotTableKey // TODO(jaffee) need to ensure writes to translation data can't // occur while this is happening. - mgr := api.serverlessStorage.GetTableKeyManager(qtid, req.PartitionNum) - if err := mgr.IncrementWLVersion(); err != nil { + resource := api.serverlessStorage.GetTableKeyResource(qtid, req.PartitionNum) + if err := resource.IncrementWLVersion(); err != nil { return errors.Wrap(err, "incrementing write log version") } // TODO(jaffee) downgrade (currently non-existent) lock to read-only - err = mgr.SnapshotTo(wrTo) + err = resource.SnapshotTo(wrTo) return errors.Wrap(err, "snapshotting table keys") } @@ -3165,12 +3165,12 @@ func (api *API) SnapshotFieldKeys(ctx context.Context, req *dax.SnapshotFieldKey return errors.Wrap(err, "getting index/field writeto") } - mgr := api.serverlessStorage.GetFieldKeyManager(qtid, req.Field) - if err := mgr.IncrementWLVersion(); err != nil { + resource := api.serverlessStorage.GetFieldKeyResource(qtid, req.Field) + if err := resource.IncrementWLVersion(); err != nil { return errors.Wrap(err, "incrementing writelog version") } // TODO(jaffee) downgrade to read lock - err = mgr.SnapshotTo(wrTo) + err = resource.SnapshotTo(wrTo) return errors.Wrap(err, "snapshotTo in FieldKeys") } diff --git a/api_directive.go b/api_directive.go index 4d6065c6b..580be0d78 100644 --- a/api_directive.go +++ b/api_directive.go @@ -355,14 +355,14 @@ func (api *API) pushJobsTableKeys(ctx context.Context, jobs chan<- directiveJobT func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey, partition dax.PartitionNum) error { qtid := tkey.QualifiedTableID() - mgr := api.serverlessStorage.GetTableKeyManager(qtid, partition) - if mgr.IsLocked() { + resource := api.serverlessStorage.GetTableKeyResource(qtid, partition) + if resource.IsLocked() { api.logger().Warnf("skipping loadTableKeys (already held) %s %d", tkey, partition) return nil } // load latest snapshot - if rc, err := mgr.LoadLatestSnapshot(); err != nil { + if rc, err := resource.LoadLatestSnapshot(); err != nil { return errors.Wrap(err, "loading table key snapshot") } else if rc != nil { defer rc.Close() @@ -374,7 +374,7 @@ func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey // define write log loading in a function since we have to do it // before and after locking loadWriteLog := func() error { - writelog, err := mgr.LoadWriteLog() + writelog, err := resource.LoadWriteLog() if err != nil { return errors.Wrap(err, "getting write log reader for table keys") } @@ -402,12 +402,12 @@ func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey } // acquire lock on this partition's keys - if err := mgr.Lock(); err != nil { + if err := resource.Lock(); err != nil { return errors.Wrap(err, "locking table key partition") } // reload writelog in case of changes between last load and - // lock. The manager object takes care of only loading new data. + // lock. The resource object takes care of only loading new data. return loadWriteLog() } @@ -429,14 +429,14 @@ func (api *API) pushJobsFieldKeys(ctx context.Context, jobs chan<- directiveJobT func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax.FieldName) error { qtid := tkey.QualifiedTableID() - mgr := api.serverlessStorage.GetFieldKeyManager(qtid, field) - if mgr.IsLocked() { + resource := api.serverlessStorage.GetFieldKeyResource(qtid, field) + if resource.IsLocked() { api.logger().Warnf("skipping loadFieldKeys (already held) %s %s", tkey, field) return nil } // load latest snapshot - if rc, err := mgr.LoadLatestSnapshot(); err != nil { + if rc, err := resource.LoadLatestSnapshot(); err != nil { return errors.Wrap(err, "loading field key snapshot") } else if rc != nil { defer rc.Close() @@ -448,7 +448,7 @@ func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax. // define write log loading in a function since we have to do it // before and after locking loadWriteLog := func() error { - writelog, err := mgr.LoadWriteLog() + writelog, err := resource.LoadWriteLog() if err != nil { return errors.Wrap(err, "getting write log reader for field keys") } @@ -483,12 +483,12 @@ func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax. } // acquire lock on this partition's keys - if err := mgr.Lock(); err != nil { + if err := resource.Lock(); err != nil { return errors.Wrap(err, "locking field key partition") } // reload writelog in case of changes between last load and - // lock. The manager object takes care of only loading new data. + // lock. The resource object takes care of only loading new data. return loadWriteLog() } @@ -515,13 +515,13 @@ func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.Shar partition := dax.PartitionNum(disco.ShardToShardPartition(string(tkey), uint64(shard), disco.DefaultPartitionN)) - mgr := api.serverlessStorage.GetShardManager(qtid, partition, shard) - if mgr.IsLocked() { + resource := api.serverlessStorage.GetShardResource(qtid, partition, shard) + if resource.IsLocked() { api.logger().Warnf("skipping loadShard (already held) %s %d", tkey, shard) return nil } - if rc, err := mgr.LoadLatestSnapshot(); err != nil { + if rc, err := resource.LoadLatestSnapshot(); err != nil { return errors.Wrap(err, "reading latest snapshot for shard") } else if rc != nil { defer rc.Close() @@ -532,7 +532,7 @@ func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.Shar // define write log loading in a func because we do it twice. loadWriteLog := func() error { - writelog, err := mgr.LoadWriteLog() + writelog, err := resource.LoadWriteLog() if err != nil { return errors.Wrap(err, "") } @@ -641,12 +641,12 @@ func (api *API) loadShard(ctx context.Context, tkey dax.TableKey, shard dax.Shar } // acquire lock on this partition's keys - if err := mgr.Lock(); err != nil { + if err := resource.Lock(); err != nil { return errors.Wrap(err, "locking field key partition") } // reload writelog in case of changes between last load and - // lock. The manager object takes care of only loading new data. + // lock. The resource object takes care of only loading new data. return loadWriteLog() } diff --git a/cluster.go b/cluster.go index 556b6e33e..25f5756c4 100644 --- a/cluster.go +++ b/cluster.go @@ -73,7 +73,7 @@ type cluster struct { // nolint: maligned partitionAssigner string - serverlessStorage *storage.ManagerManager + serverlessStorage *storage.ResourceManager versionStore dax.VersionStore // isComputeNode is set to true if this node is running as a DAX compute @@ -329,8 +329,8 @@ func (c *cluster) appendFieldKeysWriteLog(ctx context.Context, qtid dax.Qualifie if err != nil { return errors.Wrap(err, "marshalling field key map to json") } - mgr := c.serverlessStorage.GetFieldKeyManager(qtid, fieldName) - err = mgr.Append(b) + resource := c.serverlessStorage.GetFieldKeyResource(qtid, fieldName) + err = resource.Append(b) if err != nil { return errors.Wrap(err, "appending field keys") } @@ -350,8 +350,8 @@ func (c *cluster) appendTableKeysWriteLog(ctx context.Context, qtid dax.Qualifie return errors.Wrap(err, "marshalling partition key map to json") } - mgr := c.serverlessStorage.GetTableKeyManager(qtid, partition) - return errors.Wrap(mgr.Append(b), "appending table keys") + resource := c.serverlessStorage.GetTableKeyResource(qtid, partition) + return errors.Wrap(resource.Append(b), "appending table keys") } diff --git a/dax/storage/storage.go b/dax/storage/storage.go index 11d2369ed..d242e0c2d 100644 --- a/dax/storage/storage.go +++ b/dax/storage/storage.go @@ -13,29 +13,29 @@ import ( "github.com/molecula/featurebase/v3/logger" ) -// ManagerManager holds all the various Managers each of which is +// ResourceManager holds all the various Resources each of which is // specific to a particular shard, table key partition or field, but // all of which use the same underlying snapshotter and writelogger. -type ManagerManager struct { +type ResourceManager struct { Snapshotter computer.SnapshotService WriteLogger computer.WriteLogService Logger logger.Logger - mu sync.Mutex - shardManagers map[shardK]*Manager - tableKeyManagers map[tableKeyK]*Manager - fieldKeyManagers map[fieldKeyK]*Manager + mu sync.Mutex + shardResources map[shardK]*Resource + tableKeyResources map[tableKeyK]*Resource + fieldKeyResources map[fieldKeyK]*Resource } -func NewManagerManager(s computer.SnapshotService, w computer.WriteLogService, l logger.Logger) *ManagerManager { - return &ManagerManager{ +func NewResourceManager(s computer.SnapshotService, w computer.WriteLogService, l logger.Logger) *ResourceManager { + return &ResourceManager{ Snapshotter: s, WriteLogger: w, Logger: l, - shardManagers: make(map[shardK]*Manager), - tableKeyManagers: make(map[tableKeyK]*Manager), - fieldKeyManagers: make(map[fieldKeyK]*Manager), + shardResources: make(map[shardK]*Resource), + tableKeyResources: make(map[tableKeyK]*Resource), + fieldKeyResources: make(map[fieldKeyK]*Resource), } } @@ -57,14 +57,14 @@ type fieldKeyK struct { field dax.FieldName } -func (mm *ManagerManager) GetShardManager(qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum) *Manager { +func (mm *ResourceManager) GetShardResource(qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum) *Resource { mm.mu.Lock() defer mm.mu.Unlock() key := shardK{qtid: qtid, partition: partition, shard: shard} - if m, ok := mm.shardManagers[key]; ok { + if m, ok := mm.shardResources[key]; ok { return m } - mm.shardManagers[key] = (&Manager{ + mm.shardResources[key] = (&Resource{ snapshotter: mm.Snapshotter, writeLogger: mm.WriteLogger, bucket: partitionBucket(qtid.Key(), partition), @@ -72,109 +72,109 @@ func (mm *ManagerManager) GetShardManager(qtid dax.QualifiedTableID, partition d log: mm.Logger, }).initialize() - return mm.shardManagers[key] + return mm.shardResources[key] } -func (mm *ManagerManager) RemoveShardManager(qtid dax.QualifiedTableID, partition dax.PartitionNum, shard dax.ShardNum) { +func (mm *ResourceManager) RemoveShardResource(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 { + if m, ok := mm.shardResources[key]; ok { err := m.Unlock() if err != nil { - mm.Logger.Printf("unlocking shard manager during removal: %v", err) + mm.Logger.Printf("unlocking shard resource during removal: %v", err) } - delete(mm.shardManagers, key) + delete(mm.shardResources, key) } } -func (mm *ManagerManager) GetTableKeyManager(qtid dax.QualifiedTableID, partition dax.PartitionNum) *Manager { +func (mm *ResourceManager) GetTableKeyResource(qtid dax.QualifiedTableID, partition dax.PartitionNum) *Resource { mm.mu.Lock() defer mm.mu.Unlock() key := tableKeyK{qtid: qtid, partition: partition} - if m, ok := mm.tableKeyManagers[key]; ok { + if m, ok := mm.tableKeyResources[key]; ok { return m } - mm.tableKeyManagers[key] = (&Manager{ + mm.tableKeyResources[key] = (&Resource{ snapshotter: mm.Snapshotter, writeLogger: mm.WriteLogger, bucket: partitionBucket(qtid.Key(), partition), key: keysFileName, log: mm.Logger, }).initialize() - return mm.tableKeyManagers[key] + return mm.tableKeyResources[key] } -func (mm *ManagerManager) RemoveTableKeyManager(qtid dax.QualifiedTableID, partition dax.PartitionNum) { +func (mm *ResourceManager) RemoveTableKeyResource(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 { + if m, ok := mm.tableKeyResources[key]; ok { err := m.Unlock() if err != nil { - mm.Logger.Printf("unlocking table key manager during removal: %v", err) + mm.Logger.Printf("unlocking table key resource during removal: %v", err) } - delete(mm.tableKeyManagers, key) + delete(mm.tableKeyResources, key) } } -func (mm *ManagerManager) GetFieldKeyManager(qtid dax.QualifiedTableID, field dax.FieldName) *Manager { +func (mm *ResourceManager) GetFieldKeyResource(qtid dax.QualifiedTableID, field dax.FieldName) *Resource { mm.mu.Lock() defer mm.mu.Unlock() key := fieldKeyK{qtid: qtid, field: field} - if m, ok := mm.fieldKeyManagers[key]; ok { + if m, ok := mm.fieldKeyResources[key]; ok { return m } - mm.fieldKeyManagers[key] = (&Manager{ + mm.fieldKeyResources[key] = (&Resource{ snapshotter: mm.Snapshotter, writeLogger: mm.WriteLogger, bucket: fieldBucket(qtid.Key(), field), key: keysFileName, log: mm.Logger, }).initialize() - return mm.fieldKeyManagers[key] + return mm.fieldKeyResources[key] } -func (mm *ManagerManager) RemoveFieldKeyManager(qtid dax.QualifiedTableID, field dax.FieldName) { +func (mm *ResourceManager) RemoveFieldKeyResource(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 { + if m, ok := mm.fieldKeyResources[key]; ok { err := m.Unlock() if err != nil { - mm.Logger.Printf("unlocking field key manager during removal: %v", err) + mm.Logger.Printf("unlocking field key resource during removal: %v", err) } - delete(mm.fieldKeyManagers, key) + delete(mm.fieldKeyResources, key) } } -// RemoveAll unlocks and deletes all Managers held within this -// ManagerManager. -func (mm *ManagerManager) RemoveAll() error { +// RemoveAll unlocks and deletes all resources held within this +// ResourceManager. +func (mm *ResourceManager) RemoveAll() error { mm.mu.Lock() defer mm.mu.Unlock() errList := make([]error, 0) - for k, mgr := range mm.shardManagers { - err := mgr.Unlock() + for k, resource := range mm.shardResources { + err := resource.Unlock() if err != nil && !strings.Contains(err.Error(), "resource was not locked") { errList = append(errList, err) } - delete(mm.shardManagers, k) + delete(mm.shardResources, k) } - for k, mgr := range mm.tableKeyManagers { - err := mgr.Unlock() + for k, resource := range mm.tableKeyResources { + err := resource.Unlock() if err != nil && !strings.Contains(err.Error(), "resource was not locked") { errList = append(errList, err) } - delete(mm.tableKeyManagers, k) + delete(mm.tableKeyResources, k) } - for k, mgr := range mm.fieldKeyManagers { - err := mgr.Unlock() + for k, resource := range mm.fieldKeyResources { + err := resource.Unlock() if err != nil && !strings.Contains(err.Error(), "resource was not locked") { errList = append(errList, err) } - delete(mm.fieldKeyManagers, k) + delete(mm.fieldKeyResources, k) } if len(errList) > 0 { return errors.Errorf("%v", errList) @@ -182,12 +182,12 @@ func (mm *ManagerManager) RemoveAll() error { return nil } -// Manager wraps the snapshotter and writelogger to maintain messy -// state between calls. Manager is *not* threadsafe, care should be -// taken that concurrent calls are not made to Manager methods. The +// Resource wraps the snapshotter and writelogger to maintain messy +// state between calls. Resource is *not* threadsafe, care should be +// taken that concurrent calls are not made to Resource methods. The // exception being that Snapshot and Append are safe to call // concurrently. -type Manager struct { +type Resource struct { snapshotter computer.SnapshotService writeLogger computer.WriteLogService bucket string @@ -204,24 +204,24 @@ type Manager struct { // dirty bool // TODO(jaffee): dirty bit so we can skip snapshotting if there's nothing in WL } -func (m *Manager) initialize() *Manager { +func (m *Resource) initialize() *Resource { m.loadWLsPastVersion = -2 m.latestWLVersion = -1 m.lastWLPos = -1 return m } -// IsLocked checks to see if this particular instance of the Manager +// IsLocked checks to see if this particular instance of the resource // believes it holds the lock. It does not look at the state of // underlying storage to verify the lock. -func (m *Manager) IsLocked() bool { +func (m *Resource) IsLocked() bool { return m.locked } // 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) { +func (m *Resource) LoadLatestSnapshot() (data io.ReadCloser, err error) { snaps, err := m.snapshotter.List(m.bucket, m.key) if err != nil { return nil, errors.Wrap(err, "listing snapshots") @@ -241,7 +241,7 @@ func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) { // it is corrupted/incomplete. We don't want to separately check // the checksum in here because then we'd have to read the whole // snapshot twice. Need a way to catch the checksum error and tell - // Manager to mark that version as bad and remove it, then try + // Resource to mark that version as bad and remove it, then try // LoadLatestSnapshot again. return m.snapshotter.Read(m.bucket, m.key, latest.Version) } @@ -256,7 +256,7 @@ func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) { // snapshot. Subsequent calls to LoadWriteLog will only return new // 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) { +func (m *Resource) 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") } @@ -322,7 +322,7 @@ func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) { // identical to what is was before the lock was acquired. Case (b) // means that quite a lot has happened in between LoadWriteLog and // Lock, and we should probably just die and start over. -func (m *Manager) Lock() error { +func (m *Resource) Lock() error { m.log.Debugf("Lock %s/%s", m.bucket, m.key) // lock is sort of arbitrarily on the write log interface if err := m.writeLogger.Lock(m.bucket, m.key); err != nil { @@ -335,7 +335,7 @@ func (m *Manager) Lock() error { // Append appends the msg to the write log. It will fail if we // haven't properly loaded and gotten a lock for the resource // we're writing to. -func (m *Manager) Append(msg []byte) error { +func (m *Resource) Append(msg []byte) error { m.log.Debugf("Append %s/%s", m.bucket, m.key) if m.latestWLVersion < 0 { return errors.New(errors.ErrUncoded, "can't call append before loading and locking write log") @@ -348,7 +348,7 @@ func (m *Manager) Append(msg []byte) error { // writes which completed prior to the snapshot are in the prior // WL and any that complete after the snapshot are in the // incremented WL. -func (m *Manager) IncrementWLVersion() error { +func (m *Resource) IncrementWLVersion() error { m.log.Debugf("IncrementWLVersion %s/%s", m.bucket, m.key) m.latestWLVersion++ m.lastWLPos = -1 @@ -361,7 +361,7 @@ func (m *Manager) IncrementWLVersion() error { // them to the Snapshot Store. Upon a successful write it will // truncate any write logs which are now incorporated into the // snapshot. -func (m *Manager) Snapshot(rc io.ReadCloser) error { +func (m *Resource) Snapshot(rc io.ReadCloser) error { m.log.Debugf("Snapshot %s/%s", m.bucket, m.key) // latestWLVersion has already been incremented at this point, so // we write that version minus 1. @@ -376,7 +376,7 @@ func (m *Manager) Snapshot(rc io.ReadCloser) error { // SnapshotTo is Snapshot's ugly stepsister supporting the weirdness // of reading from translate stores who we're hoping to off in the // next season. -func (m *Manager) SnapshotTo(wt io.WriterTo) error { +func (m *Resource) SnapshotTo(wt io.WriterTo) error { m.log.Debugf("SnapshotTo %s/%s", m.bucket, m.key) err := m.snapshotter.WriteTo(m.bucket, m.key, m.latestWLVersion-1, wt) if err != nil { @@ -392,7 +392,7 @@ func (m *Manager) SnapshotTo(wt io.WriterTo) error { // a defer), but an implementation based on filesystem locks // should have those removed by the operating system when the // process exits anyway. -func (m *Manager) Unlock() error { +func (m *Resource) Unlock() error { m.log.Debugf("Unlock %s/%s", m.bucket, m.key) if !m.locked { return errors.New(errors.ErrUncoded, "resource was not locked") diff --git a/dax/storage/storage_test.go b/dax/storage/storage_test.go index e6c48e0c0..a8f613fc2 100644 --- a/dax/storage/storage_test.go +++ b/dax/storage/storage_test.go @@ -14,7 +14,7 @@ import ( "github.com/stretchr/testify/assert" ) -func TestManagerManager(t *testing.T) { +func TestResourceManager(t *testing.T) { sdd, err := os.MkdirTemp("", "snaptest*") assert.NoError(t, err) wdd, err := os.MkdirTemp("", "wltest*") @@ -31,7 +31,7 @@ func TestManagerManager(t *testing.T) { DataDir: wdd, }) - mm := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr)) + mm := NewResourceManager(sn, wl, logger.NewStandardLogger(os.Stderr)) qtid := dax.QualifiedTableID{ TableQualifier: dax.TableQualifier{ @@ -44,40 +44,40 @@ func TestManagerManager(t *testing.T) { var n int var d, wld io.ReadCloser - // get a manager and perform normal startup routine on empty data - mgr := mm.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1)) + // get a resource and perform normal startup routine on empty data + resource := mm.GetShardResource(qtid, dax.PartitionNum(1), dax.ShardNum(1)) - d, err = mgr.LoadLatestSnapshot() + d, err = resource.LoadLatestSnapshot() assert.NoError(t, err) assert.Nil(t, d) - wld, err = mgr.LoadWriteLog() + wld, err = resource.LoadWriteLog() assert.NoError(t, err) assert.Nil(t, wld) - err = mgr.Lock() + err = resource.Lock() assert.NoError(t, err) - wld, err = mgr.LoadWriteLog() + wld, err = resource.LoadWriteLog() assert.NoError(t, err) assert.Nil(t, wld) // append some data - err = mgr.Append([]byte("blahblah")) + err = resource.Append([]byte("blahblah")) assert.NoError(t, err) - // a new ManagerManager is necessary so we get a new Manager with - // new internal state instead of a cached Manager. - mm2 := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr)) - // get second manager for same stuff - mgr2 := mm2.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1)) - // load snapshot on 2nd manager (empty) - d, err = mgr2.LoadLatestSnapshot() + // a new ResourceManager is necessary so we get a new Resource with + // new internal state instead of a cached Resource. + mm2 := NewResourceManager(sn, wl, logger.NewStandardLogger(os.Stderr)) + // get second resource for same stuff + resource2 := mm2.GetShardResource(qtid, dax.PartitionNum(1), dax.ShardNum(1)) + // load snapshot on 2nd resource (empty) + d, err = resource2.LoadLatestSnapshot() assert.NoError(t, err) assert.Nil(t, d) - // load WL on 2nd manager (blahblah) - wld, err = mgr2.LoadWriteLog() + // load WL on 2nd resource (blahblah) + wld, err = resource2.LoadWriteLog() assert.NoError(t, err) buf := make([]byte, 16) n, _ = wld.Read(buf) @@ -87,49 +87,49 @@ func TestManagerManager(t *testing.T) { assert.Equal(t, 0, n) assert.Equal(t, io.EOF, err) - // begin snapshot procedure on 1st manager - err = mgr.IncrementWLVersion() + // begin snapshot procedure on 1st resource + err = resource.IncrementWLVersion() assert.NoError(t, err) - // do append on 1st manager mid-snapshot - err = mgr.Append([]byte("blahbla2")) + // do append on 1st resource mid-snapshot + err = resource.Append([]byte("blahbla2")) assert.NoError(t, err) - // snapshot 1st manager + // snapshot 1st resource rc := io.NopCloser(bytes.NewBufferString("hahaha")) - err = mgr.Snapshot(rc) + err = resource.Snapshot(rc) assert.NoError(t, err) - // append again on 1st manager - err = mgr.Append([]byte("blahbla3")) + // append again on 1st resource + err = resource.Append([]byte("blahbla3")) assert.NoError(t, err) - // locking 2nd manager should fail - err = mgr2.Lock() + // locking 2nd resource should fail + err = resource2.Lock() assert.NotNil(t, err) - // exit 1st manager - err = mgr.Unlock() + // exit 1st resource + err = resource.Unlock() assert.NoError(t, err) - // locking 2nd manager should succeed - err = mgr2.Lock() + // locking 2nd resource should succeed + err = resource2.Lock() assert.NoError(t, err) // loading write log should fail since there's been a snapshot // between the last load and locking. - _, err = mgr2.LoadWriteLog() + _, err = resource2.LoadWriteLog() assert.NotNil(t, err) - // mgr2 dies due to error loading write lock - err = mgr2.Unlock() + // resource2 dies due to error loading write lock + err = resource2.Unlock() assert.NoError(t, err) - // get third manager for same stuff - mm3 := NewManagerManager(sn, wl, logger.NewStandardLogger(os.Stderr)) - mgr3 := mm3.GetShardManager(qtid, dax.PartitionNum(1), dax.ShardNum(1)) - // load snapshot on 3nd manager - d, err = mgr3.LoadLatestSnapshot() + // get third resource for same stuff + mm3 := NewResourceManager(sn, wl, logger.NewStandardLogger(os.Stderr)) + resource3 := mm3.GetShardResource(qtid, dax.PartitionNum(1), dax.ShardNum(1)) + // load snapshot on 3nd resource + d, err = resource3.LoadLatestSnapshot() assert.NoError(t, err) buf = make([]byte, 6) n, err = d.Read(buf) @@ -137,8 +137,8 @@ func TestManagerManager(t *testing.T) { assert.Equal(t, "hahaha", string(buf)) assert.Equal(t, nil, err) - // load write log on 3rd manager, get previous 2 writes - wld, err = mgr3.LoadWriteLog() + // load write log on 3rd resource, get previous 2 writes + wld, err = resource3.LoadWriteLog() assert.NoError(t, err) buf = make([]byte, 20) n, _ = wld.Read(buf) @@ -148,12 +148,12 @@ func TestManagerManager(t *testing.T) { assert.Equal(t, 0, n) assert.Equal(t, io.EOF, err) - // lock 3rd manager - err = mgr3.Lock() + // lock 3rd resource + err = resource3.Lock() assert.NoError(t, err) // reload write log (should be empty) - wld, err = mgr3.LoadWriteLog() + wld, err = resource3.LoadWriteLog() assert.NoError(t, err) n, err = wld.Read(make([]byte, 8)) assert.Equal(t, 0, n) diff --git a/server.go b/server.go index 11e04d3a6..2e5424211 100644 --- a/server.go +++ b/server.go @@ -96,7 +96,7 @@ type Server struct { // nolint: maligned executionPlannerFn ExecutionPlannerFn - serverlessStorage *daxstorage.ManagerManager + serverlessStorage *daxstorage.ResourceManager dataframeEnabled bool } @@ -433,7 +433,7 @@ func OptServerExecutionPlannerFn(fn ExecutionPlannerFn) ServerOption { } } -func OptServerServerlessStorage(mm *daxstorage.ManagerManager) ServerOption { +func OptServerServerlessStorage(mm *daxstorage.ResourceManager) ServerOption { return func(s *Server) error { s.serverlessStorage = mm return nil diff --git a/server/server.go b/server/server.go index e08e2b2d5..e5cba0a50 100644 --- a/server/server.go +++ b/server/server.go @@ -77,7 +77,7 @@ type Command struct { queryLogger loggerLogger Registrar computer.Registrar - serverlessStorage *storage.ManagerManager + serverlessStorage *storage.ResourceManager writeLogService computer.WriteLogService snapshotService computer.SnapshotService @@ -553,7 +553,7 @@ func (m *Command) setupServer() error { } if m.writeLogService != nil && m.snapshotService != nil { - m.serverlessStorage = storage.NewManagerManager(m.snapshotService, m.writeLogService, m.logger) + m.serverlessStorage = storage.NewResourceManager(m.snapshotService, m.writeLogService, m.logger) } executionPlannerFn := func(e pilosa.Executor, api *pilosa.API, sql string) sql3.CompilePlanner {