Merge pull request #3763 from weaveworks/no-lock-round-send

Move send of ipsec message outside lock

net/ipsec/ipsec.go

@@ -85,7 +85,7 @@ func New(log *logrus.Logger) (*IPSec, error) {
 
 // InitSALocal initializes inbound ipsec from remotePeer and triggers
 // the initialization on remotePeer.
-func (ipsec *IPSec) InitSALocal(localPeer, remotePeer mesh.PeerName, connUID uint64, localIP, remoteIP net.IP, udpPort int, sessionKey *[32]byte, initRemote func([]byte) error) error {
+func (ipsec *IPSec) InitSALocal(localPeer, remotePeer mesh.PeerName, connUID uint64, localIP, remoteIP net.IP, udpPort int, sessionKey *[32]byte) ([]byte, error) {
 	// ID of inbound SPI
 	spiID := getSPIId(remotePeer, localPeer, connUID)
 
@@ -95,17 +95,17 @@ func (ipsec *IPSec) InitSALocal(localPeer, remotePeer mesh.PeerName, connUID uin
 	// Derive SA key
 	nonce, err := genNonce()
 	if err != nil {
-		return errors.Wrap(err, "generate nonce")
+		return nil, errors.Wrap(err, "generate nonce")
 	}
 	key, err := deriveKey(sessionKey[:], nonce, localPeer)
 	if err != nil {
-		return errors.Wrap(err, "derive key")
+		return nil, errors.Wrap(err, "derive key")
 	}
 
 	// Allocate SA
 	sa, err := netlink.XfrmStateAllocSpi(xfrmAllocSpiState(remoteIP, localIP))
 	if err != nil {
-		return errors.Wrap(err, fmt.Sprintf("ip xfrm state allocspi (in, %s, %s)", remoteIP, localIP))
+		return nil, errors.Wrap(err, fmt.Sprintf("ip xfrm state allocspi (in, %s, %s)", remoteIP, localIP))
 	}
 
 	// Use SPI generated by the kernel. The kernel ensures {dstIP, spi} to
@@ -117,28 +117,24 @@ func (ipsec *IPSec) InitSALocal(localPeer, remotePeer mesh.PeerName, connUID uin
 	// Create SA
 	if sa, err := xfrmState(remoteIP, localIP, spi, false, key); err == nil {
 		if err := netlink.XfrmStateUpdate(sa); err != nil {
-			return errors.Wrap(err, fmt.Sprintf("xfrm state update (in, %s, %s, 0x%x)", sa.Src, sa.Dst, sa.Spi))
+			return nil, errors.Wrap(err, fmt.Sprintf("xfrm state update (in, %s, %s, 0x%x)", sa.Src, sa.Dst, sa.Spi))
 		}
 	} else {
-		return errors.Wrap(err, "new xfrm state (in)")
+		return nil, errors.Wrap(err, "new xfrm state (in)")
 	}
 
 	// Install iptables rules
 	if err := ipsec.installDropNonEncrypted(localIP, remoteIP, udpPort, spi); err != nil {
-		return errors.Wrap(err, fmt.Sprintf("install protecting rules (%s, %s, %d, 0x%x)", localIP, remoteIP, udpPort, spi))
+		return nil, errors.Wrap(err, fmt.Sprintf("install protecting rules (%s, %s, %d, 0x%x)", localIP, remoteIP, udpPort, spi))
 	}
 
 	si := spiInfo{spi: spi, isDirOut: false}
 	ipsec.spiInfo[spiID] = si
 	ipsec.spis[spi] = &si
 
-	// Trigger the initialization on the remote peer
+	// Generate the message to trigger initialization on the remote peer
 	msg := &msgInitSARemote{nonce, spi}
-	if err := initRemote(msg.serialize()); err != nil {
-		return errors.Wrap(err, "send InitSARemote")
-	}
-
-	return nil
+	return msg.serialize(), nil
 }
 
 // InitSARemote initializes outbound ipsec to remotePeer.

router/fastdp.go

@@ -482,7 +482,7 @@ func (fastdp *FastDatapath) getVxlanVportID(udpPort int) (odp.VportID, error) {
 			// The netdev interface is down, so most likely bringing it up
 			// has failed due to the UDP port being in use.
 			if err := fastdp.dp.DeleteVport(vxlanVportID); err != nil {
-				log.Warning("Unable to remove vxlan vport %d: %s", vxlanVportID, err)
+				log.Warningf("Unable to remove vxlan vport %d: %s", vxlanVportID, err)
 			}
 			return 0, odp.NetlinkError(syscall.EADDRINUSE)
 		}
@@ -654,27 +654,26 @@ func (fwd *fastDatapathForwarder) logPrefix() string {
 
 func (fwd *fastDatapathForwarder) Confirm() {
 	fwd.lock.Lock()
-	defer fwd.lock.Unlock()
 
 	if fwd.confirmed {
 		log.Fatal(fwd.logPrefix(), "already confirmed")
 	}
 
+	var controlMsg []byte
 	if fwd.fastdp.ipsec != nil && fwd.sessionKey != nil {
 		fwd.isEncrypted = true
 		log.Info("Setting up IPsec between ", fwd.fastdp.localPeer, " and ", fwd.remotePeer)
-		err := fwd.fastdp.ipsec.InitSALocal(
+		var err error
+		controlMsg, err = fwd.fastdp.ipsec.InitSALocal(
 			fwd.fastdp.localPeer.Name, fwd.remotePeer.Name, fwd.connUID,
 			net.IP(fwd.localIP[:]), fwd.remoteAddr.IP,
 			fwd.remoteAddr.Port,
 			fwd.sessionKey,
-			func(msg []byte) error {
-				return fwd.sendControlMsg(FastDatapathCryptoInitSARemote, msg)
-			},
 		)
 		if err != nil {
 			log.Error(fwd.logPrefix(), "ipsec init SA local failed: ", err)
 			fwd.handleError(err)
+			fwd.lock.Unlock()
 			return
 		}
 	}
@@ -692,6 +691,15 @@ func (fwd *fastDatapathForwarder) Confirm() {
 	}
 
 	fwd.heartbeatTimeout = time.NewTimer(HeartbeatTimeout)
+
+	fwd.lock.Unlock() // unlock before calling send() which may block
+
+	if err := fwd.sendControlMsg(FastDatapathCryptoInitSARemote, controlMsg); err != nil {
+		log.Error(fwd.logPrefix(), "ipsec send InitSARemote failed: ", err)
+		fwd.handleError(err)
+		return
+	}
+
 	go fwd.doHeartbeats()
 }