Add SpaceRequestDenied message

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

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."))
 }