.

go/neo/connection.go

@@ -26,10 +26,12 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"reflect"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"lab.nexedi.com/kirr/neo/go/xcommon/xnet"
 )
@@ -92,14 +94,17 @@ type Conn struct {
 	downOnce  sync.Once	// shutdown may be called by both Close and nodelink.shutdown
 
 	rxerrOnce sync.Once     // rx error is reported only once - then it is link down or closed
-	closed    uint32        // 1 if Close was called
-				// 2 if this is temp. Conn created to reply "connection refused"
+	closed    int32         // 1 if Close was called or "connection closed" entry
+				// incremented during every replyNoConn() in progress
+
+	errMsg	  *Error	// error message for replyNoConn
 }
 
 
 var ErrLinkClosed   = errors.New("node link is closed")	// operations on closed NodeLink
 var ErrLinkDown     = errors.New("node link is down")	// e.g. due to IO error
 var ErrLinkNoListen = errors.New("node link is not listening for incoming connections")
+var ErrLinkManyConn = errors.New("too many opened connections")
 var ErrClosedConn   = errors.New("connection is closed")
 
 // XXX unify LinkError & ConnError -> NetError?
@@ -218,8 +223,24 @@ func (nl *NodeLink) NewConn() (*Conn, error) {
 		}
 		return nil, nl.err("newconn", ErrLinkDown)
 	}
+
+	// nextConnId could wrap around uint32 limits - find first free slot to
+	// not blindly replace existing connection
+	for i := uint32(0) ;; i++ {
+		_, exists := nl.connTab[nl.nextConnId]
+		if !exists {
+			break
+		}
+		nl.nextConnId += 2
+
+		if i > math.MaxUint32 / 2 {
+			return nil, nl.err("newconn", ErrLinkManyConn)
+		}
+	}
+
 	c := nl.newConn(nl.nextConnId)
 	nl.nextConnId += 2
+
 	return c, nil
 }
 
@@ -274,6 +295,9 @@ func (c *Conn) shutdown() {
 	})
 }
 
+
+var connKeepClosed = 1*time.Minute
+
 // Close closes connection.
 // Any blocked Send*() or Recv*() will be unblocked and return error
 //
