scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-05 08:10:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

add translationSyncer interface

Browse source 
This PR adds a translationSyncer interface; I tried to include
comments in the code explaining what's going on. This is taken
from those comments:

translationSyncer provides an interface allowing a function
to notify the server that an action has occurred which requires
the translation sync process to be reset. In general, this
includes anything which modifies schema (add/remove index, etc),
or anything that changes the cluster topology (add/remove node).
I originally considered leveraging the broadcaster since that was
already in place and provides similar event messages, but the
broadcaster is really meant for notifiying other nodes, while
this is more akin to an internal message bus. In fact, I think
a future iteration on this may be to make it more generic so
it can act as an internal message bus where one of the messages
being published is "translationSyncReset".
This commit is contained in:
Travis 2020-03-03 11:01:38 -06:00
parent 92eae8e715
commit 842c820366
5 changed files with 166 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

18
cluster.go
View file

@ -224,6 +224,8 @@ type cluster struct { // nolint: maligned
abortAntiEntropyCh chan struct{}
translationSyncer translationSyncer
mu sync.RWMutex
jobs map[int64]*resizeJob
currentJob *resizeJob
@ -251,6 +253,8 @@ func newCluster() *cluster {
closing: make(chan struct{}),
joining: make(chan struct{}),
translationSyncer: NopTranslationSyncer,
InternalClient: newNopInternalClient(),
logger: logger.NopLogger,
@ -463,7 +467,7 @@ func (c *cluster) unprotectedSetState(state string) {
switch state {
case ClusterStateNormal, ClusterStateDegraded:
// If state is RESIZING -> NORMAL then run cleanup.
// If state is RESIZING -> [NORMAL, DEGRADED] then run cleanup.
if c.state == ClusterStateResizing {
doCleanup = true
}
@ -471,7 +475,17 @@ func (c *cluster) unprotectedSetState(state string) {
c.state = state
if state == ClusterStateResizing {
switch state {
case ClusterStateNormal:
// Because the cluster state is changing to NORMAL,
// we [potentially] need to reset the translation sync.
// If, for example, the cluster has changed size and is
// now settling to NORMAL, the partition ownership may
// have changed, and this will force that to be recalculated.
if err := c.translationSyncer.Reset(); err != nil {
c.logger.Printf("error resetting translation syncer: %s", err)
}
case ClusterStateResizing:
c.abortAntiEntropy()
}

1
go.sum
View file

@ -105,6 +105,7 @@ github.com/pelletier/go-toml v1.2.0 h1:T5zMGML61Wp+FlcbWjRDT7yAxhJNAiPPLOFECq181
github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic=
github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021 h1:ERLyN4p3KS5Fk2ADsDENm2cq0+Lx6sF1sG8uwRlySpU=
github.com/pilosa/memberlist v0.1.4-0.20190415211605-f6512523c021/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
github.com/pilosa/pilosa v1.4.0 h1:nqHNIK4nDslFnem3yDp9R+6TgLdlkY9WdJD88Z83T8U=
github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

100
holder.go
View file

@ -85,6 +85,8 @@ type Holder struct {
OpenTranslateStore OpenTranslateStoreFunc
OpenTranslateReader OpenTranslateReaderFunc
translationSyncer translationSyncer
// Queue of fields (having a foreign index) which have
// opened before their foreign index has opened.
foreignIndexFields []*Field
@ -140,6 +142,8 @@ func NewHolder(partitionN int) *Holder {
OpenTranslateStore: OpenInMemTranslateStore,
translationSyncer: NopTranslationSyncer,
Logger: logger.NopLogger,
}
}
@ -228,7 +232,10 @@ func (h *Holder) Open() error {
h.snapshotQueue.ScanHolder(h)
h.opened.Close()
return nil
// Since the holder is just opening, this is not so much a reset,
// as it is an initial setting of the transation sync process.
return h.translationSyncer.Reset()
}
// checkForeignIndex is a check before applying a foreign
@ -484,6 +491,12 @@ func (h *Holder) createIndex(name string, opt IndexOptions) (*Index, error) {
// Update options.
h.indexes[index.Name()] = index
// Since this is a new index, we need to kick off
// its translation sync.
if err := h.translationSyncer.Reset(); err != nil {
return nil, errors.Wrap(err, "resetting translation sync")
}
return index, nil
}
@ -499,6 +512,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
index.snapshotQueue = h.snapshotQueue
index.OpenTranslateStore = h.OpenTranslateStore
index.translationSyncer = h.translationSyncer
index.holder = h
return index, nil
}
@ -527,7 +541,10 @@ func (h *Holder) DeleteIndex(name string) error {
// Remove reference.
delete(h.indexes, name)
return nil
// I'm not sure if calling Reset() here is necessary
// since closing the index stops its translation
// sync processes.
return h.translationSyncer.Reset()
}
// Field returns the field for an index and name.
@ -922,8 +939,8 @@ func (s *holderSyncer) syncFragment(index, field, view string, shard uint64) err
return nil
}
// ResetTranslationSync reinitializes streaming sync of translation data.
func (s *holderSyncer) ResetTranslationSync() error {
// resetTranslationSync reinitializes streaming sync of translation data.
func (s *holderSyncer) resetTranslationSync() error {
// Stop existing streams.
if err := s.stopTranslationSync(); err != nil {
return errors.Wrap(err, "stop translation sync")
@ -944,6 +961,53 @@ func (s *holderSyncer) ResetTranslationSync() error {
return nil
}
////////////////////////////////////////////////////////////
// translationSyncer provides an interface allowing a function
// to notify the server that an action has occurred which requires
// the translation sync process to be reset. In general, this
// includes anything which modifies schema (add/remove index, etc),
// or anything that changes the cluster topology (add/remove node).
// I originally considered leveraging the broadcaster since that was
// already in place and provides similar event messages, but the
// broadcaster is really meant for notifiying other nodes, while
// this is more akin to an internal message bus. In fact, I think
// a future iteration on this may be to make it more generic so
// it can act as an internal message bus where one of the messages
// being published is "translationSyncReset".
type translationSyncer interface {
Reset() error
}
// NopTranslationSyncer represents a translationSyncer that doesn't do anything.
var NopTranslationSyncer translationSyncer = &nopTranslationSyncer{}
type nopTranslationSyncer struct{}
// Reset is a no-op implementation of translationSyncer Reset method.
func (nopTranslationSyncer) Reset() error { return nil }
// activeTranslationSyncer represents a translationSyncer that resets
// the server's translation syncer.
type activeTranslationSyncer struct {
ch chan struct{}
}
// newActiveTranslationSyncer returns a new instance of activeTranslationSyncer.
func newActiveTranslationSyncer(ch chan struct{}) *activeTranslationSyncer {
return &activeTranslationSyncer{
ch: ch,
}
}
// Reset resets the server's translation syncer.
func (a *activeTranslationSyncer) Reset() error {
a.ch <- struct{}{}
return nil
}
////////////////////////////////////////////////////////////
// stopTranslationSync closes and waits for all outstanding translation readers
// to complete. This should be called before reconnecting to the cluster in case
// of a cluster resize or schema change.
@ -958,17 +1022,36 @@ func (s *holderSyncer) stopTranslationSync() error {
return g.Wait()
}
// setTranslateReadOnlyFlags updates all translation stores to enabled or disable
// setTranslateReadOnlyFlags updates all translation stores to enable or disable
// writing new translation keys. Index stores are writable if the node owns the
// partition. Field stores are writable if the node is the coordinator.
func (s *holderSyncer) setTranslateReadOnlyFlags() {
isCoordinator := s.Cluster.isCoordinator()
for _, index := range s.Holder.Indexes() {
// There is a race condition here:
// if Indexes() returns idx1, and then in another
// process, holder.DeleteIndex(idx1) is called,
// then the next step trying to get TranslateStore(partitionID)
// for an index that is closed (and therefore its transateStores
// no longer exist) will fail with a nil pointer error.
// For now, I just checked that the translateStore hasn't been
// set to nil before trying to use it, but another option may
// be to prevent the translateStores from being zeroed out
// while this process is active. Checking for nil as we do
// really obviates the need for the RLock around the for loop.
// Obtain a read lock on index to prevent Index.Close() from
// destroying the Index.translateStores map before this is
// done using it.
index.mu.RLock()
for partitionID := 0; partitionID < s.Cluster.partitionN; partitionID++ {
ownsPartition := s.Cluster.ownsPartition(s.Node.ID, partitionID)
index.TranslateStore(partitionID).SetReadOnly(!ownsPartition)
if ts := index.TranslateStore(partitionID); ts != nil {
ts.SetReadOnly(!ownsPartition)
}
}
index.mu.RUnlock()
for _, field := range index.Fields() {
field.TranslateStore().SetReadOnly(!isCoordinator)
@ -1030,16 +1113,13 @@ func (s *holderSyncer) initializeIndexTranslateReplication() error {
// initializeFieldTranslateReplication connects the coordinator to stream field data.
func (s *holderSyncer) initializeFieldTranslateReplication() error {
// Skip if coordinator.
if s.Cluster.Node.IsCoordinator {
if s.Cluster.isCoordinator() {
return nil
}
// Build a map of partition offsets to stream from.
m := make(TranslateOffsetMap)
for _, index := range s.Holder.Indexes() {
if !index.Keys() {
continue
}
for _, field := range index.Fields() {
store := field.TranslateStore()
offset, err := store.MaxID()

11
index.go
View file

@ -68,6 +68,8 @@ type Index struct {
// Per-partition translation stores
translateStores map[int]TranslateStore
translationSyncer translationSyncer
// Instantiates new translation stores
OpenTranslateStore OpenTranslateStoreFunc
}
@ -95,6 +97,8 @@ func NewIndex(path, name string, partitionN int) (*Index, error) {
translateStores: make(map[int]TranslateStore),
translationSyncer: NopTranslationSyncer,
OpenTranslateStore: OpenInMemTranslateStore,
}, nil
}
@ -478,6 +482,11 @@ func (i *Index) createField(name string, opt *FieldOptions) (*Field, error) {
// Add to index's field lookup.
i.fields[name] = f
// Kick off the field's translation sync process.
if err := i.translationSyncer.Reset(); err != nil {
return nil, errors.Wrap(err, "resetting translation syncer")
}
return f, nil
}
@ -533,7 +542,7 @@ func (i *Index) DeleteField(name string) error {
// Remove reference.
delete(i.fields, name)
return nil
return i.translationSyncer.Reset()
}
type indexSlice []*Index

76
server.go
View file

@ -77,6 +77,9 @@ type Server struct { // nolint: maligned
isCoordinator bool
syncer holderSyncer
translationSyncer translationSyncer
resetTranslationSyncCh chan struct{}
defaultClient InternalClient
dataDir string
}
@ -316,10 +319,16 @@ func NewServer(opts ...ServerOption) (*Server, error) {
metricInterval: 0,
diagnosticInterval: 0,
resetTranslationSyncCh: make(chan struct{}),
logger: logger.NopLogger,
}
s.cluster.InternalClient = s.defaultClient
s.translationSyncer = newActiveTranslationSyncer(s.resetTranslationSyncCh)
s.holder.translationSyncer = s.translationSyncer
s.cluster.translationSyncer = s.translationSyncer
s.diagnostics.server = s
for _, opt := range opts {
@ -503,6 +512,18 @@ func (s *Server) Open() error {
log.Println(errors.Wrap(err, "logging startup"))
}
// Set up the holderSyncer.
s.syncer.Holder = s.holder
s.syncer.Node = s.cluster.Node
s.syncer.Cluster = s.cluster
s.syncer.Closing = s.closing
s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
// Start background process listening for translation
// sync resets.
s.wg.Add(1)
go func() { defer s.wg.Done(); s.monitorResetTranslationSync() }()
// Open Cluster management.
if err := s.cluster.waitForStarted(); err != nil {
return errors.Wrap(err, "opening Cluster")
@ -523,16 +544,6 @@ func (s *Server) Open() error {
// buffered channel.
s.cluster.listenForJoins()
s.syncer.Holder = s.holder
s.syncer.Node = s.cluster.Node
s.syncer.Cluster = s.cluster
s.syncer.Closing = s.closing
s.syncer.Stats = s.holder.Stats.WithTags("HolderSyncer")
if err := s.syncer.ResetTranslationSync(); err != nil {
return err
}
// Start background monitoring.
s.wg.Add(3)
go func() { defer s.wg.Done(); s.monitorAntiEntropy() }()
@ -595,6 +606,34 @@ func (s *Server) SyncData() error {
return errors.Wrap(s.syncer.SyncHolder(), "syncing holder")
}
// monitorResetTranslationSync is a background process which
// listens for events indicating the need to reset the translation
// sync processes.
func (s *Server) monitorResetTranslationSync() {
s.logger.Printf("holder translation sync monitor initializing")
for {
// Wait for a reset or a close.
select {
case <-s.closing:
return
case <-s.resetTranslationSyncCh:
s.logger.Printf("holder translation sync beginning")
s.wg.Add(1)
go func() {
// Obtaining this lock ensures that there is only
// once instance of resetTranslationSync() running
// at once.
s.syncer.mu.Lock()
defer s.syncer.mu.Unlock()
defer s.wg.Done()
if err := s.syncer.resetTranslationSync(); err != nil {
s.logger.Printf("holder translation sync error: err=%s", err)
}
}()
}
}
}
func (s *Server) monitorAntiEntropy() {
if s.antiEntropyInterval == 0 || s.cluster.ReplicaN <= 1 {
return // anti entropy disabled
@ -666,16 +705,10 @@ func (s *Server) receiveMessage(m Message) error {
if err != nil {
return err
}
if err := s.syncer.ResetTranslationSync(); err != nil {
return err
}
case *DeleteIndexMessage:
if err := s.holder.DeleteIndex(obj.Index); err != nil {
return err
}
if err := s.syncer.ResetTranslationSync(); err != nil {
return err
}
case *CreateFieldMessage:
idx := s.holder.Index(obj.Index)
if idx == nil {
@ -686,17 +719,11 @@ func (s *Server) receiveMessage(m Message) error {
if err != nil {
return err
}
if err := s.syncer.ResetTranslationSync(); err != nil {
return err
}
case *DeleteFieldMessage:
idx := s.holder.Index(obj.Index)
if err := idx.DeleteField(obj.Field); err != nil {
return err
}
if err := s.syncer.ResetTranslationSync(); err != nil {
return err
}
case *DeleteAvailableShardMessage:
f := s.holder.Field(obj.Index, obj.Field)
if err := f.RemoveAvailableShard(obj.ShardID); err != nil {
@ -725,11 +752,6 @@ func (s *Server) receiveMessage(m Message) error {
if err != nil {
return err
}
if s.syncer.Cluster != nil {
if err := s.syncer.ResetTranslationSync(); err != nil {
return err
}
}
case *ResizeInstruction:
err := s.cluster.followResizeInstruction(obj)
if err != nil {