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

Merge pull request #196 from tgruben/async-available-shards

Browse source 
limit frequency of writes for available shards broadcast
This commit is contained in:
tgruben 2020-04-06 09:08:04 -05:00 • committed by GitHub
commit 64caffebe8
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 166 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cluster.go
View file

@ -1443,7 +1443,7 @@ func (c *cluster) followResizeInstruction(instr *ResizeInstruction) error {
return errors.Wrap(err, "merging cluster status")
}
c.logger.Printf("done MergeClusterStatus, start goroutine")
c.logger.Printf("done MergeClusterStatus, start goroutine (%s)", c.Node.ID)
// The actual resizing runs in a goroutine because we don't want to block
// the distribution of other ResizeInstructions to the rest of the cluster.

138
field.go
View file

@ -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,7 @@ func newField(path, index, name string, opts FieldOption) (*Field, error) {
OpenTranslateStore: OpenInMemTranslateStore,
}
return f, nil
}
@ -385,6 +413,13 @@ func (f *Field) AvailableShards() *roaring.Bitmap {
return b
}
// constainsShard is used for limiting unnecessary CreateShard broadcast
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)
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 blockingWriteAvailableShards(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 nonBlockingWriteAvailableShards(fieldPath string, availableShardBytes []byte, done chan bool) {
if len(availableShardBytes) == 0 {
return
}
go func() {
blockingWriteAvailableShards(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
nonBlockingWriteAvailableShards(f.path, data, tracker)
data = nil
}
}
case <-tracker:
writing = false
case <-f.doneChan:
if writing { //wait to writing is complete
<-tracker
}
if len(data) > 0 {
blockingWriteAvailableShards(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()
@ -1099,6 +1186,7 @@ func (f *Field) newView(path, name string) *view {
view.rowAttrStore = f.rowAttrStore
view.stats = f.Stats
view.broadcaster = f.broadcaster
view.remoteShardPresent = f.containsShard
if f.snapshotQueue != nil {
view.snapshotQueue = f.snapshotQueue
}

11
field_internal_test.go
View file

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

17
holder.go
View file

@ -1185,7 +1185,7 @@ func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) {
// Find appropriate store.
f := s.Holder.Field(entry.Index, entry.Field)
if f == nil {
s.Holder.Logger.Printf("field not found: %q/%q", entry.Index, entry.Field)
s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field)
return
}
@ -1233,6 +1233,15 @@ func (c *holderCleaner) CleanHolder() error {
// Get the fragments registered in memory.
for _, field := range index.Fields() {
// deletedShards is used to track which shards for the field
// were deleted. Any shards that get deleted from this node
// get added to remoteAvailableShards. This is done because
// the CleanHolder process is cleaning up shards which got
// moved to other nodes. Because those shards still exist
// (just no longer on this particular node), this node still
// needs to consider each of them as an available shard in
// the cluster.
var deletedShards []uint64
for _, view := range field.views() {
for _, fragment := range view.allFragments() {
fragShard := fragment.shard
@ -1244,6 +1253,12 @@ func (c *holderCleaner) CleanHolder() error {
if err := view.deleteFragment(fragShard); err != nil {
return errors.Wrap(err, "deleting fragment")
}
deletedShards = append(deletedShards, fragShard)
}
}
if len(deletedShards) > 0 {
if err := field.AddRemoteAvailableShards(roaring.NewBitmap(deletedShards...)); err != nil {
return errors.Wrap(err, "adding remote available shards")
}
}
}

4
holder_internal_test.go
View file

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

2
http/client.go
View file

@ -1115,6 +1115,7 @@ func (c *InternalClient) SendMessage(ctx context.Context, uri *pilosa.URI, msg [
req.Header.Set("Content-Type", "application/x-protobuf")
req.Header.Set("User-Agent", "pilosa/"+pilosa.Version)
req.Header.Set("Accept", "application/json")
req.Header.Set("Connection", "keep-alive")
// Execute request.
resp, err := c.executeRequest(req.WithContext(ctx))
@ -1232,6 +1233,7 @@ func (c *InternalClient) TranslateIDsNode(ctx context.Context, uri *pilosa.URI,
// is closed.
func (c *InternalClient) executeRequest(req *http.Request) (*http.Response, error) {
tracing.GlobalTracer.InjectHTTPHeaders(req)
req.Close = false
resp, err := c.httpClient.Do(req)
if err != nil {
if resp != nil {

2
http/handler.go
View file

@ -1590,7 +1590,7 @@ func (h *Handler) handlePostClusterMessage(w http.ResponseWriter, r *http.Reques
http.Error(w, "Unsupported media type", http.StatusUnsupportedMediaType)
return
}
defer r.Body.Close()
err := h.api.ClusterMessage(r.Context(), r.Body)
if err != nil {
// TODO this was the previous behavior, but perhaps not everything is a bad request

1
server/cluster_test.go
View file

@ -206,7 +206,6 @@ func TestClusterResize_AddNode(t *testing.T) {
`); err != nil {
t.Fatal(err)
}
// exp is the expected result for the Row queries that follow.
exp := `{"results":[{"attrs":{},"columns":[1,1300000]}]}` + "\n"

1
utils_internal_test.go
View file

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

28
view.go
View file

@ -55,11 +55,12 @@ type view struct {
// Fragments by shard.
fragments map[uint64]*fragment
broadcaster broadcaster
stats stats.StatsClient
rowAttrStore AttrStore
logger logger.Logger
snapshotQueue snapshotQueue
broadcaster broadcaster
stats stats.StatsClient
rowAttrStore AttrStore
logger logger.Logger
snapshotQueue snapshotQueue
remoteShardPresent 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,
remoteShardPresent: func(uint64) bool { return false },
}
}
@ -276,6 +278,14 @@ func (v *view) CreateFragmentIfNotExists(shard uint64) (*fragment, error) {
frag.RowAttrStore = v.rowAttrStore
v.fragments[shard] = frag
v.notifyIfNewShard(shard)
return frag, nil
}
func (v *view) notifyIfNewShard(shard uint64) {
if v.remoteShardPresent(shard) { //checks the fields remoteShards bitmap to see if broadcast needed
return
}
broadcastChan := make(chan struct{})
go func() {
@ -299,8 +309,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 {