[dbnode] Get rid of excessive locking when adding new namespaces

src/dbnode/storage/bootstrap.go

@@ -154,8 +154,10 @@ func (m *bootstrapManager) startBootstrap(asyncResult *BootstrapAsyncResult) (Bo
 	m.Unlock()
 	// NB(xichen): disable filesystem manager before we bootstrap to minimize
 	// the impact of file operations on bootstrapping performance
+	m.instrumentation.log.Info("disable fileOps and wait")
 	m.mediator.DisableFileOpsAndWait()
 	defer m.mediator.EnableFileOps()
+	m.instrumentation.log.Info("fileOps disabled")
 
 	var result BootstrapResult
 	asyncResult.bootstrapStarted.Done()

src/dbnode/storage/database.go

@@ -292,22 +292,15 @@ func (d *db) UpdateOwnedNamespaces(newNamespaces namespace.Map) error {
 		}
 	}
 
-	d.Lock()
-	defer d.Unlock()
-
+	d.RLock()
 	removes, adds, updates := d.namespaceDeltaWithLock(newNamespaces)
 	if err := d.logNamespaceUpdate(removes, adds, updates); err != nil {
 		enrichedErr := fmt.Errorf("unable to log namespace updates: %v", err)
 		d.log.Error(enrichedErr.Error())
+		d.RUnlock()
 		return enrichedErr
 	}
-
-	// add any namespaces marked for addition
-	if err := d.addNamespacesWithLock(adds); err != nil {
-		enrichedErr := fmt.Errorf("unable to add namespaces: %v", err)
-		d.log.Error(enrichedErr.Error())
-		return err
-	}
+	d.RUnlock()
 
 	// log that updates and removals are skipped
 	if len(removes) > 0 || len(updates) > 0 {
@@ -317,14 +310,40 @@ func (d *db) UpdateOwnedNamespaces(newNamespaces namespace.Map) error {
 			"restart the process if you want changes to take effect")
 	}
 
-	// enqueue bootstraps if new namespaces
 	if len(adds) > 0 {
-		d.queueBootstrapWithLock()
+		// NB: need to disable fileOps and wait for all the background processes to complete
+		// so that we could update namespaces safely. Otherwise, there is a high change in getting
+		// invariant violation panic because cold/warm flush will receive new namespaces
+		// in the middle of their operations.
+		d.disableFileOpsAndWait()
+		d.Lock()
+		// add any namespaces marked for addition
+		if err := d.addNamespacesWithLock(adds); err != nil {
+			enrichedErr := fmt.Errorf("unable to add namespaces: %w", err)
+			d.log.Error(enrichedErr.Error())
+			d.Unlock()
+			d.enableFileOps()
+			return err
+		}
+		d.Unlock()
+		// enqueue bootstrap and enable file ops when bootstrap is completed
+		d.enqueueBootstrap(d.enableFileOps)
 	}
-
 	return nil
 }
 
+func (d *db) disableFileOpsAndWait() {
+	if mediator := d.mediator; mediator != nil && mediator.IsOpen() {
+		mediator.DisableFileOpsAndWait()
+	}
+}
+
+func (d *db) enableFileOps() {
+	if mediator := d.mediator; mediator != nil && mediator.IsOpen() {
+		mediator.EnableFileOps()
+	}
+}
+
 func (d *db) namespaceDeltaWithLock(newNamespaces namespace.Map) ([]ident.ID, []namespace.Metadata, []namespace.Metadata) {
 	var (
 		existing = d.namespaces
@@ -481,7 +500,6 @@ func (d *db) AssignShardSet(shardSet sharding.ShardSet) {
 
 func (d *db) assignShardSet(shardSet sharding.ShardSet) {
 	d.Lock()
-	defer d.Unlock()
 
 	d.log.Info("assigning shards", zap.Uint32s("shards", shardSet.AllIDs()))
 
@@ -493,14 +511,18 @@ func (d *db) assignShardSet(shardSet sharding.ShardSet) {
 		ns.AssignShardSet(shardSet)
 	}
 
+	d.Unlock()
+
 	if receivedNewShards {
 		// Only trigger a bootstrap if the node received new shards otherwise
 		// the nodes will perform lots of small bootstraps (that accomplish nothing)
 		// during topology changes as other nodes mark their shards as available.
 		//
 		// These small bootstraps can significantly delay topology changes as they prevent
 		// the nodes from marking themselves as bootstrapped and durable, for example.
-		d.queueBootstrapWithLock()
+		d.enqueueBootstrap(func() {
+			d.log.Info("bootstrap completed after receiving new shards")
+		})
 	}
 }
 
@@ -533,20 +555,29 @@ func (d *db) ShardSet() sharding.ShardSet {
 	return shardSet
 }
 
-func (d *db) queueBootstrapWithLock() {
+func (d *db) enqueueBootstrap(onCompleteFn func()) {
 	// Only perform a bootstrap if at least one bootstrap has already occurred. This enables
 	// the ability to open the clustered database and assign shardsets to the non-clustered
 	// database when it receives an initial topology (as well as topology changes) without
 	// triggering a bootstrap until an external call initiates a bootstrap with an initial
 	// call to Bootstrap(). After that initial bootstrap, the clustered database will keep
 	// the non-clustered database bootstrapped by assigning it shardsets which will trigger new
 	// bootstraps since d.bootstraps > 0 will be true.
-	if d.bootstraps > 0 {
+	d.RLock()
+	shouldBootstrap := d.bootstraps > 0
+	d.RUnlock()
+	if shouldBootstrap {
+		d.log.Info("enqueuing bootstrap")
 		bootstrapAsyncResult := d.mediator.BootstrapEnqueue()
-		// NB(linasn): We need to wait for the bootstrap to start and set it's state to Bootstrapping in order
-		// to safely run fileOps in mediator later.
-		bootstrapAsyncResult.WaitForStart()
+		// NB(linasn): We need to wait for the bootstrap to complete and call onCompleteFn.
+		go func() {
+			bootstrapAsyncResult.WaitForComplete()
+			onCompleteFn()
+		}()
+		return
 	}
+
+	onCompleteFn()
 }
 
 func (d *db) Namespace(id ident.ID) (Namespace, bool) {

src/dbnode/storage/entry.go

@@ -39,11 +39,6 @@ import (
 	xtime "github.com/m3db/m3/src/x/time"
 )
 
-const (
-	maxUint64 = ^uint64(0)
-	maxInt64  = int64(maxUint64 >> 1)
-)
-
 // IndexWriter accepts index inserts.
 type IndexWriter interface {
 	// WritePending indexes the provided pending entries.

src/dbnode/storage/types.go

@@ -892,6 +892,11 @@ func (b *BootstrapAsyncResult) WaitForStart() {
 	b.bootstrapStarted.Wait()
 }
 
+// WaitForComplete waits until bootstrap has been completed.
+func (b *BootstrapAsyncResult) WaitForComplete() {
+	b.bootstrapCompleted.Wait()
+}
+
 // databaseFlushManager manages flushing in-memory data to persistent storage.
 type databaseFlushManager interface {
 	// Flush flushes in-memory data to persistent storage.