Merge pull request #1104 from weaveworks/1083-async-claim

Don't wait to claim addresses from IPAM

ipam/allocator.go

@@ -202,10 +202,9 @@ func (alloc *Allocator) Lookup(ident string, r address.Range) (address.Address,
 }
 
 // Claim an address that we think we should own (Sync)
-func (alloc *Allocator) Claim(ident string, addr address.Address, cancelChan <-chan bool) error {
+func (alloc *Allocator) Claim(ident string, addr address.Address) error {
 	resultChan := make(chan error)
-	op := &claim{resultChan: resultChan, ident: ident, addr: addr,
-		hasBeenCancelled: hasBeenCancelled(cancelChan)}
+	op := &claim{resultChan: resultChan, ident: ident, addr: addr}
 	alloc.doOperation(op, &alloc.pendingClaims)
 	return <-resultChan
 }

ipam/allocator_test.go

@@ -143,7 +143,7 @@ func TestAllocatorClaim(t *testing.T) {
 	alloc.claimRingForTesting()
 	addr1, _ := address.ParseIP(testAddr1)
 
-	err := alloc.Claim(container3, addr1, nil)
+	err := alloc.Claim(container3, addr1)
 	wt.AssertNoErr(t, err)
 	// Check we get this address back if we try an allocate
 	addr3, _ := alloc.Allocate(container3, subnet, nil)

ipam/claim.go

@@ -3,28 +3,35 @@ package ipam
 import (
 	"fmt"
 
+	"github.com/weaveworks/weave/common"
 	"github.com/weaveworks/weave/ipam/address"
 	"github.com/weaveworks/weave/router"
 )
 
 type claim struct {
-	resultChan       chan<- error
-	ident            string
-	addr             address.Address
-	hasBeenCancelled func() bool
+	resultChan chan<- error
+	ident      string
+	addr       address.Address
+}
+
+func (c *claim) sendResult(result error) {
+	// Make sure we only send a result once, since listener stops listening after that
+	if c.resultChan != nil {
+		c.resultChan <- result
+		close(c.resultChan)
+		c.resultChan = nil
+	}
+	if result != nil {
+		common.Error.Println("[allocator] " + result.Error())
+	}
 }
 
 // Try returns true for success (or failure), false if we need to try again later
 func (c *claim) Try(alloc *Allocator) bool {
-	if (c.hasBeenCancelled)() {
-		c.Cancel()
-		return true
-	}
-
 	if !alloc.ring.Contains(c.addr) {
 		// Address not within our universe; assume user knows what they are doing
 		alloc.infof("Ignored address %s claimed by %s - not in our universe\n", c.addr, c.ident)
-		c.resultChan <- nil
+		c.sendResult(nil)
 		return true
 	}
 
@@ -34,6 +41,7 @@ func (c *claim) Try(alloc *Allocator) bool {
 	owner := alloc.ring.Owner(c.addr)
 	if owner == router.UnknownPeerName {
 		alloc.infof("Ring is empty; will try later.\n", c.addr, owner)
+		c.sendResult(nil) // don't make the caller wait
 		return false
 	}
 	if owner != alloc.ourName {
@@ -41,34 +49,34 @@ func (c *claim) Try(alloc *Allocator) bool {
 		if found {
 			name = " (" + name + ")"
 		}
-		c.resultChan <- fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name)
+		c.sendResult(fmt.Errorf("address %s is owned by other peer %s%s", c.addr.String(), owner, name))
 		return true
 	}
 	// We are the owner, check we haven't given it to another container
 	existingIdent := alloc.findOwner(c.addr)
 	if existingIdent == c.ident {
 		// same identifier is claiming same address; that's OK
-		c.resultChan <- nil
+		c.sendResult(nil)
 		return true
 	}
 	if existingIdent == "" {
 		err := alloc.space.Claim(c.addr)
 		if err != nil {
-			c.resultChan <- err
+			c.sendResult(err)
 			return true
 		}
 		alloc.debugln("Claimed", c.addr, "for", c.ident)
 		alloc.addOwned(c.ident, c.addr)
-		c.resultChan <- nil
+		c.sendResult(nil)
 		return true
 	}
 	// Addr already owned by container on this machine
-	c.resultChan <- fmt.Errorf("address %s is already owned by %s", c.addr.String(), existingIdent)
+	c.sendResult(fmt.Errorf("address %s is already owned by %s", c.addr.String(), existingIdent))
 	return true
 }
 
 func (c *claim) Cancel() {
-	c.resultChan <- fmt.Errorf("Operation cancelled.")
+	c.sendResult(fmt.Errorf("Operation cancelled."))
 }
 
 func (c *claim) String() string {

ipam/http.go

@@ -19,14 +19,13 @@ func badRequest(w http.ResponseWriter, err error) {
 // HandleHTTP wires up ipams HTTP endpoints to the provided mux.
 func (alloc *Allocator) HandleHTTP(router *mux.Router, defaultSubnet address.CIDR, dockerCli *docker.Client) {
 	router.Methods("PUT").Path("/ip/{id}/{ip}").HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		closedChan := w.(http.CloseNotifier).CloseNotify()
 		vars := mux.Vars(r)
 		ident := vars["id"]
 		ipStr := vars["ip"]
 		if ip, err := address.ParseIP(ipStr); err != nil {
 			badRequest(w, err)
 			return
-		} else if err := alloc.Claim(ident, ip, closedChan); err != nil {
+		} else if err := alloc.Claim(ident, ip); err != nil {
 			badRequest(w, fmt.Errorf("Unable to claim: %s", err))
 			return
 		}