diff --git a/api.go b/api.go index f440df003..66c22c962 100644 --- a/api.go +++ b/api.go @@ -68,6 +68,10 @@ func (api *API) Holder() *Holder { return api.holder } +func (api *API) logger() logger.Logger { + return api.server.logger +} + // apiOption is a functional option type for pilosa.API type apiOption func(*API) error diff --git a/api_directive.go b/api_directive.go index 9c00a1dea..ed13f6b5e 100644 --- a/api_directive.go +++ b/api_directive.go @@ -355,6 +355,10 @@ func (api *API) loadTableKeys(ctx context.Context, idx *Index, tkey dax.TableKey qtid := tkey.QualifiedTableID() mgr := api.serverlessStorage.GetTableKeyManager(qtid, partition) + if mgr.IsLocked() { + api.logger().Debugf("skipping loadTableKeys (already held) %s %d", tkey, partition) + return nil + } // load latest snapshot rc, err := mgr.LoadLatestSnapshot() @@ -428,6 +432,10 @@ func (api *API) loadFieldKeys(ctx context.Context, tkey dax.TableKey, field dax. qtid := tkey.QualifiedTableID() mgr := api.serverlessStorage.GetFieldKeyManager(qtid, field) + if mgr.IsLocked() { + api.logger().Debugf("skipping loadFieldKeys (already held) %s %s", tkey, field) + return nil + } // load latest snapshot rc, err := mgr.LoadLatestSnapshot() @@ -512,6 +520,11 @@ 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() { + api.logger().Debugf("skipping loadShard (already held) %s %d", tkey, shard) + return nil + } + rc, err := mgr.LoadLatestSnapshot() if err != nil { return errors.Wrap(err, "reading latest snapshot for shard") diff --git a/dax/docker-compose.yml b/dax/docker-compose.yml index f97cd4f56..c6ca7f8f7 100644 --- a/dax/docker-compose.yml +++ b/dax/docker-compose.yml @@ -35,6 +35,7 @@ services: FEATUREBASE_COMPUTER_RUN: "true" FEATUREBASE_COMPUTER_CONFIG_MDS_ADDRESS: "mds:8080/mds" FEATUREBASE_COMPUTER_CONFIG_DATA_DIR: /dax-data/computer + FEATUREBASE_COMPUTER_CONFIG_VERBOSE: true FEATUREBASE_BIND: 0.0.0.0:8080 FEATUREBASE_VERBOSE: "true" FEATUREBASE_STORAGE_METHOD: boltdb diff --git a/dax/storage/storage.go b/dax/storage/storage.go index 5ee55ade0..0839e9af4 100644 --- a/dax/storage/storage.go +++ b/dax/storage/storage.go @@ -200,6 +200,8 @@ type Manager struct { lastWLPos int locked bool + + // dirty bool // TODO(jaffee): dirty bit so we can skip snapshotting if there's nothing in WL } func (m *Manager) initialize() *Manager { @@ -209,6 +211,13 @@ func (m *Manager) initialize() *Manager { return m } +// IsLocked checks to see if this particular instance of the Manager +// believes it holds the lock. It does not look at the state of +// underlying storage to verify the lock. +func (m *Manager) 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. @@ -217,6 +226,7 @@ func (m *Manager) LoadLatestSnapshot() (data io.ReadCloser, err error) { if err != nil { return nil, errors.Wrap(err, "listing snapshots") } + m.log.Debugf("LoadLatestSnapshot %s/%s: list: %v", m.bucket, m.key, snaps) m.lastWLPos = 0 if len(snaps) == 0 { @@ -251,6 +261,11 @@ func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) { return nil, errors.New(errors.ErrUncoded, "LoadWriteLog called in inconsistent state, can't tell what version to load from") } wLogs, err := m.writeLogger.List(m.bucket, m.key) + if err != nil { + return nil, errors.Wrap(err, "listing write logs") + } + + m.log.Debugf("LoadWriteLog %s/%s: list: %v", m.bucket, m.key, wLogs) versions := make([]int, 0, len(wLogs)) for _, log := range wLogs { @@ -308,6 +323,7 @@ func (m *Manager) LoadWriteLog() (data io.ReadCloser, err error) { // 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 { + 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 { return errors.Wrap(err, "acquiring lock") @@ -320,6 +336,7 @@ func (m *Manager) Lock() error { // haven't properly loaded and gotten a lock for the resource // we're writing to. func (m *Manager) 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") } @@ -332,6 +349,7 @@ func (m *Manager) Append(msg []byte) error { // WL and any that complete after the snapshot are in the // incremented WL. func (m *Manager) IncrementWLVersion() error { + m.log.Debugf("IncrementWLVersion %s/%s", m.bucket, m.key) m.latestWLVersion++ m.lastWLPos = -1 m.loadWLsPastVersion = -1 @@ -344,6 +362,7 @@ func (m *Manager) IncrementWLVersion() error { // truncate any write logs which are now incorporated into the // snapshot. func (m *Manager) 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. err := m.snapshotter.Write(m.bucket, m.key, m.latestWLVersion-1, rc) @@ -358,6 +377,7 @@ func (m *Manager) Snapshot(rc io.ReadCloser) error { // of reading from translate stores who we're hoping to off in the // next season. func (m *Manager) 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 { return errors.Wrap(err, "writing snapshot SnapshotTo") @@ -373,6 +393,7 @@ 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 { + 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/server.go b/server.go index f707bb8ac..bb0ac7d58 100644 --- a/server.go +++ b/server.go @@ -806,6 +806,7 @@ func (s *Server) Close() error { var errh, errd error var errhs error var errc error + var errSS error if s.cluster != nil { errc = s.cluster.close() @@ -817,7 +818,9 @@ func (s *Server) Close() error { if s.holder != nil { errh = s.holder.Close() } - errSS := s.serverlessStorage.RemoveAll() + if s.serverlessStorage != nil { + errSS = s.serverlessStorage.RemoveAll() + } // prefer to return holder error over cluster // error. This order is somewhat arbitrary. It would be better if we had