diff --git a/bitseq/sequence.go b/bitseq/sequence.go new file mode 100644 index 0000000..6a3dc9f --- /dev/null +++ b/bitseq/sequence.go @@ -0,0 +1,452 @@ +// Package bitseq provides a structure and utilities for representing long bitmask +// as sequence of run-lenght encoded blocks. It operates direclty on the encoded +// representation, it does not decode/encode. +package bitseq + +import ( + "fmt" + "sync" + + "github.com/docker/libkv/store" + "github.com/docker/libnetwork/datastore" + "github.com/docker/libnetwork/netutils" +) + +// Block Sequence constants +// If needed we can think of making these configurable +const ( + blockLen = 32 + blockBytes = blockLen / 8 + blockMAX = 1<%s", s.Block, s.Count, nextBlock) +} + +// GetAvailableBit returns the position of the first unset bit in the bitmask represented by this sequence +func (s *Sequence) GetAvailableBit() (bytePos, bitPos int) { + if s.Block == blockMAX || s.Count == 0 { + return -1, -1 + } + bits := 0 + bitSel := uint32(blockFirstBit) + for bitSel > 0 && s.Block&bitSel != 0 { + bitSel >>= 1 + bits++ + } + return bits / 8, bits % 8 +} + +// GetCopy returns a copy of the linked list rooted at this node +func (s *Sequence) GetCopy() *Sequence { + n := &Sequence{Block: s.Block, Count: s.Count} + pn := n + ps := s.Next + for ps != nil { + pn.Next = &Sequence{Block: ps.Block, Count: ps.Count} + pn = pn.Next + ps = ps.Next + } + return n +} + +// Equal checks if this sequence is equal to the passed one +func (s *Sequence) Equal(o *Sequence) bool { + this := s + other := o + for this != nil { + if other == nil { + return false + } + if this.Block != other.Block || this.Count != other.Count { + return false + } + this = this.Next + other = other.Next + } + // Check if other is longer than this + if other != nil { + return false + } + return true +} + +// ToByteArray converts the sequence into a byte array +// TODO (aboch): manage network/host order stuff +func (s *Sequence) ToByteArray() ([]byte, error) { + var bb []byte + + p := s + for p != nil { + bb = append(bb, netutils.U32ToA(p.Block)...) + bb = append(bb, netutils.U32ToA(p.Count)...) + p = p.Next + } + + return bb, nil +} + +// FromByteArray construct the sequence from the byte array +// TODO (aboch): manage network/host order stuff +func (s *Sequence) FromByteArray(data []byte) error { + l := len(data) + if l%8 != 0 { + return fmt.Errorf("cannot deserialize byte sequence of lenght %d", l) + } + + p := s + i := 0 + for { + p.Block = netutils.ATo32(data[i : i+4]) + p.Count = netutils.ATo32(data[i+4 : i+8]) + i += 8 + if i == l { + break + } + p.Next = &Sequence{} + p = p.Next + } + + return nil +} + +// GetFirstAvailable returns the byte and bit position of the first unset bit +func (h *Handle) GetFirstAvailable() (int, int, error) { + h.Lock() + defer h.Unlock() + return GetFirstAvailable(h.head) +} + +// CheckIfAvailable checks if the bit correspondent to the specified ordinal is unset +// If the ordinal is beyond the Sequence limits, a negative response is returned +func (h *Handle) CheckIfAvailable(ordinal int) (int, int, error) { + h.Lock() + defer h.Unlock() + return CheckIfAvailable(h.head, ordinal) +} + +// PushReservation pushes the bit reservation inside the bitmask. +func (h *Handle) PushReservation(bytePos, bitPos int, release bool) error { + // Create a copy of the current handler + h.Lock() + nh := &Handle{app: h.app, id: h.id, store: h.store, dbIndex: h.dbIndex, head: h.head.GetCopy()} + h.Unlock() + + nh.head = PushReservation(bytePos, bitPos, nh.head, release) + + err := nh.writeToStore() + if err == nil { + // Commit went through, save locally + h.Lock() + h.head = nh.head + if release { + h.unselected++ + } else { + h.unselected-- + } + h.Unlock() + } + + return err +} + +// Destroy removes from the datastore the data belonging to this handle +func (h *Handle) Destroy() { + h.deleteFromStore() +} + +// ToByteArray converts this handle's data into a byte array +func (h *Handle) ToByteArray() ([]byte, error) { + ba := make([]byte, 8) + + h.Lock() + defer h.Unlock() + copy(ba[0:4], netutils.U32ToA(h.bits)) + copy(ba[4:8], netutils.U32ToA(h.unselected)) + bm, err := h.head.ToByteArray() + if err != nil { + return nil, err + } + ba = append(ba, bm...) + + return ba, nil +} + +// FromByteArray reads his handle's data from a byte array +func (h *Handle) FromByteArray(ba []byte) error { + nh := &Sequence{} + err := nh.FromByteArray(ba[8:]) + if err != nil { + return err + } + + h.Lock() + h.head = nh + h.bits = netutils.ATo32(ba[0:4]) + h.unselected = netutils.ATo32(ba[4:8]) + h.Unlock() + + return nil +} + +// Bits returns the length of the bit sequence +func (h *Handle) Bits() uint32 { + return h.bits +} + +// Unselected returns the number of bits which are not selected +func (h *Handle) Unselected() uint32 { + h.Lock() + defer h.Unlock() + return h.unselected +} + +func (h *Handle) getDBIndex() uint64 { + h.Lock() + defer h.Unlock() + return h.dbIndex +} + +// GetFirstAvailable looks for the first unset bit in passed mask +func GetFirstAvailable(head *Sequence) (int, int, error) { + byteIndex := 0 + current := head + for current != nil { + if current.Block != blockMAX { + bytePos, bitPos := current.GetAvailableBit() + return byteIndex + bytePos, bitPos, nil + } + byteIndex += int(current.Count * blockBytes) + current = current.Next + } + return -1, -1, fmt.Errorf("no bit available") +} + +// CheckIfAvailable checks if the bit correspondent to the specified ordinal is unset +// If the ordinal is beyond the Sequence limits, a negative response is returned +func CheckIfAvailable(head *Sequence, ordinal int) (int, int, error) { + bytePos := ordinal / 8 + bitPos := ordinal % 8 + + // Find the Sequence containing this byte + current, _, _, inBlockBytePos := findSequence(head, bytePos) + + if current != nil { + // Check whether the bit corresponding to the ordinal address is unset + bitSel := uint32(blockFirstBit >> uint(inBlockBytePos*8+bitPos)) + if current.Block&bitSel == 0 { + return bytePos, bitPos, nil + } + } + + return -1, -1, fmt.Errorf("requested bit is not available") +} + +// Given the byte position and the sequences list head, return the pointer to the +// sequence containing the byte (current), the pointer to the previous sequence, +// the number of blocks preceding the block containing the byte inside the current sequence. +// If bytePos is outside of the list, function will return (nil, nil, 0, -1) +func findSequence(head *Sequence, bytePos int) (*Sequence, *Sequence, uint32, int) { + // Find the Sequence containing this byte + previous := head + current := head + n := bytePos + for current.Next != nil && n >= int(current.Count*blockBytes) { // Nil check for less than 32 addresses masks + n -= int(current.Count * blockBytes) + previous = current + current = current.Next + } + + // If byte is outside of the list, let caller know + if n >= int(current.Count*blockBytes) { + return nil, nil, 0, -1 + } + + // Find the byte position inside the block and the number of blocks + // preceding the block containing the byte inside this sequence + precBlocks := uint32(n / blockBytes) + inBlockBytePos := bytePos % blockBytes + + return current, previous, precBlocks, inBlockBytePos +} + +// PushReservation pushes the bit reservation inside the bitmask. +// Given byte and bit positions, identify the sequence (current) which holds the block containing the affected bit. +// Create a new block with the modified bit according to the operation (allocate/release). +// Create a new Sequence containing the new Block and insert it in the proper position. +// Remove current sequence if empty. +// Check if new Sequence can be merged with neighbour (previous/Next) sequences. +// +// +// Identify "current" Sequence containing block: +// [prev seq] [current seq] [Next seq] +// +// Based on block position, resulting list of sequences can be any of three forms: +// +// Block position Resulting list of sequences +// A) Block is first in current: [prev seq] [new] [modified current seq] [Next seq] +// B) Block is last in current: [prev seq] [modified current seq] [new] [Next seq] +// C) Block is in the middle of current: [prev seq] [curr pre] [new] [curr post] [Next seq] +func PushReservation(bytePos, bitPos int, head *Sequence, release bool) *Sequence { + // Store list's head + newHead := head + + // Find the Sequence containing this byte + current, previous, precBlocks, inBlockBytePos := findSequence(head, bytePos) + if current == nil { + return newHead + } + + // Construct updated block + bitSel := uint32(blockFirstBit >> uint(inBlockBytePos*8+bitPos)) + newBlock := current.Block + if release { + newBlock &^= bitSel + } else { + newBlock |= bitSel + } + + // Quit if it was a redundant request + if current.Block == newBlock { + return newHead + } + + // Current Sequence inevitably looses one block, upadate Count + current.Count-- + + // Create new sequence + newSequence := &Sequence{Block: newBlock, Count: 1} + + // Insert the new sequence in the list based on block position + if precBlocks == 0 { // First in sequence (A) + newSequence.Next = current + if current == head { + newHead = newSequence + previous = newHead + } else { + previous.Next = newSequence + } + removeCurrentIfEmpty(&newHead, newSequence, current) + mergeSequences(previous) + } else if precBlocks == current.Count-2 { // Last in sequence (B) + newSequence.Next = current.Next + current.Next = newSequence + mergeSequences(current) + } else { // In between the sequence (C) + currPre := &Sequence{Block: current.Block, Count: precBlocks, Next: newSequence} + currPost := current + currPost.Count -= precBlocks + newSequence.Next = currPost + if currPost == head { + newHead = currPre + } else { + previous.Next = currPre + } + // No merging or empty current possible here + } + + return newHead +} + +// Removes the current sequence from the list if empty, adjusting the head pointer if needed +func removeCurrentIfEmpty(head **Sequence, previous, current *Sequence) { + if current.Count == 0 { + if current == *head { + *head = current.Next + } else { + previous.Next = current.Next + current = current.Next + } + } +} + +// Given a pointer to a Sequence, it checks if it can be merged with any following sequences +// It stops when no more merging is possible. +// TODO: Optimization: only attempt merge from start to end sequence, no need to scan till the end of the list +func mergeSequences(seq *Sequence) { + if seq != nil { + // Merge all what possible from seq + for seq.Next != nil && seq.Block == seq.Next.Block { + seq.Count += seq.Next.Count + seq.Next = seq.Next.Next + } + // Move to Next + mergeSequences(seq.Next) + } +} + +func getNumBlocks(numBits uint32) uint32 { + numBlocks := numBits / blockLen + if numBits%blockLen != 0 { + numBlocks++ + } + return numBlocks +} diff --git a/bitseq/sequence_test.go b/bitseq/sequence_test.go new file mode 100644 index 0000000..54e505f --- /dev/null +++ b/bitseq/sequence_test.go @@ -0,0 +1,427 @@ +package bitseq + +import ( + "testing" +) + +func TestSequenceGetAvailableBit(t *testing.T) { + input := []struct { + head *Sequence + bytePos int + bitPos int + }{ + {&Sequence{Block: 0x0, Count: 0}, -1, -1}, + {&Sequence{Block: 0x0, Count: 1}, 0, 0}, + {&Sequence{Block: 0x0, Count: 100}, 0, 0}, + + {&Sequence{Block: 0x80000000, Count: 0}, -1, -1}, + {&Sequence{Block: 0x80000000, Count: 1}, 0, 1}, + {&Sequence{Block: 0x80000000, Count: 100}, 0, 1}, + + {&Sequence{Block: 0xFF000000, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFF000000, Count: 1}, 1, 0}, + {&Sequence{Block: 0xFF000000, Count: 100}, 1, 0}, + + {&Sequence{Block: 0xFF800000, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFF800000, Count: 1}, 1, 1}, + {&Sequence{Block: 0xFF800000, Count: 100}, 1, 1}, + + {&Sequence{Block: 0xFFC0FF00, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFC0FF00, Count: 1}, 1, 2}, + {&Sequence{Block: 0xFFC0FF00, Count: 100}, 1, 2}, + + {&Sequence{Block: 0xFFE0FF00, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFE0FF00, Count: 1}, 1, 3}, + {&Sequence{Block: 0xFFE0FF00, Count: 100}, 1, 3}, + + {&Sequence{Block: 0xFFFEFF00, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFFEFF00, Count: 1}, 1, 7}, + {&Sequence{Block: 0xFFFEFF00, Count: 100}, 1, 7}, + + {&Sequence{Block: 0xFFFFC0FF, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFFFC0FF, Count: 1}, 2, 2}, + {&Sequence{Block: 0xFFFFC0FF, Count: 100}, 2, 2}, + + {&Sequence{Block: 0xFFFFFF00, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFFFFF00, Count: 1}, 3, 0}, + {&Sequence{Block: 0xFFFFFF00, Count: 100}, 3, 0}, + + {&Sequence{Block: 0xFFFFFFFE, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFFFFFFE, Count: 1}, 3, 7}, + {&Sequence{Block: 0xFFFFFFFE, Count: 100}, 3, 7}, + + {&Sequence{Block: 0xFFFFFFFF, Count: 0}, -1, -1}, + {&Sequence{Block: 0xFFFFFFFF, Count: 1}, -1, -1}, + {&Sequence{Block: 0xFFFFFFFF, Count: 100}, -1, -1}, + } + + for n, i := range input { + b, bb := i.head.GetAvailableBit() + if b != i.bytePos || bb != i.bitPos { + t.Fatalf("Error in Sequence.getAvailableBit() (%d).\nExp: (%d, %d)\nGot: (%d, %d),", n, i.bytePos, i.bitPos, b, bb) + } + } +} + +func TestSequenceEqual(t *testing.T) { + input := []struct { + first *Sequence + second *Sequence + areEqual bool + }{ + {&Sequence{Block: 0x0, Count: 8, Next: nil}, &Sequence{Block: 0x0, Count: 8}, true}, + {&Sequence{Block: 0x0, Count: 0, Next: nil}, &Sequence{Block: 0x0, Count: 0}, true}, + {&Sequence{Block: 0x0, Count: 2, Next: nil}, &Sequence{Block: 0x0, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}, false}, + {&Sequence{Block: 0x0, Count: 2, Next: &Sequence{Block: 0x1, Count: 1}}, &Sequence{Block: 0x0, Count: 2}, false}, + + {&Sequence{Block: 0x12345678, Count: 8, Next: nil}, &Sequence{Block: 0x12345678, Count: 8}, true}, + {&Sequence{Block: 0x12345678, Count: 8, Next: nil}, &Sequence{Block: 0x12345678, Count: 9}, false}, + {&Sequence{Block: 0x12345678, Count: 1, Next: &Sequence{Block: 0XFFFFFFFF, Count: 1}}, &Sequence{Block: 0x12345678, Count: 1}, false}, + {&Sequence{Block: 0x12345678, Count: 1}, &Sequence{Block: 0x12345678, Count: 1, Next: &Sequence{Block: 0XFFFFFFFF, Count: 1}}, false}, + } + + for n, i := range input { + if i.areEqual != i.first.Equal(i.second) { + t.Fatalf("Error in Sequence.Equal() (%d).\nExp: %t\nGot: %t,", n, i.areEqual, !i.areEqual) + } + } +} + +func TestSequenceCopy(t *testing.T) { + s := &Sequence{ + Block: 0x0, + Count: 8, + Next: &Sequence{ + Block: 0x0, + Count: 8, + Next: &Sequence{ + Block: 0x0, + Count: 0, + Next: &Sequence{ + Block: 0x0, + Count: 0, + Next: &Sequence{ + Block: 0x0, + Count: 2, + Next: &Sequence{ + Block: 0x0, + Count: 1, + Next: &Sequence{ + Block: 0x0, + Count: 1, + Next: &Sequence{ + Block: 0x0, + Count: 2, + Next: &Sequence{ + Block: 0x1, + Count: 1, + Next: &Sequence{ + Block: 0x0, + Count: 2, + Next: nil, + }, + }, + }, + }, + }, + }, + }, + }, + }, + } + + n := s.GetCopy() + if !s.Equal(n) { + t.Fatalf("copy of s failed") + } + if n == s { + t.Fatalf("not true copy of s") + } +} + +func TestGetFirstAvailable(t *testing.T) { + input := []struct { + mask *Sequence + bytePos int + bitPos int + }{ + {&Sequence{Block: 0xffffffff, Count: 2048}, -1, -1}, + {&Sequence{Block: 0x0, Count: 8}, 0, 0}, + {&Sequence{Block: 0x80000000, Count: 8}, 0, 1}, + {&Sequence{Block: 0xC0000000, Count: 8}, 0, 2}, + {&Sequence{Block: 0xE0000000, Count: 8}, 0, 3}, + {&Sequence{Block: 0xF0000000, Count: 8}, 0, 4}, + {&Sequence{Block: 0xF8000000, Count: 8}, 0, 5}, + {&Sequence{Block: 0xFC000000, Count: 8}, 0, 6}, + {&Sequence{Block: 0xFE000000, Count: 8}, 0, 7}, + + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x00000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 0}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 2}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xE0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 3}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xF0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 4}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xF8000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 5}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFC000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 6}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFE000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 7}, + + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFF000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 0}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFF800000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFC00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 2}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFE00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 3}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFF00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 4}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFF80000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 5}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFC0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 6}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFE0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 7}, + + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xfffffffe, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 7, 7}, + + {&Sequence{Block: 0xffffffff, Count: 2, Next: &Sequence{Block: 0x0, Count: 6}}, 8, 0}, + } + + for n, i := range input { + bytePos, bitPos, _ := GetFirstAvailable(i.mask) + if bytePos != i.bytePos || bitPos != i.bitPos { + t.Fatalf("Error in (%d) getFirstAvailable(). Expected (%d, %d). Got (%d, %d)", n, i.bytePos, i.bitPos, bytePos, bitPos) + } + } +} + +func TestFindSequence(t *testing.T) { + input := []struct { + head *Sequence + bytePos int + precBlocks uint32 + inBlockBytePos int + }{ + {&Sequence{Block: 0xffffffff, Count: 0}, 0, 0, -1}, + {&Sequence{Block: 0xffffffff, Count: 0}, 31, 0, -1}, + {&Sequence{Block: 0xffffffff, Count: 0}, 100, 0, -1}, + + {&Sequence{Block: 0x0, Count: 1}, 0, 0, 0}, + {&Sequence{Block: 0x0, Count: 1}, 1, 0, 1}, + {&Sequence{Block: 0x0, Count: 1}, 31, 0, -1}, + {&Sequence{Block: 0x0, Count: 1}, 60, 0, -1}, + + {&Sequence{Block: 0xffffffff, Count: 10}, 0, 0, 0}, + {&Sequence{Block: 0xffffffff, Count: 10}, 3, 0, 3}, + {&Sequence{Block: 0xffffffff, Count: 10}, 4, 1, 0}, + {&Sequence{Block: 0xffffffff, Count: 10}, 7, 1, 3}, + {&Sequence{Block: 0xffffffff, Count: 10}, 8, 2, 0}, + {&Sequence{Block: 0xffffffff, Count: 10}, 39, 9, 3}, + + {&Sequence{Block: 0xffffffff, Count: 10, Next: &Sequence{Block: 0xcc000000, Count: 10}}, 79, 9, 3}, + {&Sequence{Block: 0xffffffff, Count: 10, Next: &Sequence{Block: 0xcc000000, Count: 10}}, 80, 0, -1}, + } + + for n, i := range input { + _, _, precBlocks, inBlockBytePos := findSequence(i.head, i.bytePos) + if precBlocks != i.precBlocks || inBlockBytePos != i.inBlockBytePos { + t.Fatalf("Error in (%d) findSequence(). Expected (%d, %d). Got (%d, %d)", n, i.precBlocks, i.inBlockBytePos, precBlocks, inBlockBytePos) + } + } +} + +func TestCheckIfAvailable(t *testing.T) { + input := []struct { + head *Sequence + ordinal int + bytePos int + bitPos int + }{ + {&Sequence{Block: 0xffffffff, Count: 0}, 0, -1, -1}, + {&Sequence{Block: 0xffffffff, Count: 0}, 31, -1, -1}, + {&Sequence{Block: 0xffffffff, Count: 0}, 100, -1, -1}, + + {&Sequence{Block: 0x0, Count: 1}, 0, 0, 0}, + {&Sequence{Block: 0x0, Count: 1}, 1, 0, 1}, + {&Sequence{Block: 0x0, Count: 1}, 31, 3, 7}, + {&Sequence{Block: 0x0, Count: 1}, 60, -1, -1}, + + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x800000ff, Count: 1}}, 31, -1, -1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x800000ff, Count: 1}}, 32, -1, -1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x800000ff, Count: 1}}, 33, 4, 1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1}}, 33, -1, -1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1}}, 34, 4, 2}, + + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}}, 55, 6, 7}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}}, 56, -1, -1}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}}, 63, -1, -1}, + + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}}, 64, 8, 0}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}}, 95, 11, 7}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC00000ff, Count: 1, Next: &Sequence{Block: 0x0, Count: 1}}}, 96, -1, -1}, + } + + for n, i := range input { + bytePos, bitPos, _ := CheckIfAvailable(i.head, i.ordinal) + if bytePos != i.bytePos || bitPos != i.bitPos { + t.Fatalf("Error in (%d) checkIfAvailable(ord:%d). Expected (%d, %d). Got (%d, %d)", n, i.ordinal, i.bytePos, i.bitPos, bytePos, bitPos) + } + } +} + +func TestMergeSequences(t *testing.T) { + input := []struct { + original *Sequence + merged *Sequence + }{ + {&Sequence{Block: 0xFE000000, Count: 8, Next: &Sequence{Block: 0xFE000000, Count: 2}}, &Sequence{Block: 0xFE000000, Count: 10}}, + {&Sequence{Block: 0xFFFFFFFF, Count: 8, Next: &Sequence{Block: 0xFFFFFFFF, Count: 1}}, &Sequence{Block: 0xFFFFFFFF, Count: 9}}, + {&Sequence{Block: 0xFFFFFFFF, Count: 1, Next: &Sequence{Block: 0xFFFFFFFF, Count: 8}}, &Sequence{Block: 0xFFFFFFFF, Count: 9}}, + + {&Sequence{Block: 0xFFFFFFF0, Count: 8, Next: &Sequence{Block: 0xFFFFFFF0, Count: 1}}, &Sequence{Block: 0xFFFFFFF0, Count: 9}}, + {&Sequence{Block: 0xFFFFFFF0, Count: 1, Next: &Sequence{Block: 0xFFFFFFF0, Count: 8}}, &Sequence{Block: 0xFFFFFFF0, Count: 9}}, + + {&Sequence{Block: 0xFE, Count: 8, Next: &Sequence{Block: 0xFE, Count: 1, Next: &Sequence{Block: 0xFE, Count: 5}}}, &Sequence{Block: 0xFE, Count: 14}}, + {&Sequence{Block: 0xFE, Count: 8, Next: &Sequence{Block: 0xFE, Count: 1, Next: &Sequence{Block: 0xFE, Count: 5, Next: &Sequence{Block: 0xFF, Count: 1}}}}, + &Sequence{Block: 0xFE, Count: 14, Next: &Sequence{Block: 0xFF, Count: 1}}}, + + // No merge + {&Sequence{Block: 0xFE, Count: 8, Next: &Sequence{Block: 0xF8, Count: 1, Next: &Sequence{Block: 0xFE, Count: 5}}}, + &Sequence{Block: 0xFE, Count: 8, Next: &Sequence{Block: 0xF8, Count: 1, Next: &Sequence{Block: 0xFE, Count: 5}}}}, + + // No merge from head: // Merge function tries to merge from passed head. If it can't merge with Next, it does not reattempt with Next as head + {&Sequence{Block: 0xFE, Count: 8, Next: &Sequence{Block: 0xFF, Count: 1, Next: &Sequence{Block: 0xFF, Count: 5}}}, + &Sequence{Block: 0xFE, Count: 8, Next: &Sequence{Block: 0xFF, Count: 6}}}, + } + + for n, i := range input { + mergeSequences(i.original) + for !i.merged.Equal(i.original) { + t.Fatalf("Error in (%d) mergeSequences().\nExp: %s\nGot: %s,", n, i.merged, i.original) + } + } +} + +func TestPushReservation(t *testing.T) { + input := []struct { + mask *Sequence + bytePos int + bitPos int + newMask *Sequence + }{ + // Create first Sequence and fill in 8 addresses starting from address 0 + {&Sequence{Block: 0x0, Count: 8, Next: nil}, 0, 0, &Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0x0, Count: 7, Next: nil}}}, + {&Sequence{Block: 0x80000000, Count: 8}, 0, 1, &Sequence{Block: 0xC0000000, Count: 1, Next: &Sequence{Block: 0x80000000, Count: 7, Next: nil}}}, + {&Sequence{Block: 0xC0000000, Count: 8}, 0, 2, &Sequence{Block: 0xE0000000, Count: 1, Next: &Sequence{Block: 0xC0000000, Count: 7, Next: nil}}}, + {&Sequence{Block: 0xE0000000, Count: 8}, 0, 3, &Sequence{Block: 0xF0000000, Count: 1, Next: &Sequence{Block: 0xE0000000, Count: 7, Next: nil}}}, + {&Sequence{Block: 0xF0000000, Count: 8}, 0, 4, &Sequence{Block: 0xF8000000, Count: 1, Next: &Sequence{Block: 0xF0000000, Count: 7, Next: nil}}}, + {&Sequence{Block: 0xF8000000, Count: 8}, 0, 5, &Sequence{Block: 0xFC000000, Count: 1, Next: &Sequence{Block: 0xF8000000, Count: 7, Next: nil}}}, + {&Sequence{Block: 0xFC000000, Count: 8}, 0, 6, &Sequence{Block: 0xFE000000, Count: 1, Next: &Sequence{Block: 0xFC000000, Count: 7, Next: nil}}}, + {&Sequence{Block: 0xFE000000, Count: 8}, 0, 7, &Sequence{Block: 0xFF000000, Count: 1, Next: &Sequence{Block: 0xFE000000, Count: 7, Next: nil}}}, + + {&Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0x0, Count: 7}}, 0, 1, &Sequence{Block: 0xC0000000, Count: 1, Next: &Sequence{Block: 0x0, Count: 7, Next: nil}}}, + + // Create second Sequence and fill in 8 addresses starting from address 32 + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x00000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6, Next: nil}}}, 4, 0, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 1, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xC0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 2, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xE0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xE0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 3, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xF0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xF0000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 4, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xF8000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xF8000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 5, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFC000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFC000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 6, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFE000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFE000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 4, 7, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFF000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + // fill in 8 addresses starting from address 40 + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFF000000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 0, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFF800000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFF800000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 1, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFC00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFC00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 2, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFE00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFE00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 3, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFF00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFF00000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 4, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFF80000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFF80000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 5, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFC0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFC0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 6, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFE0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFE0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}, 5, 7, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xFFFF0000, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 6}}}}, + + // Insert new Sequence + {&Sequence{Block: 0xffffffff, Count: 2, Next: &Sequence{Block: 0x0, Count: 6}}, 8, 0, + &Sequence{Block: 0xffffffff, Count: 2, Next: &Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0x0, Count: 5}}}}, + {&Sequence{Block: 0xffffffff, Count: 2, Next: &Sequence{Block: 0x80000000, Count: 1, Next: &Sequence{Block: 0x0, Count: 5}}}, 8, 1, + &Sequence{Block: 0xffffffff, Count: 2, Next: &Sequence{Block: 0xC0000000, Count: 1, Next: &Sequence{Block: 0x0, Count: 5}}}}, + + // Merge affected with Next + {&Sequence{Block: 0xffffffff, Count: 7, Next: &Sequence{Block: 0xfffffffe, Count: 2, Next: &Sequence{Block: 0xffffffff, Count: 1}}}, 31, 7, + &Sequence{Block: 0xffffffff, Count: 8, Next: &Sequence{Block: 0xfffffffe, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 1}}}}, + {&Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xfffffffc, Count: 1, Next: &Sequence{Block: 0xfffffffe, Count: 6}}}, 7, 6, + &Sequence{Block: 0xffffffff, Count: 1, Next: &Sequence{Block: 0xfffffffe, Count: 7}}}, + + // Merge affected with Next and Next.Next + {&Sequence{Block: 0xffffffff, Count: 7, Next: &Sequence{Block: 0xfffffffe, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 1}}}, 31, 7, + &Sequence{Block: 0xffffffff, Count: 9}}, + {&Sequence{Block: 0xffffffff, Count: 7, Next: &Sequence{Block: 0xfffffffe, Count: 1}}, 31, 7, + &Sequence{Block: 0xffffffff, Count: 8}}, + + // Merge affected with previous and Next + {&Sequence{Block: 0xffffffff, Count: 7, Next: &Sequence{Block: 0xfffffffe, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 1}}}, 31, 7, + &Sequence{Block: 0xffffffff, Count: 9}}, + + // Redundant push: No change + {&Sequence{Block: 0xffff0000, Count: 1}, 0, 0, &Sequence{Block: 0xffff0000, Count: 1}}, + {&Sequence{Block: 0xffff0000, Count: 7}, 25, 7, &Sequence{Block: 0xffff0000, Count: 7}}, + {&Sequence{Block: 0xffffffff, Count: 7, Next: &Sequence{Block: 0xfffffffe, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 1}}}, 7, 7, + &Sequence{Block: 0xffffffff, Count: 7, Next: &Sequence{Block: 0xfffffffe, Count: 1, Next: &Sequence{Block: 0xffffffff, Count: 1}}}}, + } + + for n, i := range input { + mask := PushReservation(i.bytePos, i.bitPos, i.mask, false) + if !mask.Equal(i.newMask) { + t.Fatalf("Error in (%d) pushReservation():\n%s + (%d,%d):\nExp: %s\nGot: %s,", n, i.mask, i.bytePos, i.bitPos, i.newMask, mask) + } + } +} + +func TestSerializeDeserialize(t *testing.T) { + s := &Sequence{ + Block: 0xffffffff, + Count: 1, + Next: &Sequence{ + Block: 0xFF000000, + Count: 1, + Next: &Sequence{ + Block: 0xffffffff, + Count: 6, + Next: &Sequence{ + Block: 0xffffffff, + Count: 1, + Next: &Sequence{ + Block: 0xFF800000, + Count: 1, + Next: &Sequence{ + Block: 0xffffffff, + Count: 6, + }, + }, + }, + }, + }, + } + + data, err := s.ToByteArray() + if err != nil { + t.Fatal(err) + } + + r := &Sequence{} + err = r.FromByteArray(data) + if err != nil { + t.Fatal(err) + } + + if !s.Equal(r) { + t.Fatalf("Sequences are different: \n%v\n%v", s, r) + } +} diff --git a/bitseq/store.go b/bitseq/store.go new file mode 100644 index 0000000..bbbe4c2 --- /dev/null +++ b/bitseq/store.go @@ -0,0 +1,97 @@ +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 marshals the data to be stored in the KV store +func (h *Handle) Value() []byte { + b, err := h.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: + // Only process remote update + if kvPair != nil && (kvPair.LastIndex != h.getDBIndex()) { + h.Lock() + h.dbIndex = kvPair.LastIndex + h.Unlock() + h.FromByteArray(kvPair.Value) + } + } + } + }() + 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) +} diff --git a/datastore/datastore.go b/datastore/datastore.go index 9f8d500..85aecc9 100644 --- a/datastore/datastore.go +++ b/datastore/datastore.go @@ -15,6 +15,8 @@ import ( type DataStore interface { // GetObject gets data from datastore and unmarshals to the specified object GetObject(key string, o interface{}) error + // GetUpdatedObject gets data from datastore along with its index and unmarshals to the specified object + GetUpdatedObject(key string, o interface{}) (uint64, error) // PutObject adds a new Record based on an object into the datastore PutObject(kvObject KV) error // PutObjectAtomic provides an atomic add and update operation for a Record @@ -152,6 +154,18 @@ func (ds *datastore) GetObject(key string, o interface{}) error { return json.Unmarshal(kvPair.Value, o) } +// GetUpdateObject returns a record matching the key +func (ds *datastore) GetUpdatedObject(key string, o interface{}) (uint64, error) { + kvPair, err := ds.store.Get(key) + if err != nil { + return 0, err + } + if err := json.Unmarshal(kvPair.Value, o); err != nil { + return 0, err + } + return kvPair.LastIndex, nil +} + // DeleteObject unconditionally deletes a record from the store func (ds *datastore) DeleteObject(kvObject KV) error { return ds.store.Delete(Key(kvObject.Key()...)) diff --git a/idm/idm.go b/idm/idm.go new file mode 100644 index 0000000..0057091 --- /dev/null +++ b/idm/idm.go @@ -0,0 +1,94 @@ +// Package idm manages resevation/release of numerical ids from a configured set of contiguos ids +package idm + +import ( + "fmt" + + "github.com/docker/libnetwork/bitseq" + "github.com/docker/libnetwork/datastore" + "github.com/docker/libnetwork/types" +) + +// Idm manages the reservation/release of numerical ids from a contiguos set +type Idm struct { + start uint32 + end uint32 + handle *bitseq.Handle +} + +// New returns an instance of id manager for a set of [start-end] numerical ids +func New(ds datastore.DataStore, id string, start, end uint32) (*Idm, error) { + if id == "" { + return nil, fmt.Errorf("Invalid id") + } + if end <= start { + return nil, fmt.Errorf("Invalid set range: [%d, %d]", start, end) + } + + h, err := bitseq.NewHandle("idm", ds, id, uint32(1+end-start)) + if err != nil { + return nil, err + } + + return &Idm{start: start, end: end, handle: h}, nil +} + +// GetID returns the first available id in the set +func (i *Idm) GetID() (uint32, error) { + if i.handle == nil { + return 0, fmt.Errorf("ID set is not initialized") + } + + for { + bytePos, bitPos, err := i.handle.GetFirstAvailable() + if err != nil { + return 0, fmt.Errorf("no available ids") + } + id := i.start + uint32(bitPos+bytePos*8) + + // for sets which length is non multiple of 32 this check is needed + if i.end < id { + return 0, fmt.Errorf("no available ids") + } + + if err := i.handle.PushReservation(bytePos, bitPos, false); err != nil { + if _, ok := err.(types.RetryError); !ok { + return 0, fmt.Errorf("internal failure while reserving the id: %s", err.Error()) + } + continue + } + + return id, nil + } +} + +// GetSpecificID tries to reserve the specified id +func (i *Idm) GetSpecificID(id uint32) error { + if i.handle == nil { + return fmt.Errorf("ID set is not initialized") + } + + if id < i.start || id > i.end { + return fmt.Errorf("Requested id does not belong to the set") + } + + for { + bytePos, bitPos, err := i.handle.CheckIfAvailable(int(id - i.start)) + if err != nil { + return fmt.Errorf("requested id is not available") + } + if err := i.handle.PushReservation(bytePos, bitPos, false); err != nil { + if _, ok := err.(types.RetryError); !ok { + return fmt.Errorf("internal failure while reserving the id: %s", err.Error()) + } + continue + } + return nil + } +} + +// Release releases the specified id +func (i *Idm) Release(id uint32) { + ordinal := id - i.start + i.handle.PushReservation(int(ordinal/8), int(ordinal%8), true) +} diff --git a/idm/idm_test.go b/idm/idm_test.go new file mode 100644 index 0000000..1004cb8 --- /dev/null +++ b/idm/idm_test.go @@ -0,0 +1,108 @@ +package idm + +import ( + "testing" +) + +func TestNew(t *testing.T) { + _, err := New(nil, "", 0, 1) + if err == nil { + t.Fatalf("Expected failure, but succeeded") + } + + _, err = New(nil, "myset", 1<<10, 0) + if err == nil { + t.Fatalf("Expected failure, but succeeded") + } + + i, err := New(nil, "myset", 0, 10) + if err != nil { + t.Fatalf("Unexpected failure: %v", err) + } + if i.handle == nil { + t.Fatalf("set is not initialized") + } + if i.start != 0 { + t.Fatalf("unexpected start") + } + if i.end != 10 { + t.Fatalf("unexpected end") + } +} + +func TestAllocate(t *testing.T) { + i, err := New(nil, "myids", 50, 52) + if err != nil { + t.Fatal(err) + } + + if err = i.GetSpecificID(49); err == nil { + t.Fatalf("Expected failure but succeeded") + } + + if err = i.GetSpecificID(53); err == nil { + t.Fatalf("Expected failure but succeeded") + } + + o, err := i.GetID() + if err != nil { + t.Fatal(err) + } + if o != 50 { + t.Fatalf("Unexpected first id returned: %d", o) + } + + err = i.GetSpecificID(50) + if err == nil { + t.Fatal(err) + } + + o, err = i.GetID() + if err != nil { + t.Fatal(err) + } + if o != 51 { + t.Fatalf("Unexpected id returned: %d", o) + } + + o, err = i.GetID() + if err != nil { + t.Fatal(err) + } + if o != 52 { + t.Fatalf("Unexpected id returned: %d", o) + } + + o, err = i.GetID() + if err == nil { + t.Fatalf("Expected failure but succeeded: %d", o) + } + + i.Release(50) + + o, err = i.GetID() + if err != nil { + t.Fatal(err) + } + if o != 50 { + t.Fatalf("Unexpected id returned") + } + + i.Release(52) + err = i.GetSpecificID(52) + if err != nil { + t.Fatal(err) + } +} + +func TestUninitialized(t *testing.T) { + i := &Idm{} + + if _, err := i.GetID(); err == nil { + t.Fatalf("Expected failure but succeeded") + } + + if err := i.GetSpecificID(44); err == nil { + t.Fatalf("Expected failure but succeeded") + } +} diff --git a/ipam/allocator.go b/ipam/allocator.go new file mode 100644 index 0000000..cec8ca2 --- /dev/null +++ b/ipam/allocator.go @@ -0,0 +1,632 @@ +package ipam + +import ( + "fmt" + "net" + "strings" + "sync" + + log "github.com/Sirupsen/logrus" + "github.com/docker/libkv/store" + "github.com/docker/libnetwork/bitseq" + "github.com/docker/libnetwork/datastore" + "github.com/docker/libnetwork/types" +) + +const ( + // The biggest configurable host subnets + minNetSize = 8 + minNetSizeV6 = 64 + // The effective network size for v6 + minNetSizeV6Eff = 96 + // The size of the host subnet used internally, it's the most granular sequence addresses + defaultInternalHostSize = 16 + // datastore keyes for ipam obkects + dsConfigKey = "ipam-config" // ipam-config// + dsDataKey = "ipam-data" // ipam-data//// +) + +// Allocator provides per address space ipv4/ipv6 book keeping +type Allocator struct { + // The internal subnets host size + internalHostSize int + // Static subnet information + subnets map[subnetKey]*SubnetInfo + // Allocated addresses in each address space's internal subnet + addresses map[subnetKey]*bitseq.Handle + // Datastore + store datastore.DataStore + App string + ID string + dbIndex uint64 + sync.Mutex +} + +// NewAllocator returns an instance of libnetwork ipam +func NewAllocator(ds datastore.DataStore) (*Allocator, error) { + a := &Allocator{} + a.subnets = make(map[subnetKey]*SubnetInfo) + a.addresses = make(map[subnetKey]*bitseq.Handle) + a.internalHostSize = defaultInternalHostSize + a.store = ds + a.App = "ipam" + a.ID = dsConfigKey + + if a.store == nil { + return a, nil + } + + // Register for status changes + a.watchForChanges() + + // Get the initial subnet configs status from the ds if present. + kvPair, err := a.store.KVStore().Get(datastore.Key(a.Key()...)) + if err != nil { + if err != store.ErrKeyNotFound { + return nil, fmt.Errorf("failed to retrieve the ipam subnet configs from datastore: %v", err) + } + return a, nil + } + a.subnetConfigFromStore(kvPair) + + // Now retrieve the list of small subnets + var inserterList []func() error + a.Lock() + for k, v := range a.subnets { + inserterList = append(inserterList, + func() error { + subnetList, err := getInternalSubnets(v.Subnet, a.internalHostSize) + if err != nil { + return fmt.Errorf("failed to load address bitmask for configured subnet %s because of %s", v.Subnet.String(), err.Error()) + } + a.insertAddressMasks(k, subnetList) + return nil + }) + } + a.Unlock() + + // Add the bitmasks, data could come from datastore + for _, f := range inserterList { + if err := f(); err != nil { + return nil, err + } + } + + return a, nil +} + +func (a *Allocator) subnetConfigFromStore(kvPair *store.KVPair) { + a.Lock() + if a.dbIndex < kvPair.LastIndex { + a.subnets = byteArrayToSubnets(kvPair.Value) + a.dbIndex = kvPair.LastIndex + } + a.Unlock() +} + +// Pointer to the configured subnets in each address space +type subnetKey struct { + addressSpace AddressSpace + subnet string + childSubnet string +} + +func (s *subnetKey) String() string { + k := fmt.Sprintf("%s/%s", s.addressSpace, s.subnet) + if s.childSubnet != "" { + k = fmt.Sprintf("%s/%s", k, s.childSubnet) + } + return k +} + +func (s *subnetKey) FromString(str string) error { + if str == "" || !strings.Contains(str, "/") { + return fmt.Errorf("invalid string form for subnetkey: %s", str) + } + + p := strings.Split(str, "/") + if len(p) != 3 && len(p) != 5 { + return fmt.Errorf("invalid string form for subnetkey: %s", str) + } + s.addressSpace = AddressSpace(p[0]) + s.subnet = fmt.Sprintf("%s/%s", p[1], p[2]) + if len(p) == 5 { + s.childSubnet = fmt.Sprintf("%s/%s", p[1], p[2]) + } + + return nil +} + +func (s *subnetKey) canonicalSubnet() *net.IPNet { + if _, sub, err := net.ParseCIDR(s.subnet); err == nil { + return sub + } + return nil +} + +func (s *subnetKey) canonicalChildSubnet() *net.IPNet { + if _, sub, err := net.ParseCIDR(s.childSubnet); err == nil { + return sub + } + return nil +} + +type ipVersion int + +const ( + v4 = 4 + v6 = 6 +) + +/******************* + * IPAMConf Contract + ********************/ + +// AddSubnet adds a subnet for the specified address space +func (a *Allocator) AddSubnet(addrSpace AddressSpace, subnetInfo *SubnetInfo) error { + // Sanity check + if addrSpace == "" { + return ErrInvalidAddressSpace + } + if subnetInfo == nil || subnetInfo.Subnet == nil { + return ErrInvalidSubnet + } + // Convert to smaller internal subnets (if needed) + subnetList, err := getInternalSubnets(subnetInfo.Subnet, a.internalHostSize) + if err != nil { + return err + } +retry: + if a.contains(addrSpace, subnetInfo) { + return ErrOverlapSubnet + } + + // Store the configured subnet and sync to datatstore + key := subnetKey{addrSpace, subnetInfo.Subnet.String(), ""} + a.Lock() + a.subnets[key] = subnetInfo + a.Unlock() + err = a.writeToStore() + if err != nil { + if _, ok := err.(types.RetryError); !ok { + return types.InternalErrorf("subnet configuration failed because of %s", err.Error()) + } + // Update to latest + if erru := a.readFromStore(); erru != nil { + // Restore and bail out + a.Lock() + delete(a.addresses, key) + a.Unlock() + return fmt.Errorf("failed to get updated subnets config from datastore (%v) after (%v)", erru, err) + } + goto retry + } + + // Insert respective bitmasks for this subnet + a.insertAddressMasks(key, subnetList) + + return nil +} + +// Create and insert the internal subnet(s) addresses masks into the address database. Mask data may come from the bitseq datastore. +func (a *Allocator) insertAddressMasks(parentKey subnetKey, internalSubnetList []*net.IPNet) error { + for _, intSub := range internalSubnetList { + var err error + ones, bits := intSub.Mask.Size() + numAddresses := 1 << uint(bits-ones) + smallKey := subnetKey{parentKey.addressSpace, parentKey.subnet, intSub.String()} + + // Insert the new address masks. AddressMask content may come from datastore + a.Lock() + a.addresses[smallKey], err = bitseq.NewHandle(dsDataKey, a.store, smallKey.String(), uint32(numAddresses)) + a.Unlock() + if err != nil { + return err + } + } + return nil +} + +// Check subnets size. In case configured subnet is v6 and host size is +// greater than 32 bits, adjust subnet to /96. +func adjustAndCheckSubnetSize(subnet *net.IPNet) (*net.IPNet, error) { + ones, bits := subnet.Mask.Size() + if v6 == getAddressVersion(subnet.IP) { + if ones < minNetSizeV6 { + return nil, ErrInvalidSubnet + } + if ones < minNetSizeV6Eff { + newMask := net.CIDRMask(minNetSizeV6Eff, bits) + return &net.IPNet{IP: subnet.IP, Mask: newMask}, nil + } + } else { + if ones < minNetSize { + return nil, ErrInvalidSubnet + } + } + return subnet, nil +} + +// Checks whether the passed subnet is a superset or subset of any of the subset in the db +func (a *Allocator) contains(space AddressSpace, subInfo *SubnetInfo) bool { + a.Lock() + defer a.Unlock() + for k, v := range a.subnets { + if space == k.addressSpace { + if subInfo.Subnet.Contains(v.Subnet.IP) || + v.Subnet.Contains(subInfo.Subnet.IP) { + return true + } + } + } + return false +} + +// Splits the passed subnet into N internal subnets with host size equal to internalHostSize. +// If the subnet's host size is equal to or smaller than internalHostSize, there won't be any +// split and the return list will contain only the passed subnet. +func getInternalSubnets(inSubnet *net.IPNet, internalHostSize int) ([]*net.IPNet, error) { + var subnetList []*net.IPNet + + // Sanity check and size adjustment for v6 + subnet, err := adjustAndCheckSubnetSize(inSubnet) + if err != nil { + return subnetList, err + } + + // Get network/host subnet information + netBits, bits := subnet.Mask.Size() + hostBits := bits - netBits + + extraBits := hostBits - internalHostSize + if extraBits <= 0 { + subnetList = make([]*net.IPNet, 1) + subnetList[0] = subnet + } else { + // Split in smaller internal subnets + numIntSubs := 1 << uint(extraBits) + subnetList = make([]*net.IPNet, numIntSubs) + + // Construct one copy of the internal subnets's mask + intNetBits := bits - internalHostSize + intMask := net.CIDRMask(intNetBits, bits) + + // Construct the prefix portion for each internal subnet + for i := 0; i < numIntSubs; i++ { + intIP := make([]byte, len(subnet.IP)) + copy(intIP, subnet.IP) // IPv6 is too big, just work on the extra portion + addIntToIP(intIP, i< 192.168.53 +func addIntToIP(array []byte, ordinal int) { + for i := len(array) - 1; i >= 0; i-- { + array[i] |= (byte)(ordinal & 0xff) + ordinal >>= 8 + } +} + +// Convert an ordinal to the respective IP address +func ipToInt(ip []byte) int { + value := 0 + for i := 0; i < len(ip); i++ { + j := len(ip) - 1 - i + value += int(ip[i]) << uint(j*8) + } + return value +} + +// Given an address and subnet, returns the host portion address +func getHostPortionIP(address net.IP, subnet *net.IPNet) net.IP { + hostPortion := make([]byte, len(address)) + for i := 0; i < len(subnet.Mask); i++ { + hostPortion[i] = address[i] &^ subnet.Mask[i] + } + return hostPortion +} + +func printLine(head *bitseq.Sequence) { + fmt.Println() + for head != nil { + fmt.Printf("-") + head = head.Next + } +} diff --git a/ipam/allocator_test.go b/ipam/allocator_test.go new file mode 100644 index 0000000..cf67a75 --- /dev/null +++ b/ipam/allocator_test.go @@ -0,0 +1,587 @@ +package ipam + +import ( + "fmt" + "net" + "testing" + "time" + + "github.com/docker/libnetwork/bitseq" +) + +func getAllocator(t *testing.T, subnet *net.IPNet) *Allocator { + a, err := NewAllocator(nil) + if err != nil { + t.Fatal(err) + } + a.AddSubnet("default", &SubnetInfo{Subnet: subnet}) + return a +} + +func TestInt2IP2IntConversion(t *testing.T) { + for i := 0; i < 256*256*256; i++ { + var array [4]byte // new array at each cycle + addIntToIP(array[:], i) + j := ipToInt(array[:]) + if j != i { + t.Fatalf("Failed to convert ordinal %d to IP % x and back to ordinal. Got %d", i, array, j) + } + } +} + +func TestIsValid(t *testing.T) { + list := []int{0, 255, 256, 511, 512, 767, 768} + for _, i := range list { + if isValidIP(i) { + t.Fatalf("Failed to detect invalid IPv4 ordinal: %d", i) + } + } + + list = []int{1, 254, 257, 258, 510, 513, 769, 770} + for _, i := range list { + if !isValidIP(i) { + t.Fatalf("Marked valid ipv4 as invalid: %d", i) + } + } +} + +func TestGetAddressVersion(t *testing.T) { + if v4 != getAddressVersion(net.ParseIP("172.28.30.112")) { + t.Fatalf("Failed to detect IPv4 version") + } + if v4 != getAddressVersion(net.ParseIP("0.0.0.1")) { + t.Fatalf("Failed to detect IPv4 version") + } + if v6 != getAddressVersion(net.ParseIP("ff01::1")) { + t.Fatalf("Failed to detect IPv6 version") + } + if v6 != getAddressVersion(net.ParseIP("2001:56::76:51")) { + t.Fatalf("Failed to detect IPv6 version") + } +} + +func TestKeyString(t *testing.T) { + + k := &subnetKey{addressSpace: "default", subnet: "172.27.0.0/16"} + expected := "default/172.27.0.0/16" + if expected != k.String() { + t.Fatalf("Unexpected key string: %s", k.String()) + } + + k2 := &subnetKey{} + err := k2.FromString(expected) + if err != nil { + t.Fatal(err) + } + if k2.addressSpace != k.addressSpace || k2.subnet != k.subnet { + t.Fatalf("subnetKey.FromString() failed. Expected %v. Got %v", k, k2) + } + + expected = fmt.Sprintf("%s/%s", expected, "172.27.3.0/24") + k.childSubnet = "172.27.3.0/24" + if expected != k.String() { + t.Fatalf("Unexpected key string: %s", k.String()) + } + + err = k2.FromString(expected) + if err != nil { + t.Fatal(err) + } + if k2.addressSpace != k.addressSpace || k2.subnet != k.subnet { + t.Fatalf("subnetKey.FromString() failed. Expected %v. Got %v", k, k2) + } +} + +func TestAddSubnets(t *testing.T) { + a, err := NewAllocator(nil) + if err != nil { + t.Fatal(err) + } + + _, sub0, _ := net.ParseCIDR("10.0.0.0/8") + err = a.AddSubnet("default", &SubnetInfo{Subnet: sub0}) + if err != nil { + t.Fatalf("Unexpected failure in adding subent") + } + + err = a.AddSubnet("abc", &SubnetInfo{Subnet: sub0}) + if err != nil { + t.Fatalf("Unexpected failure in adding overlapping subents to different address spaces") + } + + err = a.AddSubnet("abc", &SubnetInfo{Subnet: sub0}) + if err == nil { + t.Fatalf("Failed to detect overlapping subnets: %s and %s", sub0, sub0) + } + + _, sub1, _ := net.ParseCIDR("10.20.2.0/24") + err = a.AddSubnet("default", &SubnetInfo{Subnet: sub1}) + if err == nil { + t.Fatalf("Failed to detect overlapping subnets: %s and %s", sub0, sub1) + } + + _, sub2, _ := net.ParseCIDR("10.128.0.0/9") + err = a.AddSubnet("default", &SubnetInfo{Subnet: sub2}) + if err == nil { + t.Fatalf("Failed to detect overlapping subnets: %s and %s", sub1, sub2) + } + + _, sub6, err := net.ParseCIDR("1003:1:2:3:4:5:6::/112") + if err != nil { + t.Fatalf("Wrong input, Can't proceed: %s", err.Error()) + } + err = a.AddSubnet("default", &SubnetInfo{Subnet: sub6}) + if err != nil { + t.Fatalf("Failed to add v6 subnet: %s", err.Error()) + } + + _, sub6, err = net.ParseCIDR("1003:1:2:3::/64") + if err != nil { + t.Fatalf("Wrong input, Can't proceed: %s", err.Error()) + } + err = a.AddSubnet("default", &SubnetInfo{Subnet: sub6}) + if err == nil { + t.Fatalf("Failed to detect overlapping v6 subnet") + } +} + +func TestAdjustAndCheckSubnet(t *testing.T) { + _, sub6, _ := net.ParseCIDR("1003:1:2:300::/63") + _, err := adjustAndCheckSubnetSize(sub6) + if err == nil { + t.Fatalf("Failed detect too big v6 subnet") + } + + _, sub, _ := net.ParseCIDR("192.0.0.0/7") + _, err = adjustAndCheckSubnetSize(sub) + if err == nil { + t.Fatalf("Failed detect too big v4 subnet") + } + + subnet := "1004:1:2:6::/64" + _, sub6, _ = net.ParseCIDR(subnet) + subnetToSplit, err := adjustAndCheckSubnetSize(sub6) + if err != nil { + t.Fatalf("Unexpected error returned by adjustAndCheckSubnetSize()") + } + ones, _ := subnetToSplit.Mask.Size() + if ones < minNetSizeV6Eff { + t.Fatalf("Wrong effective network size for %s. Expected: %d. Got: %d", subnet, minNetSizeV6Eff, ones) + } +} + +func TestRemoveSubnet(t *testing.T) { + a, err := NewAllocator(nil) + if err != nil { + t.Fatal(err) + } + + input := []struct { + addrSpace AddressSpace + subnet string + }{ + {"default", "192.168.0.0/16"}, + {"default", "172.17.0.0/16"}, + {"default", "10.0.0.0/8"}, + {"default", "2002:1:2:3:4:5:ffff::/112"}, + {"splane", "172.17.0.0/16"}, + {"splane", "10.0.0.0/8"}, + {"splane", "2002:1:2:3:4:5:6::/112"}, + {"splane", "2002:1:2:3:4:5:ffff::/112"}, + } + + for _, i := range input { + _, sub, err := net.ParseCIDR(i.subnet) + if err != nil { + t.Fatalf("Wrong input, Can't proceed: %s", err.Error()) + } + err = a.AddSubnet(i.addrSpace, &SubnetInfo{Subnet: sub}) + if err != nil { + t.Fatalf("Failed to apply input. Can't proceed: %s", err.Error()) + } + } + + _, sub, _ := net.ParseCIDR("172.17.0.0/16") + a.RemoveSubnet("default", sub) + if len(a.subnets) != 7 { + t.Fatalf("Failed to remove subnet info") + } + list := a.getSubnetList("default", v4) + if len(list) != 257 { + t.Fatalf("Failed to effectively remove subnet address space") + } + + _, sub, _ = net.ParseCIDR("2002:1:2:3:4:5:ffff::/112") + a.RemoveSubnet("default", sub) + if len(a.subnets) != 6 { + t.Fatalf("Failed to remove subnet info") + } + list = a.getSubnetList("default", v6) + if len(list) != 0 { + t.Fatalf("Failed to effectively remove subnet address space") + } + + _, sub, _ = net.ParseCIDR("2002:1:2:3:4:5:6::/112") + a.RemoveSubnet("splane", sub) + if len(a.subnets) != 5 { + t.Fatalf("Failed to remove subnet info") + } + list = a.getSubnetList("splane", v6) + if len(list) != 1 { + t.Fatalf("Failed to effectively remove subnet address space") + } +} + +func TestGetInternalSubnets(t *testing.T) { + // This function tests the splitting of a parent subnet in small host subnets. + // The splitting is controlled by the max host size, which is the first parameter + // passed to the function. It basically says if the parent subnet host size is + // greater than the max host size, split the parent subnet into N internal small + // subnets with host size = max host size to cover the same address space. + + input := []struct { + internalHostSize int + parentSubnet string + firstIntSubnet string + lastIntSubnet string + }{ + // Test 8 bits prefix network + {24, "10.0.0.0/8", "10.0.0.0/8", "10.0.0.0/8"}, + {16, "10.0.0.0/8", "10.0.0.0/16", "10.255.0.0/16"}, + {8, "10.0.0.0/8", "10.0.0.0/24", "10.255.255.0/24"}, + // Test 16 bits prefix network + {16, "192.168.0.0/16", "192.168.0.0/16", "192.168.0.0/16"}, + {8, "192.168.0.0/16", "192.168.0.0/24", "192.168.255.0/24"}, + // Test 24 bits prefix network + {16, "192.168.57.0/24", "192.168.57.0/24", "192.168.57.0/24"}, + {8, "192.168.57.0/24", "192.168.57.0/24", "192.168.57.0/24"}, + // Test non byte multiple host size + {24, "10.0.0.0/8", "10.0.0.0/8", "10.0.0.0/8"}, + {20, "10.0.0.0/12", "10.0.0.0/12", "10.0.0.0/12"}, + {20, "10.128.0.0/12", "10.128.0.0/12", "10.128.0.0/12"}, + {12, "10.16.0.0/16", "10.16.0.0/20", "10.16.240.0/20"}, + {13, "10.0.0.0/8", "10.0.0.0/19", "10.255.224.0/19"}, + {15, "10.0.0.0/8", "10.0.0.0/17", "10.255.128.0/17"}, + // Test v6 network + {16, "2002:1:2:3:4:5:6000::/110", "2002:1:2:3:4:5:6000:0/112", "2002:1:2:3:4:5:6003:0/112"}, + {16, "2002:1:2:3:4:5:ff00::/104", "2002:1:2:3:4:5:ff00:0/112", "2002:1:2:3:4:5:ffff:0/112"}, + {12, "2002:1:2:3:4:5:ffff::/112", "2002:1:2:3:4:5:ffff:0/116", "2002:1:2:3:4:5:ffff:f000/116"}, + {11, "2002:1:2:3:4:5:ffff::/112", "2002:1:2:3:4:5:ffff:0/117", "2002:1:2:3:4:5:ffff:f800/117"}, + } + + for _, d := range input { + assertInternalSubnet(t, d.internalHostSize, d.parentSubnet, d.firstIntSubnet, d.lastIntSubnet) + } + +} + +func TestGetAddress(t *testing.T) { + input := []string{ + /*"10.0.0.0/8", "10.0.0.0/9", */ "10.0.0.0/10", "10.0.0.0/11", "10.0.0.0/12", "10.0.0.0/13", "10.0.0.0/14", + "10.0.0.0/15", "10.0.0.0/16", "10.0.0.0/17", "10.0.0.0/18", "10.0.0.0/19", "10.0.0.0/20", "10.0.0.0/21", + "10.0.0.0/22", "10.0.0.0/23", "10.0.0.0/24", "10.0.0.0/25", "10.0.0.0/26", "10.0.0.0/27", "10.0.0.0/28", + "10.0.0.0/29", "10.0.0.0/30", "10.0.0.0/31"} + + for _, subnet := range input { + assertGetAddress(t, subnet) + } +} + +func TestGetSubnetList(t *testing.T) { + a, err := NewAllocator(nil) + if err != nil { + t.Fatal(err) + } + input := []struct { + addrSpace AddressSpace + subnet string + }{ + {"default", "192.168.0.0/16"}, + {"default", "172.17.0.0/16"}, + {"default", "10.0.0.0/8"}, + {"default", "2002:1:2:3:4:5:6::/112"}, + {"default", "2002:1:2:3:4:5:ffff::/112"}, + {"splane", "172.17.0.0/16"}, + {"splane", "10.0.0.0/8"}, + {"splane", "2002:1:2:3:4:5:ff00::/104"}, + } + + for _, i := range input { + _, sub, err := net.ParseCIDR(i.subnet) + if err != nil { + t.Fatalf("Wrong input, Can't proceed: %s", err.Error()) + } + err = a.AddSubnet(i.addrSpace, &SubnetInfo{Subnet: sub}) + if err != nil { + t.Fatalf("Failed to apply input. Can't proceed: %s", err.Error()) + } + } + + list := a.getSubnetList("default", v4) + if len(list) != 258 { + t.Fatalf("Incorrect number of internal subnets for ipv4 version. Expected 258. Got %d.", len(list)) + } + list = a.getSubnetList("splane", v4) + if len(list) != 257 { + t.Fatalf("Incorrect number of internal subnets for ipv4 version. Expected 257. Got %d.", len(list)) + } + + list = a.getSubnetList("default", v6) + if len(list) != 2 { + t.Fatalf("Incorrect number of internal subnets for ipv6 version. Expected 2. Got %d.", len(list)) + } + list = a.getSubnetList("splane", v6) + if len(list) != 256 { + t.Fatalf("Incorrect number of internal subnets for ipv6 version. Expected 256. Got %d.", len(list)) + } + +} + +func TestRequestSyntaxCheck(t *testing.T) { + var ( + subnet = "192.168.0.0/16" + addSpace = AddressSpace("green") + ) + + a, err := NewAllocator(nil) + if err != nil { + t.Fatal(err) + } + + // Add subnet and create base request + _, sub, _ := net.ParseCIDR(subnet) + a.AddSubnet(addSpace, &SubnetInfo{Subnet: sub}) + req := &AddressRequest{Subnet: *sub} + + // Empty address space request + _, err = a.Request("", req) + if err == nil { + t.Fatalf("Failed to detect wrong request: empty address space") + } + + // Preferred address from different subnet in request + req.Address = net.ParseIP("172.17.0.23") + _, err = a.Request(addSpace, req) + if err == nil { + t.Fatalf("Failed to detect wrong request: preferred IP from different subnet") + } + + // Preferred address specified and nil subnet + req = &AddressRequest{Address: net.ParseIP("172.17.0.23")} + _, err = a.Request(addSpace, req) + if err == nil { + t.Fatalf("Failed to detect wrong request: subnet not specified but preferred address specified") + } +} + +func TestRequest(t *testing.T) { + // Request N addresses from different size subnets, verifying last request + // returns expected address. Internal subnet host size is Allocator's default, 16 + input := []struct { + subnet string + numReq int + lastIP string + }{ + {"192.168.59.0/24", 254, "192.168.59.254"}, + {"192.168.240.0/20", 254, "192.168.240.254"}, + {"192.168.0.0/16", 254, "192.168.0.254"}, + {"10.16.0.0/16", 254, "10.16.0.254"}, + {"10.128.0.0/12", 254, "10.128.0.254"}, + {"10.0.0.0/8", 254, "10.0.0.254"}, + {"192.168.0.0/16", 256, "192.168.1.2"}, + {"10.0.0.0/8", 256, "10.0.1.2"}, + + {"192.168.128.0/18", 4 * 254, "192.168.131.254"}, + {"192.168.240.0/20", 16 * 254, "192.168.255.254"}, + + {"192.168.0.0/16", 256 * 254, "192.168.255.254"}, + {"10.0.0.0/8", 2 * 254, "10.0.1.254"}, + {"10.0.0.0/8", 5 * 254, "10.0.4.254"}, + //{"10.0.0.0/8", 100 * 256 * 254, "10.99.255.254"}, + } + + for _, d := range input { + assertNRequests(t, d.subnet, d.numReq, d.lastIP) + } +} + +func TestRelease(t *testing.T) { + var ( + err error + req *AddressRequest + subnet = "192.168.0.0/16" + ) + + _, sub, _ := net.ParseCIDR(subnet) + a := getAllocator(t, sub) + req = &AddressRequest{Subnet: *sub} + bm := a.addresses[subnetKey{"default", subnet, subnet}] + + // Allocate all addresses + for err != ErrNoAvailableIPs { + _, err = a.Request("default", req) + } + + toRelease := []struct { + address string + }{ + {"192.168.0.1"}, + {"192.168.0.2"}, + {"192.168.0.3"}, + {"192.168.0.4"}, + {"192.168.0.5"}, + {"192.168.0.6"}, + {"192.168.0.7"}, + {"192.168.0.8"}, + {"192.168.0.9"}, + {"192.168.0.10"}, + {"192.168.0.30"}, + {"192.168.0.31"}, + {"192.168.1.32"}, + + {"192.168.0.254"}, + {"192.168.1.1"}, + {"192.168.1.2"}, + + {"192.168.1.3"}, + + {"192.168.255.253"}, + {"192.168.255.254"}, + } + + // One by one, relase the address and request again. We should get the same IP + req = &AddressRequest{Subnet: *sub} + for i, inp := range toRelease { + address := net.ParseIP(inp.address) + a.Release("default", address) + if bm.Unselected() != 1 { + t.Fatalf("Failed to update free address count after release. Expected %d, Found: %d", i+1, bm.Unselected()) + } + + rsp, err := a.Request("default", req) + if err != nil { + t.Fatalf("Failed to obtain the address: %s", err.Error()) + } + if !address.Equal(rsp.Address) { + t.Fatalf("Failed to obtain the same address. Expected: %s, Got: %s", address, rsp.Address) + } + } +} + +func assertInternalSubnet(t *testing.T, hostSize int, bigSubnet, firstSmall, lastSmall string) { + _, subnet, _ := net.ParseCIDR(bigSubnet) + list, _ := getInternalSubnets(subnet, hostSize) + count := 1 + ones, bits := subnet.Mask.Size() + diff := bits - ones - hostSize + if diff > 0 { + count <<= uint(diff) + } + + if len(list) != count { + t.Fatalf("Wrong small subnets number. Expected: %d, Got: %d", count, len(list)) + } + if firstSmall != list[0].String() { + t.Fatalf("Wrong first small subent. Expected: %v, Got: %v", firstSmall, list[0]) + } + if lastSmall != list[count-1].String() { + t.Fatalf("Wrong last small subent. Expected: %v, Got: %v", lastSmall, list[count-1]) + } +} + +func assertGetAddress(t *testing.T, subnet string) { + var ( + err error + printTime = false + a = &Allocator{} + ) + + _, sub, _ := net.ParseCIDR(subnet) + ones, bits := sub.Mask.Size() + zeroes := bits - ones + numAddresses := 1 << uint(zeroes) + + bm, err := bitseq.NewHandle("ipam_test", nil, "default/192.168.0.0/24", uint32(numAddresses)) + if err != nil { + t.Fatal(err) + } + + start := time.Now() + run := 0 + for err != ErrNoAvailableIPs { + _, err = a.getAddress(sub, bm, nil, v4) + run++ + } + if printTime { + fmt.Printf("\nTaken %v, to allocate all addresses on %s. (nemAddresses: %d. Runs: %d)", time.Since(start), subnet, numAddresses, run) + } + if bm.Unselected() != 0 { + t.Fatalf("Unexpected free count after reserving all addresses: %d", bm.Unselected()) + } + /* + if bm.Head.Block != expectedMax || bm.Head.Count != numBlocks { + t.Fatalf("Failed to effectively reserve all addresses on %s. Expected (0x%x, %d) as first sequence. Found (0x%x,%d)", + subnet, expectedMax, numBlocks, bm.Head.Block, bm.Head.Count) + } + */ +} + +func assertNRequests(t *testing.T, subnet string, numReq int, lastExpectedIP string) { + var ( + err error + req *AddressRequest + rsp *AddressResponse + printTime = false + ) + + _, sub, _ := net.ParseCIDR(subnet) + lastIP := net.ParseIP(lastExpectedIP) + + a := getAllocator(t, sub) + req = &AddressRequest{Subnet: *sub} + + i := 0 + start := time.Now() + for ; i < numReq; i++ { + rsp, err = a.Request("default", req) + } + if printTime { + fmt.Printf("\nTaken %v, to allocate %d addresses on %s\n", time.Since(start), numReq, subnet) + } + + if !lastIP.Equal(rsp.Address) { + t.Fatalf("Wrong last IP. Expected %s. Got: %s (err: %v, ind: %d)", lastExpectedIP, rsp.Address.String(), err, i) + } +} + +func benchmarkRequest(subnet *net.IPNet) { + var err error + + a, _ := NewAllocator(nil) + a.internalHostSize = 20 + a.AddSubnet("default", &SubnetInfo{Subnet: subnet}) + + req := &AddressRequest{Subnet: *subnet} + for err != ErrNoAvailableIPs { + _, err = a.Request("default", req) + + } +} + +func benchMarkRequest(subnet *net.IPNet, b *testing.B) { + for n := 0; n < b.N; n++ { + benchmarkRequest(subnet) + } +} + +func BenchmarkRequest_24(b *testing.B) { + benchmarkRequest(&net.IPNet{IP: []byte{10, 0, 0, 0}, Mask: []byte{255, 255, 255, 0}}) +} + +func BenchmarkRequest_16(b *testing.B) { + benchmarkRequest(&net.IPNet{IP: []byte{10, 0, 0, 0}, Mask: []byte{255, 255, 0, 0}}) +} + +func BenchmarkRequest_8(b *testing.B) { + benchmarkRequest(&net.IPNet{IP: []byte{10, 0, 0, 0}, Mask: []byte{255, 0xfc, 0, 0}}) +} diff --git a/ipam/contract.go b/ipam/contract.go new file mode 100644 index 0000000..618ada2 --- /dev/null +++ b/ipam/contract.go @@ -0,0 +1,100 @@ +// Package ipam that specifies the contract the IPAM plugin need to satisfy, +// decoupling IPAM interface and implementation. +package ipam + +import ( + "errors" + "net" +) + +/************** + * IPAM Errors + **************/ + +// ErrIpamNotAvailable is returned when the plugin prviding the IPAM service is not available +var ( + ErrInvalidIpamService = errors.New("Invalid IPAM Service") + ErrInvalidIpamConfigService = errors.New("Invalid IPAM Config Service") + ErrIpamNotAvailable = errors.New("IPAM Service not available") + ErrIpamInternalError = errors.New("IPAM Internal Error") + ErrInvalidAddressSpace = errors.New("Invalid Address Space") + ErrInvalidSubnet = errors.New("Invalid Subnet") + ErrInvalidRequest = errors.New("Invalid Request") + ErrSubnetNotFound = errors.New("Subnet not found") + ErrOverlapSubnet = errors.New("Subnet overlaps with existing subnet on this address space") + ErrNoAvailableSubnet = errors.New("No available subnet") + ErrNoAvailableIPs = errors.New("No available addresses on subnet") + ErrIPAlreadyAllocated = errors.New("Address already in use") + ErrIPOutOfRange = errors.New("Requested address is out of range") + ErrSubnetAlreadyRegistered = errors.New("Subnet already registered on this address space") + ErrBadSubnet = errors.New("Address space does not contain specified subnet") +) + +// AddressSpace identifies a unique pool of network addresses +type AddressSpace string + +/******************************* + * IPAM Configuration Interface + *******************************/ + +// Config represents the interface the IPAM service plugins must implement +// in order to allow injection/modification of IPAM database. +// Common key is a addressspace +type Config interface { + // AddSubnet adds a subnet to the specified address space + AddSubnet(AddressSpace, *SubnetInfo) error + // RemoveSubnet removes a subnet from the specified address space + RemoveSubnet(AddressSpace, *net.IPNet) error + // AddVendorInfo adds Vendor specific data + AddVendorInfo([]byte) error +} + +// SubnetInfo contains the information subnet hosts need in order to communicate +type SubnetInfo struct { + Subnet *net.IPNet + Gateway net.IP + OpaqueData []byte // Vendor specific +} + +/************************* + * IPAM Service Interface + *************************/ + +// IPAM defines the interface that needs to be implemented by IPAM service plugin +// Common key is a unique address space identifier +type IPAM interface { + // Request address from the specified address space + Request(AddressSpace, *AddressRequest) (*AddressResponse, error) + // Separate API for IPv6 + RequestV6(AddressSpace, *AddressRequest) (*AddressResponse, error) + // Release the address from the specified address space + Release(AddressSpace, net.IP) +} + +// AddressRequest encloses the information a client +// needs to pass to IPAM when requesting an address +type AddressRequest struct { + Subnet net.IPNet // Preferred subnet pool (Optional) + Address net.IP // Preferred address (Optional) + Endpoint string // For static IP mapping (Optional) + OpaqueData []byte // Vendor specific request data +} + +// Validate runs syntactic validation on this AddressRequest object +func (req *AddressRequest) Validate() error { + var byteArray []byte = req.Address + + // Check preferred address + if byteArray != nil && (&req.Subnet == nil || !req.Subnet.Contains(req.Address)) { + return ErrInvalidRequest + } + + return nil +} + +// AddressResponse represents the IPAM service's +// response to an address request +type AddressResponse struct { + Address net.IP + Subnet SubnetInfo +} diff --git a/ipam/store.go b/ipam/store.go new file mode 100644 index 0000000..f1fb8ae --- /dev/null +++ b/ipam/store.go @@ -0,0 +1,164 @@ +package ipam + +import ( + "encoding/json" + "net" + + log "github.com/Sirupsen/logrus" + "github.com/docker/libnetwork/datastore" + "github.com/docker/libnetwork/types" +) + +// Key provides the Key to be used in KV Store +func (a *Allocator) Key() []string { + a.Lock() + defer a.Unlock() + return []string{a.App, a.ID} +} + +// KeyPrefix returns the immediate parent key that can be used for tree walk +func (a *Allocator) KeyPrefix() []string { + a.Lock() + defer a.Unlock() + return []string{a.App} +} + +// Value marshals the data to be stored in the KV store +func (a *Allocator) Value() []byte { + a.Lock() + defer a.Unlock() + + if a.subnets == nil { + return []byte{} + } + + b, err := subnetsToByteArray(a.subnets) + if err != nil { + return nil + } + return b +} + +func subnetsToByteArray(m map[subnetKey]*SubnetInfo) ([]byte, error) { + if m == nil { + return nil, nil + } + + mm := make(map[string]string, len(m)) + for k, v := range m { + mm[k.String()] = v.Subnet.String() + } + + return json.Marshal(mm) +} + +func byteArrayToSubnets(ba []byte) map[subnetKey]*SubnetInfo { + m := map[subnetKey]*SubnetInfo{} + + if ba == nil || len(ba) == 0 { + return m + } + + var mm map[string]string + err := json.Unmarshal(ba, &mm) + if err != nil { + log.Warnf("Failed to decode subnets byte array: %v", err) + return m + } + for ks, vs := range mm { + sk := subnetKey{} + if err := sk.FromString(ks); err != nil { + log.Warnf("Failed to decode subnets map entry: (%s, %s)", ks, vs) + continue + } + si := &SubnetInfo{} + _, nw, err := net.ParseCIDR(vs) + if err != nil { + log.Warnf("Failed to decode subnets map entry value: (%s, %s)", ks, vs) + continue + } + si.Subnet = nw + m[sk] = si + } + return m +} + +// Index returns the latest DB Index as seen by this object +func (a *Allocator) Index() uint64 { + a.Lock() + defer a.Unlock() + return a.dbIndex +} + +// SetIndex method allows the datastore to store the latest DB Index into this object +func (a *Allocator) SetIndex(index uint64) { + a.Lock() + a.dbIndex = index + a.Unlock() +} + +func (a *Allocator) watchForChanges() error { + if a.store == nil { + return nil + } + + kvpChan, err := a.store.KVStore().Watch(datastore.Key(a.Key()...), nil) + if err != nil { + return err + } + go func() { + for { + select { + case kvPair := <-kvpChan: + if kvPair != nil { + log.Debugf("Got notification for key %v: %v", kvPair.Key, kvPair.Value) + a.subnetConfigFromStore(kvPair) + } + } + } + }() + return nil +} + +func (a *Allocator) readFromStore() error { + a.Lock() + store := a.store + a.Unlock() + + if store == nil { + return nil + } + + kvPair, err := a.store.KVStore().Get(datastore.Key(a.Key()...)) + if err != nil { + return err + } + + a.subnetConfigFromStore(kvPair) + + return nil +} + +func (a *Allocator) writeToStore() error { + a.Lock() + store := a.store + a.Unlock() + if store == nil { + return nil + } + err := store.PutObjectAtomic(a) + if err == datastore.ErrKeyModified { + return types.RetryErrorf("failed to perform atomic write (%v). retry might fix the error", err) + } + return err +} + +func (a *Allocator) deleteFromStore() error { + a.Lock() + store := a.store + a.Unlock() + if store == nil { + return nil + } + return store.DeleteObjectAtomic(a) +} diff --git a/netutils/utils.go b/netutils/utils.go index 222b7a2..0ef357e 100644 --- a/netutils/utils.go +++ b/netutils/utils.go @@ -168,3 +168,53 @@ func GenerateIfaceName(prefix string, len int) (string, error) { } return "", types.InternalErrorf("could not generate interface name") } + +func byteArrayToInt(array []byte, numBytes int) uint64 { + if numBytes <= 0 || numBytes > 8 { + panic("Invalid argument") + } + num := 0 + for i := 0; i <= len(array)-1; i++ { + num += int(array[len(array)-1-i]) << uint(i*8) + } + return uint64(num) +} + +// ATo64 converts a byte array into a uint32 +func ATo64(array []byte) uint64 { + return byteArrayToInt(array, 8) +} + +// ATo32 converts a byte array into a uint32 +func ATo32(array []byte) uint32 { + return uint32(byteArrayToInt(array, 4)) +} + +// ATo16 converts a byte array into a uint16 +func ATo16(array []byte) uint16 { + return uint16(byteArrayToInt(array, 2)) +} + +func intToByteArray(val uint64, numBytes int) []byte { + array := make([]byte, numBytes) + for i := numBytes - 1; i >= 0; i-- { + array[i] = byte(val & 0xff) + val = val >> 8 + } + return array +} + +// U64ToA converts a uint64 to a byte array +func U64ToA(val uint64) []byte { + return intToByteArray(uint64(val), 8) +} + +// U32ToA converts a uint64 to a byte array +func U32ToA(val uint32) []byte { + return intToByteArray(uint64(val), 4) +} + +// U16ToA converts a uint64 to a byte array +func U16ToA(val uint16) []byte { + return intToByteArray(uint64(val), 2) +} diff --git a/types/types.go b/types/types.go index 02fbcb1..9344068 100644 --- a/types/types.go +++ b/types/types.go @@ -228,6 +228,12 @@ type MaskableError interface { Maskable() } +// RetryError is an interface for errors which might get resolved through retry +type RetryError interface { + // Retry makes implementer into RetryError type + Retry() +} + // BadRequestError is an interface for errors originated by a bad request type BadRequestError interface { // BadRequest makes implementer into BadRequestError type @@ -271,7 +277,7 @@ type InternalError interface { } /****************************** - * Weel-known Error Formatters + * Well-known Error Formatters ******************************/ // BadRequestErrorf creates an instance of BadRequestError @@ -314,6 +320,11 @@ func InternalMaskableErrorf(format string, params ...interface{}) error { return maskInternal(fmt.Sprintf(format, params...)) } +// RetryErrorf creates an instance of RetryError +func RetryErrorf(format string, params ...interface{}) error { + return retry(fmt.Sprintf(format, params...)) +} + /*********************** * Internal Error Types ***********************/ @@ -377,3 +388,10 @@ func (mnt maskInternal) Error() string { } func (mnt maskInternal) Internal() {} func (mnt maskInternal) Maskable() {} + +type retry string + +func (r retry) Error() string { + return string(r) +} +func (r retry) Retry() {} diff --git a/types/types_test.go b/types/types_test.go index 15a58e1..efa26d7 100644 --- a/types/types_test.go +++ b/types/types_test.go @@ -21,6 +21,17 @@ func TestErrorConstructors(t *testing.T) { t.Fatal(err) } + err = RetryErrorf("Incy wincy %s went up the spout again", "spider") + if err.Error() != "Incy wincy spider went up the spout again" { + t.Fatal(err) + } + if _, ok := err.(RetryError); !ok { + t.Fatal(err) + } + if _, ok := err.(MaskableError); ok { + t.Fatal(err) + } + err = NotFoundErrorf("Can't find the %s", "keys") if err.Error() != "Can't find the keys" { t.Fatal(err)