diff --git a/examples/xnet/main.go b/examples/xnet/main.go index a26f017..f54a755 100644 --- a/examples/xnet/main.go +++ b/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 diff --git a/tcp/conn.go b/tcp/conn.go index c7577d6..e5e5580 100644 --- a/tcp/conn.go +++ b/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 { - return n, conn.abortErr - } else if connid != conn.h.ConnectionID() { - return n, net.ErrClosed + if err := conn.checkPipe(connid, &conn.wdead); err != nil { + return 0, err } 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 { - return 0, conn.abortErr - } else if connid != conn.h.ConnectionID() { - return 0, net.ErrClosed - } - if conn.deadlineExceeded(&conn.rdead) { - return 0, errDeadlineExceeded + if err := conn.checkPipe(connid, &conn.rdead); err != nil { + return 0, err } 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 diff --git a/tcp/handler.go b/tcp/handler.go index af9eb9e..155f447 100644 --- a/tcp/handler.go +++ b/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 { diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 337a294..f3e6904 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -90,8 +90,8 @@ func (rtx *ringTx) Free() int { return r.Free() } -// Buffered returns the amount of written but unsent bytes. -func (rtx *ringTx) Buffered() int { +// BufferedUnsent returns the amount of written but unsent bytes. +func (rtx *ringTx) BufferedUnsent() int { r, _ := rtx.unsentRing() return r.Buffered() } diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index a85c41c..3ee6043 100644 --- a/tcp/txqueue_test.go +++ b/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, ¤tAck) - 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, ¤tAck) - 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))