tcp.Conn:add Flush method and more lock protection

examples/xnet/main.go

@@ -297,11 +297,16 @@ func run() (err error) {
 		timeHTTPSend := timer("send HTTP request")
 		conn.SetDeadline(time.Now().Add(internetTimeout))
 		_, err = conn.Write(req)
-		timeHTTPSend()
-		timeHTTPRcv := timer("recv http request")
 		if err != nil {
 			return err
 		}
+		err = conn.Flush()
+		if err != nil {
+			return err
+		}
+		timeHTTPSend()
+		timeHTTPRcv := timer("recv http request")
+
 		rxbuf := make([]byte, 2048)
 
 		var page []byte

tcp/conn.go

@@ -93,6 +93,14 @@ func (conn *Conn) BufferedInput() int {
 	return conn.h.BufferedInput()
 }
 
+// BufferedUnsent returns the number of bytes in the socket's transmit(output) buffer
+// that has yet to be sent.
+func (conn *Conn) BufferedUnsent() int {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	return conn.h.BufferedUnsent()
+}
+
 func (conn *Conn) AvailableInput() int {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
@@ -174,13 +182,12 @@ func (conn *Conn) InternalHandler() *Handler {
 
 // Write writes argument data to the TCPConns's output buffer which is queued to be sent.
 func (conn *Conn) Write(b []byte) (int, error) {
-	err := conn.checkPipeOpen()
+	connid, err := conn.lockPipeConnID()
 	if err != nil {
 		return 0, err
 	}
 	plen := len(b)
 	conn.trace("TCPConn.Write:start")
-	connid := conn.h.ConnectionID()
 	if conn.deadlineExceeded(&conn.wdead) {
 		return 0, errDeadlineExceeded
 	} else if plen == 0 {
@@ -189,10 +196,8 @@ func (conn *Conn) Write(b []byte) (int, error) {
 	backoff := internal.NewBackoff(internal.BackoffTCPConn)
 	n := 0
 	for {
-		if conn.abortErr != nil {
+		if err := conn.checkPipe(connid, &conn.wdead); err != nil {
-			return n, conn.abortErr
+			return 0, err
-		} else if connid != conn.h.ConnectionID() {
-			return n, net.ErrClosed
 		}
 		conn.mu.Lock()
 		ngot, _ := conn.h.Write(b)
@@ -215,24 +220,38 @@ func (conn *Conn) Write(b []byte) (int, error) {
 	return n, nil
 }
 
+func (conn *Conn) Flush() error {
+	connid, err := conn.lockPipeConnID()
+	if err != nil {
+		return err
+	}
+	if conn.deadlineExceeded(&conn.wdead) {
+		return errDeadlineExceeded
+	} else if conn.BufferedUnsent() == 0 {
+		return nil
+	}
+	backoff := internal.NewBackoff(internal.BackoffTCPConn)
+	for conn.BufferedUnsent() != 0 {
+		if err := conn.checkPipe(connid, &conn.wdead); err != nil {
+			return err
+		}
+		backoff.Miss()
+	}
+	return nil
+}
+
 // Read reads data from the socket's input buffer. If the buffer is empty,
 // Read will block until data is available or connection closes.
 func (conn *Conn) Read(b []byte) (int, error) {
-	err := conn.checkPipeOpen()
+	connid, err := conn.lockPipeConnID()
 	if err != nil {
 		return 0, err
 	}
 	conn.trace("TCPConn.Read:start")
-	connid := conn.h.ConnectionID()
 	backoff := internal.NewBackoff(internal.BackoffTCPConn)
 	for conn.h.BufferedInput() == 0 && conn.State() == StateEstablished {
-		if conn.abortErr != nil {
+		if err := conn.checkPipe(connid, &conn.rdead); err != nil {
-			return 0, conn.abortErr
+			return 0, err
-		} else if connid != conn.h.ConnectionID() {
-			return 0, net.ErrClosed
-		}
-		if conn.deadlineExceeded(&conn.rdead) {
-			return 0, errDeadlineExceeded
 		}
 		backoff.Miss()
 	}
@@ -242,6 +261,29 @@ func (conn *Conn) Read(b []byte) (int, error) {
 	return n, err
 }
 
+func (conn *Conn) lockPipeConnID() (uint64, error) {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	err := conn.checkPipeOpen()
+	if err != nil {
+		return 0, err
+	}
+	return conn.h.connid, nil
+}
+
+func (conn *Conn) checkPipe(connID uint64, deadline *time.Time) (err error) {
+	conn.mu.Lock()
+	defer conn.mu.Unlock()
+	if conn.abortErr != nil {
+		err = conn.abortErr
+	} else if connID != conn.h.connid {
+		err = net.ErrClosed
+	} else if !deadline.IsZero() && time.Since(*deadline) > 0 {
+		err = errDeadlineExceeded
+	}
+	return nil
+}
+
 func (conn *Conn) checkPipeOpen() error {
 	if conn.abortErr != nil {
 		return conn.abortErr

tcp/handler.go

@@ -228,7 +228,7 @@ func (h *Handler) Send(b []byte) (int, error) {
 	if err != nil {
 		return 0, err
 	}
-	buffered := h.bufTx.Buffered()
+	buffered := h.bufTx.BufferedUnsent()
 	if buffered == 0 && h.closing {
 		// If Close called and no more data to be sent, terminate connection!
 		h.closing = false
@@ -328,6 +328,15 @@ func (h *Handler) BufferedInput() int {
 	return h.bufRx.Buffered()
 }
 
+// BufferedUnsent returns the number of bytes in the socket's transmit(output) buffer
+// that has yet to be sent.
+func (h *Handler) BufferedUnsent() int {
+	if h.State().IsClosed() {
+		return 0
+	}
+	return h.bufTx.BufferedUnsent()
+}
+
 // AvailableOutput returns amount of bytes available to write to output
 // before [Handler.Write] returns an error.
 func (h *Handler) AvailableOutput() int {

tcp/txqueue.go

@@ -90,8 +90,8 @@ func (rtx *ringTx) Free() int {
 	return r.Free()
 }
 
-// Buffered returns the amount of written but unsent bytes.
+// BufferedUnsent returns the amount of written but unsent bytes.
-func (rtx *ringTx) Buffered() int {
+func (rtx *ringTx) BufferedUnsent() int {
 	r, _ := rtx.unsentRing()
 	return r.Buffered()
 }

tcp/txqueue_test.go

@@ -305,7 +305,7 @@ func TestTxQueue(t *testing.T) {
 						seq := currentAck
 						currentAck = Add(currentAck, Size(len(msg)))
 						operateOnRing(t, &rtx, msg, readBuf[:], aux[:], seq, &currentAck)
-						buffered := rtx.Buffered()
+						buffered := rtx.BufferedUnsent()
 						if buffered != 0 {
 							t.Fatalf("msg%d: want no buffered data after transaction, got %d", imsg, buffered)
 						}
@@ -334,14 +334,14 @@ func TestTxQueue(t *testing.T) {
 							return
 						}
 						expectBuffered += len(msg)
-						buffered := rtx.Buffered()
+						buffered := rtx.BufferedUnsent()
 						if buffered != expectBuffered {
 							t.Fatalf("expected seq to not change during writes")
 						}
 						currentAck = Add(currentAck, Size(len(msg)))
 					}
 					sent := rtx.BufferedSent()
-					unsent := rtx.Buffered()
+					unsent := rtx.BufferedUnsent()
 					wantUnsent := int(Add(currentAck, -startAck))
 					if unsent != wantUnsent {
 						t.Fatalf("want %d data buffered, got %d", wantUnsent, unsent)
@@ -349,7 +349,7 @@ func TestTxQueue(t *testing.T) {
 						t.Fatalf("want no data sent, got %d", sent)
 					}
 					operateOnRing(t, &rtx, nil, readBuf[:], aux[:], 0, &currentAck)
-					unsent = rtx.Buffered()
+					unsent = rtx.BufferedUnsent()
 					if unsent != 0 {
 						t.Fatalf("expected all data to be sent after ack of most recent packet, %d", unsent)
 					} else if rtx.BufferedSent() != 0 {
@@ -375,7 +375,7 @@ func TestTxQueue(t *testing.T) {
 					// Send all bytes over wire.
 					currentSeq := Value(startAck)
 					datalens = datalens[:0]
-					for rtx.Buffered() != 0 {
+					for rtx.BufferedUnsent() != 0 {
 						nbytes := rng.Intn(maxPacketSize-minBufferSize) + minBufferSize
 						n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq)
 						if err != nil {
@@ -445,7 +445,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) {
 
 	free := rtx.Free()
 	sent := rtx.BufferedSent()
-	unsent := rtx.Buffered()
+	unsent := rtx.BufferedUnsent()
 	sz := rtx.Size()
 	gotSz := free + sent + unsent
 	if gotSz != sz {
@@ -553,14 +553,14 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 
 	if len(write) != 0 {
 		testQueueSanity(t, rtx)
-		preBuffered := rtx.Buffered()
+		preBuffered := rtx.BufferedUnsent()
 		n, err := rtx.Write(write)
 		if err != nil && wantWritten > 0 {
 			t.Errorf("error writing packet: %s", err)
 		} else if n != wantWritten {
 			t.Errorf("want %d written, got %d", wantWritten, n)
 		}
-		newBuffered := rtx.Buffered()
+		newBuffered := rtx.BufferedUnsent()
 		gotWritten := newBuffered - preBuffered
 		if gotWritten != wantWritten {
 			t.Errorf("expected %d data written, got %d", wantWritten, gotWritten)
@@ -570,7 +570,7 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 	if !t.Failed() && len(readPacket) != 0 {
 		testQueueSanity(t, rtx)
 		preSent := rtx.BufferedSent()
-		canRead := rtx.Buffered()
+		canRead := rtx.BufferedUnsent()
 		wantRead := min(canRead, len(readPacket))
 		if wantRead != len(wantBufRead) {
 			t.Fatalf("miscalculated expect read %d != %d", wantRead, len(wantBufRead))