From d90a9b97c8ce442d5d9c2c5ca1ade255d94dd987 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Fri, 30 Sep 2016 11:46:01 -0600 Subject: [PATCH] Add profile & bitmap attribute anti-entropy. Adds block-based anti-entropy for the attribute stores and hooks into the existing `IndexSyncer` to regulate frequency of syncs. --- attr.go | 178 +++++++++++++++++++++++++++++++++++++++++++++ attr_test.go | 42 +++++++++++ client.go | 74 +++++++++++++++++++ cmd/pilosa/main.go | 3 + fragment.go | 1 + handler.go | 127 ++++++++++++++++++++++++++++++++ handler_test.go | 118 ++++++++++++++++++++++++++++++ index.go | 128 ++++++++++++++++++++++++++++++-- server.go | 4 +- 9 files changed, 668 insertions(+), 7 deletions(-) diff --git a/attr.go b/attr.go index bf22faef9..abd1baabb 100644 --- a/attr.go +++ b/attr.go @@ -1,6 +1,8 @@ package pilosa import ( + "bytes" + "crypto/sha1" "encoding/binary" "fmt" "sort" @@ -12,6 +14,9 @@ import ( "github.com/umbel/pilosa/internal" ) +// AttrBlockSize is the size of attribute blocks for anti-entropy. +const AttrBlockSize = 100 + // AttrStore represents a storage layer for attributes. type AttrStore struct { mu sync.Mutex @@ -150,6 +155,69 @@ func (s *AttrStore) SetBulkAttrs(m map[uint64]map[string]interface{}) error { return nil } +// Blocks returns a list of all blocks in the store. +func (s *AttrStore) Blocks() ([]AttrBlock, error) { + tx, err := s.db.Begin(false) + if err != nil { + return nil, err + } + defer tx.Rollback() + + // Wrap cursor to segment by block. + cur := newBlockCursor(tx.Bucket([]byte("attrs")).Cursor(), AttrBlockSize) + + // Iterate over each block. + var blocks []AttrBlock + for cur.nextBlock() { + block := AttrBlock{ID: cur.blockID()} + + // Compute checksum of every key/value in block. + h := sha1.New() + for k, v := cur.next(); k != nil; k, v = cur.next() { + h.Write(k) + h.Write(v) + } + block.Checksum = h.Sum(nil) + + // Append block. + blocks = append(blocks, block) + } + + return blocks, nil +} + +// BlockData returns all data for a single block. +func (s *AttrStore) BlockData(i uint64) (map[uint64]map[string]interface{}, error) { + m := make(map[uint64]map[string]interface{}) + + // Start read-only transaction. + tx, err := s.db.Begin(false) + if err != nil { + return nil, err + } + defer tx.Rollback() + + // Move to the start of the block. + min := u64tob(uint64(i) * AttrBlockSize) + max := u64tob(uint64(i+1) * AttrBlockSize) + cur := tx.Bucket([]byte("attrs")).Cursor() + for k, v := cur.Seek(min); k != nil; k, v = cur.Next() { + // Exit if we're past the end of the block. + if bytes.Compare(k, max) != -1 { + break + } + + // Decode attribute map and associate with id. + var pb internal.AttrMap + if err := proto.Unmarshal(v, &pb); err != nil { + return nil, err + } + m[btou64(k)] = decodeAttrs(pb.GetAttrs()) + } + + return m, nil +} + // txAttrs returns a map of attributes for a bitmap. func txAttrs(tx *bolt.Tx, id uint64) (map[string]interface{}, error) { v := tx.Bucket([]byte("attrs")).Get(u64tob(id)) @@ -284,3 +352,113 @@ func btou64(b []byte) uint64 { return binary.BigEndian.Uint64(b) } // emptyMap is a reusable map that contains no keys. var emptyMap = make(map[string]interface{}) + +// AttrBlock represents a checksummed block of the attribute store. +type AttrBlock struct { + ID uint64 `json:"id"` + Checksum []byte `json:"checksum"` +} + +// AttrBlocks represents a list of blocks. +type AttrBlocks []AttrBlock + +// Diff returns a list of block ids that are different or are new in other. +// Block lists must be in sorted order. +func (a AttrBlocks) Diff(other []AttrBlock) []uint64 { + var ids []uint64 + for { + // Read next block from each list. + var blk0, blk1 *AttrBlock + if len(a) > 0 { + blk0 = &a[0] + } + if len(other) > 0 { + blk1 = &other[0] + } + + // Exit if "a" contains no more blocks. + if blk0 == nil { + return ids + } + + // Add block ID if it's different or if it's only in "a". + if blk1 == nil || blk0.ID < blk1.ID { + ids = append(ids, blk0.ID) + a = a[1:] + } else if blk1.ID < blk0.ID { + other = other[1:] + } else { + if !bytes.Equal(blk0.Checksum, blk1.Checksum) { + ids = append(ids, blk0.ID) + } + a, other = a[1:], other[1:] + } + } +} + +// blockCursor represents a cursor for iterating over blocks of a bolt bucket. +type blockCursor struct { + cur *bolt.Cursor + base uint64 + n uint64 + + buf struct { + key []byte + value []byte + filled bool + } +} + +// newBlockCursor returns a new block cursor that wraps cur using n sized blocks. +func newBlockCursor(c *bolt.Cursor, n int) blockCursor { + cur := blockCursor{ + cur: c, + n: uint64(n), + } + cur.buf.key, cur.buf.value = c.First() + cur.buf.filled = true + return cur +} + +// blockID returns the current block ID. Only valid after call to nextBlock(). +func (cur *blockCursor) blockID() uint64 { return cur.base } + +// nextBlock moves the cursor to the next block. +// Returns true if another block exists, otherwise returns false. +func (cur *blockCursor) nextBlock() bool { + if cur.buf.key == nil { + return false + } + + cur.base = binary.BigEndian.Uint64(cur.buf.key) / cur.n + return true +} + +// next returns the next key/value within the block. +// Returns nils at the end of the block. +func (cur *blockCursor) next() (key, value []byte) { + // Use buffered value, if set. + if cur.buf.filled { + key, value = cur.buf.key, cur.buf.value + cur.buf.filled = false + return key, value + } + + // Read next key. + key, value = cur.cur.Next() + + // Fill buffer for EOF. + if key == nil { + cur.buf.key, cur.buf.value, cur.buf.filled = key, value, false + return nil, nil + } + + // Parse key and buffer if outside of block. + id := binary.BigEndian.Uint64(key) + if id/cur.n > cur.base { + cur.buf.key, cur.buf.value, cur.buf.filled = key, value, true + return nil, nil + } + + return key, value +} diff --git a/attr_test.go b/attr_test.go index c5a909642..20c57a51c 100644 --- a/attr_test.go +++ b/attr_test.go @@ -70,6 +70,48 @@ func TestAttrStore_Attrs_Unset(t *testing.T) { } } +// Ensure attribute block checksums can be returned. +func TestAttrStore_Blocks(t *testing.T) { + s := MustOpenAttrStore() + defer s.Close() + + // Set attributes. + if err := s.SetAttrs(1, map[string]interface{}{"A": uint64(100)}); err != nil { + t.Fatal(err) + } else if err := s.SetAttrs(2, map[string]interface{}{"A": uint64(200)}); err != nil { + t.Fatal(err) + } else if err := s.SetAttrs(100, map[string]interface{}{"B": "VALUE"}); err != nil { + t.Fatal(err) + } else if err := s.SetAttrs(350, map[string]interface{}{"C": "FOO"}); err != nil { + t.Fatal(err) + } + + // Retrieve blocks. + blks0, err := s.Blocks() + if err != nil { + t.Fatal(err) + } else if len(blks0) != 3 || blks0[0].ID != 0 || blks0[1].ID != 1 || blks0[2].ID != 3 { + t.Fatalf("unexpected blocks: %#v", blks0) + } + + // Change second block. + if err := s.SetAttrs(100, map[string]interface{}{"X": 12}); err != nil { + t.Fatal(err) + } + + // Ensure second block changed. + blks1, err := s.Blocks() + if err != nil { + t.Fatal(err) + } else if !reflect.DeepEqual(blks0[0], blks1[0]) { + t.Fatalf("block 0 mismatch: %#v != %#v", blks0[0], blks1[0]) + } else if reflect.DeepEqual(blks0[1], blks1[1]) { + t.Fatalf("block 1 match: %#v ", blks0[0]) + } else if !reflect.DeepEqual(blks0[2], blks1[2]) { + t.Fatalf("block 2 mismatch: %#v != %#v", blks0[2], blks1[2]) + } +} + // AttrStore represents a test wrapper for pilosa.AttrStore. type AttrStore struct { *pilosa.AttrStore diff --git a/client.go b/client.go index fde2313d8..67389c33c 100644 --- a/client.go +++ b/client.go @@ -630,6 +630,80 @@ func (c *Client) BlockData(db, frame string, slice uint64, block int) ([]uint64, return rsp.BitmapIDs, rsp.ProfileIDs, nil } +// ProfileAttrDiff returns data from differing blocks on a remote host. +func (c *Client) ProfileAttrDiff(db string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { + u := url.URL{ + Scheme: "http", + Host: c.host, + Path: "/db/attr/diff", + RawQuery: url.Values{"db": {db}}.Encode(), + } + + // Encode request. + buf, err := json.Marshal(postDBAttrDiffRequest{DB: db, Blocks: blks}) + if err != nil { + return nil, err + } + + // Send request. + resp, err := c.HTTPClient.Post(u.String(), "application/json", bytes.NewReader(buf)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + // Return error if status is not OK. + switch resp.StatusCode { + case http.StatusOK: // ok + default: + return nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode) + } + + // Decode response object. + var rsp postDBAttrDiffResponse + if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + return nil, err + } + return rsp.Attrs, nil +} + +// BitmapAttrDiff returns data from differing blocks on a remote host. +func (c *Client) BitmapAttrDiff(db, frame string, blks []AttrBlock) (map[uint64]map[string]interface{}, error) { + u := url.URL{ + Scheme: "http", + Host: c.host, + Path: "/frame/attr/diff", + RawQuery: url.Values{"db": {db}, "frame": {frame}}.Encode(), + } + + // Encode request. + buf, err := json.Marshal(postFrameAttrDiffRequest{DB: db, Frame: frame, Blocks: blks}) + if err != nil { + return nil, err + } + + // Send request. + resp, err := c.HTTPClient.Post(u.String(), "application/json", bytes.NewReader(buf)) + if err != nil { + return nil, err + } + defer resp.Body.Close() + + // Return error if status is not OK. + switch resp.StatusCode { + case http.StatusOK: // ok + default: + return nil, fmt.Errorf("unexpected status: code=%d", resp.StatusCode) + } + + // Decode response object. + var rsp postFrameAttrDiffResponse + if err := json.NewDecoder(resp.Body).Decode(&rsp); err != nil { + return nil, err + } + return rsp.Attrs, nil +} + // Bit represents the location of a single bit. type Bit struct { BitmapID uint64 diff --git a/cmd/pilosa/main.go b/cmd/pilosa/main.go index 7107745fe..a1a2fc8ac 100644 --- a/cmd/pilosa/main.go +++ b/cmd/pilosa/main.go @@ -134,6 +134,9 @@ func (m *Main) Run(args ...string) error { m.Server.Host = m.Config.Host m.Server.Cluster = m.Config.PilosaCluster() + // Set configuration options. + m.Server.AntiEntropyInterval = time.Duration(m.Config.AntiEntropy.Interval) + // Initialize server. if err := m.Server.Open(); err != nil { return err diff --git a/fragment.go b/fragment.go index 4b5eeb9bb..e9d81e8fd 100644 --- a/fragment.go +++ b/fragment.go @@ -1254,6 +1254,7 @@ func (s *FragmentSyncer) SyncFragment() error { //fmt.Println("no place to replicate", s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice()) return nil } + // Create a set of blocks. blockSets := make([][]FragmentBlock, 0, len(nodes)) for _, node := range nodes { diff --git a/handler.go b/handler.go index c5a4607d4..b686cbbb9 100644 --- a/handler.go +++ b/handler.go @@ -106,6 +106,20 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { default: http.Error(w, "method not allowed", http.StatusMethodNotAllowed) } + case "/db/attr/diff": + switch r.Method { + case "POST": + h.handlePostDBAttrDiff(w, r) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } + case "/frame/attr/diff": + switch r.Method { + case "POST": + h.handlePostFrameAttrDiff(w, r) + default: + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + } case "/fragment/nodes": switch r.Method { case "GET": @@ -250,6 +264,119 @@ type sliceMaxResponse struct { SliceMax uint64 `json:"SliceMax"` } +// handlePostDBAttrDiff handles POST /db/attr/diff requests. +func (h *Handler) handlePostDBAttrDiff(w http.ResponseWriter, r *http.Request) { + // Decode request. + var req postDBAttrDiffRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + // Retrieve database from index. + db, err := h.Index.CreateDBIfNotExists(req.DB) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Retrieve local blocks. + blks, err := db.ProfileAttrStore().Blocks() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Read all attributes from all mismatched blocks. + attrs := make(map[uint64]map[string]interface{}) + for _, blockID := range AttrBlocks(blks).Diff(req.Blocks) { + // Retrieve block data. + m, err := db.ProfileAttrStore().BlockData(blockID) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Copy to database-wide struct. + for k, v := range m { + attrs[k] = v + } + } + + // Encode response. + if err := json.NewEncoder(w).Encode(postDBAttrDiffResponse{ + Attrs: attrs, + }); err != nil { + h.logger().Printf("response encoding error: %s", err) + } +} + +type postDBAttrDiffRequest struct { + DB string `json:"db"` + Blocks []AttrBlock `json:"blocks"` +} + +type postDBAttrDiffResponse struct { + Attrs map[uint64]map[string]interface{} `json:"attrs"` +} + +// handlePostFrameAttrDiff handles POST /frame/attr/diff requests. +func (h *Handler) handlePostFrameAttrDiff(w http.ResponseWriter, r *http.Request) { + // Decode request. + var req postFrameAttrDiffRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + + // Retrieve database from index. + f, err := h.Index.CreateFrameIfNotExists(req.DB, req.Frame) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Retrieve local blocks. + blks, err := f.BitmapAttrStore().Blocks() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Read all attributes from all mismatched blocks. + attrs := make(map[uint64]map[string]interface{}) + for _, blockID := range AttrBlocks(blks).Diff(req.Blocks) { + // Retrieve block data. + m, err := f.BitmapAttrStore().BlockData(blockID) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + + // Copy to database-wide struct. + for k, v := range m { + attrs[k] = v + } + } + + // Encode response. + if err := json.NewEncoder(w).Encode(postFrameAttrDiffResponse{ + Attrs: attrs, + }); err != nil { + h.logger().Printf("response encoding error: %s", err) + } +} + +type postFrameAttrDiffRequest struct { + DB string `json:"db"` + Frame string `json:"frame"` + Blocks []AttrBlock `json:"blocks"` +} + +type postFrameAttrDiffResponse struct { + Attrs map[uint64]map[string]interface{} `json:"attrs"` +} + // readProfiles returns a list of profile objects by id. func (h *Handler) readProfiles(db *DB, ids []uint64) ([]*Profile, error) { if db == nil { diff --git a/handler_test.go b/handler_test.go index 89f7148e2..672f51e02 100644 --- a/handler_test.go +++ b/handler_test.go @@ -2,8 +2,10 @@ package pilosa_test import ( "bytes" + "encoding/json" "errors" "io" + "io/ioutil" "net/http" "net/http/httptest" "net/url" @@ -414,6 +416,104 @@ func TestHandler_Query_ErrParse(t *testing.T) { } } +// Ensure the handler can return data in differing blocks for a database. +func TestHandler_DB_AttrStore_Diff(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + s := NewServer() + s.Handler.Index = idx.Index + defer s.Close() + + // Set attributes on the database. + db, err := idx.CreateDBIfNotExists("d") + if err != nil { + t.Fatal(err) + } + if err := db.ProfileAttrStore().SetAttrs(1, map[string]interface{}{"foo": 1, "bar": 2}); err != nil { + t.Fatal(err) + } else if err := db.ProfileAttrStore().SetAttrs(100, map[string]interface{}{"x": "y"}); err != nil { + t.Fatal(err) + } else if err := db.ProfileAttrStore().SetAttrs(200, map[string]interface{}{"snowman": "☃"}); err != nil { + t.Fatal(err) + } + + // Retrieve block checksums. + blks, err := db.ProfileAttrStore().Blocks() + if err != nil { + t.Fatal(err) + } + + // Remove block #0 and alter block 2's checksum. + blks = blks[1:] + blks[1].Checksum = []byte("MISMATCHED_CHECKSUM") + + // Send block checksums to determine diff. + resp, err := http.Post( + s.URL+"/db/attr/diff?db=d", + "application/json", + strings.NewReader(`{"db":"d", "blocks":`+string(MustMarshalJSON(blks))+`}`), + ) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + // Read and validate body. + if body := string(MustReadAll(resp.Body)); body != `{"attrs":{"1":{"bar":2,"foo":1},"200":{"snowman":"☃"}}}`+"\n" { + t.Fatalf("unexpected body: %s", body) + } +} + +// Ensure the handler can return data in differing blocks for a frame. +func TestHandler_Frame_AttrStore_Diff(t *testing.T) { + idx := MustOpenIndex() + defer idx.Close() + + s := NewServer() + s.Handler.Index = idx.Index + defer s.Close() + + // Set attributes on the database. + f, err := idx.CreateFrameIfNotExists("d", "f") + if err != nil { + t.Fatal(err) + } + if err := f.BitmapAttrStore().SetAttrs(1, map[string]interface{}{"foo": 1, "bar": 2}); err != nil { + t.Fatal(err) + } else if err := f.BitmapAttrStore().SetAttrs(100, map[string]interface{}{"x": "y"}); err != nil { + t.Fatal(err) + } else if err := f.BitmapAttrStore().SetAttrs(200, map[string]interface{}{"snowman": "☃"}); err != nil { + t.Fatal(err) + } + + // Retrieve block checksums. + blks, err := f.BitmapAttrStore().Blocks() + if err != nil { + t.Fatal(err) + } + + // Remove block #0 and alter block 2's checksum. + blks = blks[1:] + blks[1].Checksum = []byte("MISMATCHED_CHECKSUM") + + // Send block checksums to determine diff. + resp, err := http.Post( + s.URL+"/frame/attr/diff?db=d", + "application/json", + strings.NewReader(`{"db":"d", "frame":"f", "blocks":`+string(MustMarshalJSON(blks))+`}`), + ) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + // Read and validate body. + if body := string(MustReadAll(resp.Body)); body != `{"attrs":{"1":{"bar":2,"foo":1},"200":{"snowman":"☃"}}}`+"\n" { + t.Fatalf("unexpected body: %s", body) + } +} + // Ensure the handler can backup a fragment and then restore it. func TestHandler_Fragment_BackupRestore(t *testing.T) { idx := MustOpenIndex() @@ -570,3 +670,21 @@ func MustNewHTTPRequest(method, urlStr string, body io.Reader) *http.Request { } return req } + +// MustMarshalJSON marshals v to JSON. Panic on error. +func MustMarshalJSON(v interface{}) []byte { + buf, err := json.Marshal(v) + if err != nil { + panic(err) + } + return buf +} + +// MustReadAll reads a reader into a buffer and returns it. Panic on error. +func MustReadAll(r io.Reader) []byte { + buf, err := ioutil.ReadAll(r) + if err != nil { + panic(err) + } + return buf +} diff --git a/index.go b/index.go index cb9f033cd..d05a507bd 100644 --- a/index.go +++ b/index.go @@ -275,13 +275,43 @@ type IndexSyncer struct { Closing <-chan struct{} } +// Returns true if the syncer has been marked to close. +func (s *IndexSyncer) IsClosing() bool { + select { + case <-s.Closing: + return true + default: + return false + } +} + // SyncIndex compares the index on host with the local index and resolves differences. func (s *IndexSyncer) SyncIndex() error { sliceN := s.Index.SliceN() // Iterate over schema in sorted order. for _, di := range s.Index.Schema() { + // Verify syncer has not closed. + if s.IsClosing() { + return nil + } + + // Sync database profile attributes. + if err := s.syncDatabase(di.Name); err != nil { + return fmt.Errorf("db sync error: db=%s, err=%s", di.Name, err) + } + for _, fi := range di.Frames { + // Verify syncer has not closed. + if s.IsClosing() { + return nil + } + + // Sync frame bitmap attributes. + if err := s.syncFrame(di.Name, fi.Name); err != nil { + return fmt.Errorf("frame sync error: db=%s, frame=%s, err=%s", di.Name, fi.Name, err) + } + for slice := uint64(0); slice <= sliceN; slice++ { // Ignore slices that this host doesn't own. if !s.Cluster.OwnsFragment(s.Host, di.Name, slice) { @@ -289,15 +319,13 @@ func (s *IndexSyncer) SyncIndex() error { } // Verify syncer has not closed. - select { - case <-s.Closing: + if s.IsClosing() { return nil - default: } // Sync fragment if own it. if err := s.syncFragment(di.Name, fi.Name, slice); err != nil { - return fmt.Errorf("sync error: db=%s, frame=%s, slice=%d, err=%s", di.Name, fi.Name, slice, err) + return fmt.Errorf("fragment sync error: db=%s, frame=%s, slice=%d, err=%s", di.Name, fi.Name, slice, err) } } } @@ -306,7 +334,97 @@ func (s *IndexSyncer) SyncIndex() error { return nil } -// syncFragment synchronizes +// syncDatabase synchronizes database attributes with the rest of the cluster. +func (s *IndexSyncer) syncDatabase(db string) error { + // Retrieve database reference. + d := s.Index.DB(db) + if d == nil { + return nil + } + + // Read block checksums. + blks, err := d.ProfileAttrStore().Blocks() + if err != nil { + return err + } + + // Sync with every other host. + for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) { + client, err := NewClient(node.Host) + if err != nil { + return err + } + + // Retrieve attributes from differing blocks. + // Skip update and recomputation if no attributes have changed. + m, err := client.ProfileAttrDiff(db, blks) + if err != nil { + return err + } else if len(m) == 0 { + continue + } + + // Update local copy. + if err := d.ProfileAttrStore().SetBulkAttrs(m); err != nil { + return err + } + + // Recompute blocks. + blks, err = d.ProfileAttrStore().Blocks() + if err != nil { + return err + } + } + + return nil +} + +// syncFrame synchronizes frame attributes with the rest of the cluster. +func (s *IndexSyncer) syncFrame(db, name string) error { + // Retrieve database reference. + f := s.Index.Frame(db, name) + if f == nil { + return nil + } + + // Read block checksums. + blks, err := f.BitmapAttrStore().Blocks() + if err != nil { + return err + } + + // Sync with every other host. + for _, node := range Nodes(s.Cluster.Nodes).FilterHost(s.Host) { + client, err := NewClient(node.Host) + if err != nil { + return err + } + + // Retrieve attributes from differing blocks. + // Skip update and recomputation if no attributes have changed. + m, err := client.BitmapAttrDiff(db, name, blks) + if err != nil { + return err + } else if len(m) == 0 { + continue + } + + // Update local copy. + if err := f.BitmapAttrStore().SetBulkAttrs(m); err != nil { + return err + } + + // Recompute blocks. + blks, err = f.BitmapAttrStore().Blocks() + if err != nil { + return err + } + } + + return nil +} + +// syncFragment synchronizes a fragment with the rest of the cluster. func (s *IndexSyncer) syncFragment(db, frame string, slice uint64) error { // Ensure fragment exists locally. f, err := s.Index.CreateFragmentIfNotExists(db, frame, slice) diff --git a/server.go b/server.go index 5fceb9602..04853a03d 100644 --- a/server.go +++ b/server.go @@ -147,10 +147,10 @@ func (s *Server) Addr() net.Addr { func (s *Server) logger() *log.Logger { return log.New(s.LogOutput, "", log.LstdFlags) } func (s *Server) monitorAntiEntropy() { - ticker := time.NewTicker(time.Duration(s.AntiEntropyInterval)) + ticker := time.NewTicker(s.AntiEntropyInterval) defer ticker.Stop() - s.logger().Printf("index sync monitor initializing") + s.logger().Printf("index sync monitor initializing (%s interval)", s.AntiEntropyInterval) for { // Wait for tick or a close.