Datastore additions to bitmask management

Signed-off-by: Madhu Venugopal <madhu@docker.com>

bitseq/sequence.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/docker/libnetwork/datastore"
 	"github.com/docker/libnetwork/netutils"
 )
 
@@ -21,21 +22,28 @@ const (
 
 // Handle contains the sequece representing the bitmask and its identifier
 type Handle struct {
-	ID   string
-	Head *Sequence
+	App     string
+	ID      string
+	Head    *Sequence
+	store   datastore.DataStore
+	dbIndex uint64
 	sync.Mutex
 }
 
 // NewHandle returns a thread-safe instance of the bitmask handler
-func NewHandle(id string, numElements uint32) *Handle {
-	return &Handle{
-		ID: id,
+func NewHandle(app string, ds datastore.DataStore, id string, numElements uint32) *Handle {
+	h := &Handle{
+		App:   app,
+		ID:    id,
+		store: ds,
 		Head: &Sequence{
 			Block: 0x0,
 			Count: getNumBlocks(numElements),
 			Next:  nil,
 		},
 	}
+	h.watchForChanges()
+	return h
 }
 
 // Sequence reresents a recurring sequence of 32 bits long bitmasks
@@ -151,10 +159,11 @@ func (h *Handle) CheckIfAvailable(ordinal int) (int, int, error) {
 }
 
 // PushReservation pushes the bit reservation inside the bitmask.
-func (h *Handle) PushReservation(bytePos, bitPos int, release bool) {
+func (h *Handle) PushReservation(bytePos, bitPos int, release bool) error {
 	h.Lock()
-	defer h.Unlock()
 	h.Head = PushReservation(bytePos, bitPos, h.Head, release)
+	h.Unlock()
+	return h.writeToStore()
 }
 
 // GetFirstAvailable looks for the first unset bit in passed mask

bitseq/store.go

@@ -0,0 +1,100 @@
+package bitseq
+
+import (
+	"github.com/docker/libnetwork/datastore"
+	"github.com/docker/libnetwork/types"
+)
+
+// Key provides the Key to be used in KV Store
+func (h *Handle) Key() []string {
+	h.Lock()
+	defer h.Unlock()
+	return []string{h.App, h.ID}
+}
+
+// KeyPrefix returns the immediate parent key that can be used for tree walk
+func (h *Handle) KeyPrefix() []string {
+	h.Lock()
+	defer h.Unlock()
+	return []string{h.App}
+}
+
+// Value marshala the data to be stored in the KV store
+func (h *Handle) Value() []byte {
+	h.Lock()
+	defer h.Unlock()
+	head := h.Head
+	if head == nil {
+		return []byte{}
+	}
+	b, err := head.ToByteArray()
+	if err != nil {
+		return []byte{}
+	}
+	return b
+}
+
+// Index returns the latest DB Index as seen by this object
+func (h *Handle) Index() uint64 {
+	h.Lock()
+	defer h.Unlock()
+	return h.dbIndex
+}
+
+// SetIndex method allows the datastore to store the latest DB Index into this object
+func (h *Handle) SetIndex(index uint64) {
+	h.Lock()
+	h.dbIndex = index
+	h.Unlock()
+}
+
+func (h *Handle) watchForChanges() error {
+	h.Lock()
+	store := h.store
+	h.Unlock()
+
+	if store == nil {
+		return nil
+	}
+
+	kvpChan, err := store.KVStore().Watch(datastore.Key(h.Key()...), nil)
+	if err != nil {
+		return err
+	}
+	go func() {
+		for {
+			select {
+			case kvPair := <-kvpChan:
+				h.Lock()
+				h.dbIndex = kvPair.LastIndex
+				h.Head.FromByteArray(kvPair.Value)
+				h.Unlock()
+			}
+		}
+	}()
+	return nil
+}
+
+func (h *Handle) writeToStore() error {
+	h.Lock()
+	store := h.store
+	h.Unlock()
+	if store == nil {
+		return nil
+	}
+	err := store.PutObjectAtomic(h)
+	if err == datastore.ErrKeyModified {
+		return types.RetryErrorf("failed to perform atomic write (%v). retry might fix the error", err)
+	}
+	return err
+}
+
+func (h *Handle) deleteFromStore() error {
+	h.Lock()
+	store := h.store
+	h.Unlock()
+	if store == nil {
+		return nil
+	}
+	return store.DeleteObjectAtomic(h)
+}