diff --git a/pkg/tcpip/link/sharedmem/pipe/rx.go b/pkg/tcpip/link/sharedmem/pipe/rx.go index 1f44f5f14..b4d2d1e74 100644 --- a/pkg/tcpip/link/sharedmem/pipe/rx.go +++ b/pkg/tcpip/link/sharedmem/pipe/rx.go @@ -54,7 +54,15 @@ func (r *Rx) Pull() []byte { // Check if this is a wrapping slot. If that's the case, it carries no // data, so we just skip it and try again from the first slot. if int64(newHead-headWrap) >= 0 { - if int64(newHead-headWrap) > int64(jump) || newHead&offsetMask != 0 { + // If newHead passes the tail, the pipe is either damaged or the + // RX view of the pipe has completely wrapped without an + // intervening flush. + if int64(newHead-(r.tail+jump)) > 0 { + return nil + } + // The pipe is damaged if newHead doesn't point to the start of + // the ring. + if newHead&offsetMask != 0 { return nil } diff --git a/pkg/tcpip/link/sharedmem/pipe/tx.go b/pkg/tcpip/link/sharedmem/pipe/tx.go index 9841eb231..5ab0e211e 100644 --- a/pkg/tcpip/link/sharedmem/pipe/tx.go +++ b/pkg/tcpip/link/sharedmem/pipe/tx.go @@ -61,6 +61,9 @@ func (t *Tx) Push(payloadSize uint64) []byte { return nil } + // True if TxPipe currently has a pushed message, i.e., it is not + // Flush()'ed. + messageAhead := t.next != t.tail totalLen := payloadToSlotSize(payloadSize) newNext := t.next + totalLen nextWrap := (t.next & revolutionMask) | uint64(len(t.p.buffer)) @@ -69,21 +72,19 @@ func (t *Tx) Push(payloadSize uint64) []byte { // slot, then try to add the actual slot to the front of the // pipe. newNext = (newNext & revolutionMask) + jump - wrappingPayloadSize := slotToPayloadSize(newNext - t.next) if !t.reclaim(newNext) { return nil } - + wrappingPayloadSize := slotToPayloadSize(newNext - t.next) oldNext := t.next t.next = newNext - if oldNext != t.tail { + if messageAhead { t.p.write(oldNext, wrappingPayloadSize) } else { t.tailHeader = wrappingPayloadSize t.Flush() } - - newNext += totalLen + return t.Push(payloadSize) } // Check that we have enough room for the buffer. @@ -91,7 +92,7 @@ func (t *Tx) Push(payloadSize uint64) []byte { return nil } - if t.next != t.tail { + if messageAhead { t.p.write(t.next, payloadSize) } else { t.tailHeader = payloadSize