@@ -282,18 +306,37 @@ func (c *Conn) shutdown() {
 //
 // It is safe to call Close several times.
 func (c *Conn) Close() error {
-	// adjust nodeLink.connTab to have special entry noting this connection is closed.
-	// this way serveRecv knows to reply errConnClosed over network if peer
-	// sends us something over this conn.
-	c.nodeLink.connMu.Lock()
-	if c.nodeLink.connTab != nil {
-		cc := c.nodeLink.newConn(c.connId)
-		atomic.StoreUint32(&cc.closed, 1)
-		// note cc.down stays not closed so that send can work
+	nl := c.nodeLink
+
+	// adjust nodeLink.connTab
+	nl.connMu.Lock()
+	if nl.connTab != nil {
+		// connection was initiated by us - simply delete - we always
+		// know if a packet comes to such connection it is closed.
+		if c.connId == nl.nextConnId % 2 {
+			delete(nl.connTab, c.connId)
+
+		// connection was initiated by peer which we accepted - put special
+		// "closed" connection into connTab entry for some time to reply
+		// "connection closed" if another packet comes to it.
+		} else {
+			cc := nl.newConn(c.connId)
+			// 1 so that cc is not freed by replyNoConn
+			atomic.StoreInt32(&cc.closed, 1)
+			// NOTE cc.down stays not closed so Send could work
+			time.AfterFunc(connKeepClosed, func() {
+				nl.connMu.Lock()
+				delete(nl.connTab, cc.connId)
+				nl.connMu.Unlock()
+
+				cc.shutdown()
+			})
 		}
-	c.nodeLink.connMu.Unlock()
 
-	atomic.StoreUint32(&c.closed, 1)
+	}
+	nl.connMu.Unlock()
+
+	atomic.StoreInt32(&c.closed, 1)
 	c.shutdown()
 	return nil
 }
@@ -328,7 +371,7 @@ func (nl *NodeLink) Accept() (c *Conn, err error) {
 // errRecvShutdown returns appropriate error when c.down is found ready in recvPkt
 func (c *Conn) errRecvShutdown() error {
 	switch {
-	case atomic.LoadUint32(&c.closed) != 0:
+	case atomic.LoadInt32(&c.closed) != 0:
 		return ErrClosedConn
 
 	case atomic.LoadUint32(&c.nodeLink.closed) != 0:
@@ -393,36 +436,39 @@ func (nl *NodeLink) serveRecv() {
 		// connTab is never nil here - because shutdown before
 		// resetting it waits for us to finish.
 		conn := nl.connTab[connId]
+
+		fmt.Printf("RX .%d -> %v\n", connId, conn)
 		if conn == nil {
-			// XXX check connId is proper for peer originated streams
+			// "new" connection will be needed in all cases - e.g.
+			// temporarily to reply "connection refused"
+			conn = nl.newConn(connId)
+
+			fmt.Printf("connId: %d (%d)\n", connId, connId % 2)
+			fmt.Printf("nextConnId: %d (%d)\n", nl.nextConnId, nl.nextConnId % 2)
 
-			if nl.acceptq != nil {
+			// message with connid that should be initiated by us
+			if connId % 2 == nl.nextConnId % 2 {
+				conn.errMsg = errConnClosed
+
+			// message with connid for a stream initiated by peer
+			} else {
+				if nl.acceptq == nil {
+					conn.errMsg = errConnRefused
+				} else {
 					// we are accepting new incoming connection
-				conn = nl.newConn(connId)
 					accept = true
 				}
-		}
 
-		// connection not accepted - reply "connection refused"
-		if conn == nil {
-			conn = nl.newConn(connId)
-			atomic.StoreUint32(&conn.closed, 2)
-			// NOTE conn.down stays not closed so that Send can work
-			nl.connMu.Unlock()
-			go conn.replyNoConn(errConnRefused)
-			continue
+				fmt.Println("ZZZ", conn.errMsg, accept)
+			}
 		}
 
-		switch atomic.LoadUint32(&conn.closed) {
-		case 1:
-			// connection closed - reply "connection closed"
-			nl.connMu.Unlock()
-			go conn.replyNoConn(errConnClosed)
-			continue
-
-		case 2:
-			// "connection refused" reply is already in progress
+		// we are not accepting packet in any way
+		if conn.errMsg != nil {
+			fmt.Printf(".%d EMSG: %v\n", connId, conn.errMsg)
+			atomic.AddInt32(&conn.closed, 1)
 			nl.connMu.Unlock()
+			go conn.replyNoConn()
 			continue
 		}
 
@@ -468,15 +514,18 @@ var errConnClosed  = &Error{PROTOCOL_ERROR, "connection closed"}
 var errConnRefused = &Error{PROTOCOL_ERROR, "connection refused"}
 
 // replyNoConn sends error message to peer when a packet was sent to closed / nonexistent connection
-func (c *Conn) replyNoConn(e Msg) {
-	c.Send(e) // ignore errors
+// and removes connection from nodeLink connTab if ekeep==false.
+//func (c *Conn) replyNoConn(e Msg, ekeep bool) {
+func (c *Conn) replyNoConn() {
+	c.Send(c.errMsg) // ignore errors
 
-	// remove connTab entry if it was temporary conn created only to send errConnRefused
-	if e == errConnRefused {
+	// remove connTab entry - if all users of this temporary conn created
+	// only to send the error are now gone.
 	c.nodeLink.connMu.Lock()
+	if atomic.AddInt32(&c.closed, -1) == 0 {
 		delete(c.nodeLink.connTab, c.connId)
-		c.nodeLink.connMu.Unlock()
 	}
+	c.nodeLink.connMu.Unlock()
 }
 
 // ---- transmit ----
@@ -490,7 +539,7 @@ type txReq struct {
 // errSendShutdown returns appropriate error when c.down is found ready in Send
 func (c *Conn) errSendShutdown() error {
 	switch {
-	case atomic.LoadUint32(&c.closed) != 0:
+	case atomic.LoadInt32(&c.closed) != 0:
 		return ErrClosedConn
 
 	// the only other error possible besides Conn being .Close()'ed is that
@@ -586,7 +635,7 @@ const dumpio = true
 func (nl *NodeLink) sendPkt(pkt *PktBuf) error {
 	if dumpio {
 		// XXX -> log
-		fmt.Printf("%v > %v: %v\n", nl.peerLink.LocalAddr(), nl.peerLink.RemoteAddr(), pkt)
+		fmt.Printf("%v > %v: %v\n", nl.peerLink.LocalAddr(), nl.peerLink.RemoteAddr(), pkt.Dump())
 		//defer fmt.Printf("\t-> sendPkt err: %v\n", err)
 	}
 
@@ -639,7 +688,7 @@ func (nl *NodeLink) recvPkt() (*PktBuf, error) {
 
 	if dumpio {
 		// XXX -> log
-		fmt.Printf("%v < %v: %v\n", nl.peerLink.LocalAddr(), nl.peerLink.RemoteAddr(), pkt)
+		fmt.Printf("%v < %v: %v\n", nl.peerLink.LocalAddr(), nl.peerLink.RemoteAddr(), pkt.Dump())
 	}
 
 	return pkt, nil

go/neo/connection_test.go

@@ -32,6 +32,8 @@ import (
 
 	"lab.nexedi.com/kirr/go123/exc"
 	"lab.nexedi.com/kirr/go123/xerr"
+
+        "github.com/kylelemons/godebug/pretty"
 )
 
 func xclose(c io.Closer) {
@@ -121,12 +123,20 @@ func xverifyPkt(pkt *PktBuf, connid uint32, msgcode uint16, payload []byte) {
 		errv.Appendf("header: unexpected msglen %v  (want %v)", ntoh32(h.MsgLen), len(payload))
 	}
 	if !bytes.Equal(pkt.Payload(), payload) {
-		errv.Appendf("payload differ")
+		errv.Appendf("payload differ:\n%s",
+			pretty.Compare(string(payload), string(pkt.Payload())))
 	}
 
 	exc.Raiseif( errv.Err() )
 }
 
+// Verify PktBuf to match expected message
+func xverifyMsg(pkt *PktBuf, connid uint32, msg Msg) {
+	data := make([]byte, msg.neoMsgEncodedLen())
+	msg.neoMsgEncode(data)
+	xverifyPkt(pkt, connid, msg.neoMsgCode(), data)
+}
+
 // delay a bit
 // needed e.g. to test Close interaction with waiting read or write
 // (we cannot easily sync and make sure e.g. read is started and became asleep)
@@ -266,7 +276,7 @@ func TestNodeLink(t *testing.T) {
 	xwait(wgclose)
 
 
-	// Test connections on top of nodelink
+	// ---- connections on top of nodelink ----
 
 	// Close vs recvPkt
 	nl1, nl2 = _nodeLinkPipe(0, linkNoRecvSend)
@@ -439,6 +449,8 @@ func TestNodeLink(t *testing.T) {
 	}
 
 
+	println("\n---------------------\n")
+
 	// Conn accept + exchange
 	nl1, nl2 = nodeLinkPipe()
 	wg = &xsync.WorkGroup{}
@@ -457,7 +469,24 @@ func TestNodeLink(t *testing.T) {
 		xsendPkt(c, mkpkt(36, []byte("pong2")))
 
 		xclose(c)
+
+		println("B.111")
+
+		// "connection refused" when trying to connect to not-listening peer
+		c = xnewconn(nl2) // XXX should get error here?
+		xsendPkt(c, mkpkt(38, []byte("pong3")))
+		pkt = xrecvPkt(c)
+		xverifyMsg(pkt, c.connId, errConnRefused)
+		println("B.222")
+		xsendPkt(c, mkpkt(40, []byte("pong4"))) // once again
+		pkt = xrecvPkt(c)
+		xverifyMsg(pkt, c.connId, errConnRefused)
+		println("B.333")
+
+		xclose(c)
+
 	})
+	println("A.111")
 	c = xnewconn(nl1)
 	xsendPkt(c, mkpkt(33, []byte("ping")))
 	pkt = xrecvPkt(c)
@@ -467,6 +496,22 @@ func TestNodeLink(t *testing.T) {
 	xverifyPkt(pkt, c.connId, 36, []byte("pong2"))
 	xwait(wg)
 
+	println()
+	println()
+	println("A.222")
+	// "connection closed" after peer closed its end
+	xsendPkt(c, mkpkt(37, []byte("ping3")))
+	println("A.qqq")
+	pkt = xrecvPkt(c)
+	xverifyMsg(pkt, c.connId, errConnClosed)
+	println("A.zzz")
+	xsendPkt(c, mkpkt(39, []byte("ping4"))) // once again
+	pkt = xrecvPkt(c)
+	xverifyMsg(pkt, c.connId, errConnClosed)
+	// XXX also should get EOF on recv
+
+	println("A.333")
+
 	xclose(c)
 	xclose(nl1)
 	xclose(nl2)