limit frequency of writes for available shards

field.go

@@ -15,11 +15,12 @@
 package pilosa
 
 import (
-	"bufio"
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
+	"log"
 	"math"
 	"os"
 	"path/filepath"
@@ -63,6 +64,26 @@ const (
 	FieldTypeDecimal = "decimal"
 )
 
+type protected struct {
+	mu       sync.Mutex
+	duration time.Duration
+}
+
+func (p *protected) Set(d time.Duration) {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	p.duration = d
+}
+func (p *protected) Get() time.Duration {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return p.duration
+}
+
+var AvailableShardFileFlushDuration = &protected{
+	duration: 5 * time.Second,
+}
+
 // Field represents a container for views.
 type Field struct {
 	mu    sync.RWMutex
@@ -112,6 +133,12 @@ type Field struct {
 	// based on a foreign index; this prevents having to
 	// call holder.index.Keys() every time.
 	usesKeys bool
+
+	// Synchronization primitives needed for async writing of
+	// the remoteAvailableShards
+	availableShardChan chan []byte
+	doneChan           chan struct{}
+	wg                 sync.WaitGroup
 }
 
 // FieldOption is a functional option type for pilosa.fieldOptions.
@@ -348,6 +375,8 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) {
 
 		OpenTranslateStore: OpenInMemTranslateStore,
 	}
+	f.options.ContainsShard = f.containsShard //used for notification optimization
+
 	return f, nil
 }
 
@@ -385,6 +414,12 @@ func (f *Field) AvailableShards() *roaring.Bitmap {
 	return b
 }
 
