scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

rename stupid manager names

Browse source 
ManagerManager -> ResourceManager
Manager -> Resource
This commit is contained in:
Matthew Jaffee 2022-12-14 12:30:50 -06:00 • committed by Matthew Jaffee
parent cf1c9dae9a
commit 033be81799
7 changed files with 154 additions and 154 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
api.go
View file

@ -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")
}

36
api_directive.go
View file

@ -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()
}

10
cluster.go
View file

@ -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")
}

126
dax/storage/storage.go
View file

@ -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")

90
dax/storage/storage_test.go
View file

@ -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)

4
server.go
View file

@ -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

4
server/server.go
View file

@ -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 {