.

go/neo/connection.go

@@ -613,7 +613,7 @@ func (nl *NodeLink) serveRecv() {
 		// receive 1 packet
 		// XXX if nl.peerLink was just closed by tx->shutdown we'll get ErrNetClosing
 		pkt, err := nl.recvPkt()
-		//fmt.Printf("recvPkt -> %v, %v\n", pkt, err)
+		//fmt.Printf("\n%p recvPkt -> %v, %v\n", nl, pkt, err)
 		if err != nil {
 			// on IO error framing over peerLink becomes broken
 			// so we shut down node link and all connections over it.
@@ -650,6 +650,7 @@ func (nl *NodeLink) serveRecv() {
 
 		nl.connMu.Unlock()
 
+		//fmt.Printf("%p\tconn: %v\n", nl, conn)
 		if conn == nil {
 			// see ^^^ "message with connid that should be initiated by us"
 			go nl.replyNoConn(connId, errConnClosed)
@@ -671,6 +672,8 @@ func (nl *NodeLink) serveRecv() {
 		}
 		conn.rxqActive.Set(0)
 
+		//fmt.Printf("%p\tconn.rxdown: %v\taccept: %v\n", nl, rxdown, accept)
+
 
 		// conn exists but rx is down - "connection closed"
 		// (this cannot happen for newly accepted connection)
@@ -701,6 +704,7 @@ func (nl *NodeLink) serveRecv() {
 					axdown = true
 
 				case nl.acceptq <- conn:
+					//fmt.Printf("%p\t.acceptq <- conn  ok\n", nl)
 					// ok
 				}
 			}
@@ -713,6 +717,8 @@ func (nl *NodeLink) serveRecv() {
 				nl.connMu.Unlock()
 			}
 		}
+
+		//fmt.Printf("%p\tafter accept\n", nl)
 /*
 		// XXX goes away in favour of .rxdownFlag; reasons
 		// - no need to reallocate rxdown for light conn
@@ -758,9 +764,9 @@ var errConnRefused = &Error{PROTOCOL_ERROR, "connection refused"}
 
 // replyNoConn sends error message to peer when a packet was sent to closed / nonexistent connection
 func (link *NodeLink) replyNoConn(connId uint32, errMsg Msg) {
-	//fmt.Printf("%s .%d: -> replyNoConn %v\n", link, connId, c.errMsg)
+	//fmt.Printf("%s .%d: -> replyNoConn %v\n", link, connId, errMsg)
 	link.sendMsg(connId, errMsg) // ignore errors
-	//fmt.Printf("%s .%d: replyNoConn(%v) -> %v\n", link, connId, c.errMsg, err)
+	//fmt.Printf("%s .%d: replyNoConn(%v) -> %v\n", link, connId, errMsg, err)
 }
 
 // ---- transmit ----
@@ -1513,7 +1519,8 @@ func (link *NodeLink) Send1(msg Msg) error {
 	conn.downRX(errConnClosed)	// FIXME just new conn this way
 
 	err = conn.Send(msg)
-	conn.release()
+	//conn.release()	XXX temp
+	conn.Close()
 	return err
 }
 

go/neo/connection_test.go

@@ -37,8 +37,6 @@ import (
 
 	"github.com/kylelemons/godebug/pretty"
 	"github.com/pkg/errors"
-
-	//"runtime/debug"
 )
 
 func xclose(c io.Closer) {
@@ -763,55 +761,77 @@ func xSend1(l *NodeLink, msg Msg) {
 
 func xverifyMsg(msg1, msg2 Msg) {
 	if !reflect.DeepEqual(msg1, msg2) {
-		//debug.PrintStack()
 		exc.Raisef("messages differ:\n%s", pretty.Compare(msg1, msg2))
 	}
 }
 
 func TestRecv1Mode(t *testing.T) {
-	println("000")
+	//println("000")
 	// Send1
 	nl1, nl2 := nodeLinkPipe()
 	wg := &xsync.WorkGroup{}
+	sync := make(chan int)
 	wg.Gox(func() {
+		defer func() {
+			if e := recover(); e != nil {
+				panic(e)
+			}
+		}()
+
+		//println("X aaa")
 		c := xaccept(nl2)
+		//println("X aaa + 1")
 		msg := xRecv(c)
+		//println("X aaa + 2")
 		xverifyMsg(msg, &Ping{})
 		xSend(c, &Pong{})
+		//println("X aaa + 3")
 		msg = xRecv(c)
+		//println("X aaa + 4")
 		xverifyMsg(msg, errConnClosed)
 		xclose(c)
 
-		println("X zzz")
+		sync <- 1
+		//println("X zzz")
 
 		c = xaccept(nl2)
 		msg = xRecv(c)
-		println("X zzz + 1")
+		//fmt.Println("X zzz + 1", c, msg)
 		xverifyMsg(msg, &Error{ACK, "hello up there"})
 		xSend(c, &Error{ACK, "hello to you too"})
 		msg = xRecv(c)
-		println("X zzz + 2")
+		//println("X zzz + 2")
 		xverifyMsg(msg, errConnClosed)
-		println("X zzz + 3")
+		//println("X zzz + 3")
+		xclose(c)
 	})
 
-	println("aaa")
+	//println("aaa")
 	xSend1(nl1, &Ping{})
-	println("bbb")
+
+	// before next Send1 wait till peer receives errConnClosed from us
+	// otherwise peer could be already in accept while our errConnClosed is received
+	// and there is only one receiving thread there ^^^
+	<-sync
+	//println("bbb")
 	xSend1(nl1, &Error{ACK, "hello up there"})
-	println("ccc")
+	//println("ccc")
 	xwait(wg)
 
-	println("111")
+	//println("111\n")
 
 	// Recv1: further packets with same connid are rejected with "connection closed"
 	wg = &xsync.WorkGroup{}
 	wg.Gox(func() {
 		c := xnewconn(nl2)
 
+		//println("aaa")
 		xSend(c, &Ping{})
+		//println("bbb")
 		xSend(c, &Ping{})
+		//println("ccc")
 		msg := xRecv(c)
+		//println("ddd")
 		xverifyMsg(msg, errConnClosed)
 	})