+func (f *Field) containsShard(shard uint64) bool {
+	f.mu.RLock()
+	defer f.mu.RUnlock()
+	return f.remoteAvailableShards.Contains(shard)
+}
+
 // AddRemoteAvailableShards merges the set of available shards into the current known set
 // and saves the set to a file.
 func (f *Field) AddRemoteAvailableShards(b *roaring.Bitmap) error {
@@ -441,29 +476,11 @@ func (f *Field) saveAvailableShards() error {
 }
 
 func (f *Field) unprotectedSaveAvailableShards() error {
-	path := filepath.Join(f.path, ".available.shards")
-	// Create a temporary file to save to.
-	tempPath := path + tempExt
-
-	// Open or create file.
-	file, err := os.OpenFile(tempPath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0666)
-	if err != nil {
-		return errors.Wrap(err, "opening temporary available shards file")
+	var buf bytes.Buffer
+	if _, err := f.remoteAvailableShards.WriteTo(&buf); err != nil {
+		return errors.Wrap(err, "rendering available shards ")
 	}
-	defer file.Close()
-
-	// Write available shards to file.
-	bw := bufio.NewWriter(file)
-	if _, err = f.remoteAvailableShards.WriteTo(bw); err != nil {
-		return errors.Wrap(err, "writing bitmap to buffer")
-	}
-	bw.Flush()
-
-	// Move snapshot to data file location.
-	if err := os.Rename(tempPath, path); err != nil {
-		return fmt.Errorf("rename snapshot: %s", err)
-	}
-
+	f.availableShardChan <- buf.Bytes()
 	return nil
 }
 
@@ -578,7 +595,10 @@ func (f *Field) Open() error {
 				return errors.Wrap(err, "checking foreign index")
 			}
 		}
-
+		f.availableShardChan = make(chan []byte, 1)
+		f.doneChan = make(chan struct{})
+		f.wg.Add(1)
+		go f.writeAvailableShards()
 		return nil
 	}(); err != nil {
 		f.Close()
@@ -588,6 +608,65 @@ func (f *Field) Open() error {
 	f.logger.Debugf("successfully opened field index/field: %s/%s", f.index, f.name)
 	return nil
 }
+func saveIt(fieldPath string, availableShardBytes []byte) {
+	path := filepath.Join(fieldPath, ".available.shards")
+	// Create a temporary file to save to.
+	tempPath := path + tempExt
+	err := ioutil.WriteFile(tempPath, availableShardBytes, 0666)
+	if err != nil {
+		log.Println("failed to write ", tempPath)
+		return
+	}
+
+	// Move snapshot to data file location.
+	if err := os.Rename(tempPath, path); err != nil {
+		log.Printf("rename snapshot: %s", err)
+	}
+
+}
+func saveItAsync(fieldPath string, availableShardBytes []byte, done chan bool) {
+	if len(availableShardBytes) == 0 {
+		return
+	}
+	go func() {
+		saveIt(fieldPath, availableShardBytes)
+		done <- true
+	}()
+}
+
+func (f *Field) writeAvailableShards() {
+	defer f.wg.Done()
+	ticker := time.NewTicker(AvailableShardFileFlushDuration.Get())
+	var data []byte
+	tracker := make(chan bool)
+	writing := false
+
+	for alive := true; alive; {
+		select {
+		case newdata := <-f.availableShardChan:
+			data = newdata
+		case <-ticker.C:
+			if len(data) > 0 {
+				if !writing {
+					writing = true
+					saveItAsync(f.path, data, tracker)
+					data = nil
+				}
+			}
+		case <-tracker:
+			writing = false
+		case <-f.doneChan:
+			if writing {
+				<-tracker
+			}
+			if len(data) > 0 {
+				saveIt(f.path, data)
+			}
+			alive = false
+		}
+	}
+	ticker.Stop()
+}
 
 // applyTranslateStore opens the configured translate store.
 func (f *Field) applyTranslateStore() error {
@@ -887,7 +966,15 @@ func (f *Field) applyOptions(opt FieldOptions) error {
 func (f *Field) Close() error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-
+	// Shutdown the available shards writer
+	if f.doneChan != nil {
+		f.doneChan <- struct{}{}
+		f.wg.Wait()
+		close(f.availableShardChan)
+		close(f.doneChan)
+		f.availableShardChan = nil
+		f.doneChan = nil
+	}
 	// Close the attribute store.
 	if f.rowAttrStore != nil {
 		_ = f.rowAttrStore.Close()
@@ -1918,12 +2005,16 @@ type FieldOptions struct {
 	Type           string      `json:"type,omitempty"`
 	TimeQuantum    TimeQuantum `json:"timeQuantum,omitempty"`
 	ForeignIndex   string      `json:"foreignIndex"`
+	ContainsShard  func(uint64) bool
 }
 
 // newFieldOptions returns a new instance of FieldOptions
 // with applied and validated functional options.
 func newFieldOptions(opts ...FieldOption) (*FieldOptions, error) {
 	fo := FieldOptions{}
+	fo.ContainsShard = func(uint64) bool {
+		return false
+	}
 	for _, opt := range opts {
 		err := opt(&fo)
 		if err != nil {
@@ -1952,6 +2043,8 @@ func applyDefaultOptions(o *FieldOptions) *FieldOptions {
 		o.CacheType = DefaultCacheType
 		o.CacheSize = DefaultCacheSize
 	}
+	o.ContainsShard = func(uint64) bool { return false } //used for shardnotify optimization
+
 	return o
 }
 

field_internal_test.go

@@ -351,6 +351,7 @@ func TestField_RowTime(t *testing.T) {
 }
 
 func TestField_PersistAvailableShards(t *testing.T) {
+	AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
 	f := OpenField(t, OptFieldTypeDefault())
 
 	// bm represents remote available shards.
@@ -359,6 +360,7 @@ func TestField_PersistAvailableShards(t *testing.T) {
 	if err := f.AddRemoteAvailableShards(bm); err != nil {
 		t.Fatal(err)
 	}
+	time.Sleep(2 * AvailableShardFileFlushDuration.Get())
 
 	// Reload field and verify that shard data is persisted.
 	if err := f.Reopen(); err != nil {
@@ -370,6 +372,7 @@ func TestField_PersistAvailableShards(t *testing.T) {
 }
 
 func TestField_CorruptAvailableShards(t *testing.T) {
+	AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
 	f := OpenField(t, OptFieldTypeDefault())
 
 	// bm represents remote available shards.
@@ -378,6 +381,7 @@ func TestField_CorruptAvailableShards(t *testing.T) {
 	if err := f.AddRemoteAvailableShards(bm); err != nil {
 		t.Fatal(err)
 	}
+	time.Sleep(2 * AvailableShardFileFlushDuration.Get())
 
 	path := filepath.Join(f.path, ".available.shards")
 
@@ -400,6 +404,7 @@ func TestField_CorruptAvailableShards(t *testing.T) {
 }
 
 func TestField_TruncatedAvailableShards(t *testing.T) {
+	AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
 	f := OpenField(t, OptFieldTypeDefault())
 
 	// bm represents remote available shards.
@@ -408,6 +413,7 @@ func TestField_TruncatedAvailableShards(t *testing.T) {
 	if err := f.AddRemoteAvailableShards(bm); err != nil {
 		t.Fatal(err)
 	}
+	time.Sleep(2 * AvailableShardFileFlushDuration.Get())
 
 	path := filepath.Join(f.path, ".available.shards")
 
@@ -428,6 +434,7 @@ func TestField_TruncatedAvailableShards(t *testing.T) {
 // Ensure that persisting available shards having a smaller footprint (for example,
 // when going from a bitmap to a smaller, RLE representation) succeeds.
 func TestField_PersistAvailableShardsFootprint(t *testing.T) {
+	AvailableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
 	f := OpenField(t, OptFieldTypeDefault())
 
 	// bm represents remote available shards.
@@ -442,12 +449,14 @@ func TestField_PersistAvailableShardsFootprint(t *testing.T) {
 	if err := f.AddRemoteAvailableShards(bm); err != nil {
 		t.Fatal(err)
 	}
+	time.Sleep(2 * AvailableShardFileFlushDuration.Get())
 
 	// Reload field and verify that shard data is persisted.
 	if err := f.Reopen(); err != nil {
 		t.Fatal(err)
 	} else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), bm.Slice()) {
-		t.Fatalf("unexpected available shards (reopen). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice())
+		t.Fatalf("unexpected available shards (reopen). expected: %v, \n but got: %v", bm.Slice(), f.remoteAvailableShards.Slice())
+
 	}
 
 	bm1 := roaring.NewBitmap()

holder_internal_test.go

@@ -21,6 +21,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/pilosa/pilosa/v2/roaring"
 )
@@ -98,6 +99,7 @@ func TestHolder_Optn(t *testing.T) {
 		if os.Geteuid() == 0 {
 			t.Skip("Skipping permissions test since user is root.")
 		}
+		AvailableShardFileFlushDuration.Set(100 * time.Millisecond)
 		h := newHolder()
 		defer h.Close()
 
@@ -182,6 +184,7 @@ func TestHolder_Optn(t *testing.T) {
 
 // Ensure holder can clean up orphaned fragments.
 func TestHolderCleaner_CleanHolder(t *testing.T) {
+	AvailableShardFileFlushDuration.Set(100 * time.Millisecond) //shorten the default time to force a file write
 	cluster := NewTestCluster(2)
 
 	// Create a local holder.
@@ -223,6 +226,7 @@ func TestHolderCleaner_CleanHolder(t *testing.T) {
 	if err != nil {
 		t.Fatalf("adding remote shards: %v", err)
 	}
+	time.Sleep(2 * AvailableShardFileFlushDuration.Get())
 
 	// Keep replication the same and ensure we get the expected results.
 	cluster.ReplicaN = 2

utils_internal_test.go

@@ -34,6 +34,7 @@ func NewTestCluster(n int) *cluster {
 		panic(err)
 	}
 
+	AvailableShardFileFlushDuration.Set(100 * time.Millisecond)
 	c := newCluster()
 	c.ReplicaN = 1
 	c.Hasher = NewTestModHasher()

view.go

@@ -60,6 +60,7 @@ type view struct {
 	rowAttrStore  AttrStore
 	logger        logger.Logger
 	snapshotQueue snapshotQueue
+	shardPresent  func(uint64) bool
 }
 
 // newView returns a new instance of View.
@@ -76,9 +77,10 @@ func newView(path, index, field, name string, fieldOptions FieldOptions) *view {
 
 		fragments: make(map[uint64]*fragment),
 
-		broadcaster: NopBroadcaster,
-		stats:       stats.NopStatsClient,
-		logger:      logger.NopLogger,
+		broadcaster:  NopBroadcaster,
+		stats:        stats.NopStatsClient,
+		logger:       logger.NopLogger,
+		shardPresent: fieldOptions.ContainsShard,
 	}
 }
 
@@ -276,6 +278,15 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
 	frag.RowAttrStore = v.rowAttrStore
 
 	v.fragments[shard] = frag
+	v.notifyIfNew(shard)
+	return frag, nil
+}
+
+func (v *view) notifyIfNew(shard uint64) {
+	if v.shardPresent(shard) {
+		return
+	}
+
 	broadcastChan := make(chan struct{})
 
 	go func() {
@@ -299,8 +310,6 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
 	case <-time.After(50 * time.Millisecond):
 		v.logger.Debugf("broadcasting create shard took >50ms")
 	}
-
-	return frag, nil
 }
 
 func (v *view) newFragment(path string, shard uint64) *fragment {

view_internal_test.go

@@ -30,8 +30,9 @@ func mustOpenView(index, field, name string) *view {
 	}
 
 	fo := FieldOptions{
-		CacheType: DefaultCacheType,
-		CacheSize: DefaultCacheSize,
+		CacheType:     DefaultCacheType,
+		CacheSize:     DefaultCacheSize,
+		ContainsShard: func(uint64) bool { return false },
 	}
 
 	v := newView(path, index, field, name, fo)