diff --git a/ipam/allocator.go b/ipam/allocator.go index feed9741..dcd673b5 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -19,6 +19,7 @@ import ( const ( msgSpaceRequest = iota msgRingUpdate + msgSpaceRequestDenied paxosInterval = time.Second * 5 MinSubnetSize = 4 // first and last addresses are excluded, so 2 would be too small @@ -163,6 +164,18 @@ func (alloc *Allocator) tryPendingOps() { } } +func (alloc *Allocator) spaceRequestDenied(sender router.PeerName, r address.Range) { + for i := 0; i < len(alloc.pendingClaims); { + claim := alloc.pendingClaims[i].(*claim) + if r.Contains(claim.addr) { + claim.DeniedBy(alloc, sender) + alloc.pendingClaims = append(alloc.pendingClaims[:i], alloc.pendingClaims[i+1:]...) + continue + } + i++ + } +} + func hasBeenCancelled(cancelChan <-chan bool) func() bool { return func() bool { select { @@ -360,6 +373,12 @@ func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) erro alloc.donateSpace(r, sender) } resultChan <- err + case msgSpaceRequestDenied: + r, err := decodeRange(msg[1:]) + if err == nil { + alloc.spaceRequestDenied(sender, r) + } + resultChan <- err case msgRingUpdate: resultChan <- alloc.update(sender, msg[1:]) } @@ -576,13 +595,22 @@ func (alloc *Allocator) propose() { alloc.gossip.GossipBroadcast(alloc.Gossip()) } -func (alloc *Allocator) sendSpaceRequest(dest router.PeerName, r address.Range) error { +func encodeRange(r address.Range) []byte { buf := new(bytes.Buffer) enc := gob.NewEncoder(buf) if err := enc.Encode(r); err != nil { panic(err) } - msg := router.Concat([]byte{msgSpaceRequest}, buf.Bytes()) + return buf.Bytes() +} + +func (alloc *Allocator) sendSpaceRequest(dest router.PeerName, r address.Range) error { + msg := router.Concat([]byte{msgSpaceRequest}, encodeRange(r)) + return alloc.gossip.GossipUnicast(dest, msg) +} + +func (alloc *Allocator) sendSpaceRequestDenied(dest router.PeerName, r address.Range) error { + msg := router.Concat([]byte{msgSpaceRequestDenied}, encodeRange(r)) return alloc.gossip.GossipUnicast(dest, msg) } @@ -659,10 +687,14 @@ func (alloc *Allocator) donateSpace(r address.Range, to router.PeerName) { free := alloc.space.NumFreeAddressesInRange(r) common.Assert(free == 0) alloc.debugln("No space to give to peer", to) + // separate message maintains backwards-compatibility: + // down-level peers will ignore this and still get the ring update. + alloc.sendSpaceRequestDenied(to, r) return } alloc.debugln("Giving range", chunk, "to", to) alloc.ring.GrantRangeToHost(chunk.Start, chunk.End, to) + alloc.sendRingUpdate(to) } func (alloc *Allocator) assertInvariants() { diff --git a/ipam/claim.go b/ipam/claim.go index d37f0bce..642898f1 100644 --- a/ipam/claim.go +++ b/ipam/claim.go @@ -44,12 +44,9 @@ func (c *claim) Try(alloc *Allocator) bool { c.sendResult(nil) // don't make the caller wait return false default: - name, found := alloc.nicknames[owner] - if found { - name = " (" + name + ")" - } - c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name)) - return true + alloc.debugf("requesting address %s from other peer %s", c.addr, owner) + alloc.sendSpaceRequest(owner, address.NewRange(c.addr, 1)) + return false } // We are the owner, check we haven't given it to another container @@ -72,6 +69,14 @@ func (c *claim) Try(alloc *Allocator) bool { return true } +func (c *claim) DeniedBy(alloc *Allocator, owner router.PeerName) { + name, found := alloc.nicknames[owner] + if found { + name = " (" + name + ")" + } + c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name)) +} + func (c *claim) Cancel() { c.sendResult(fmt.Errorf("Operation cancelled.")) }