func (r *Rx) Dequeue(bufs []RxBuffer) ([]RxBuffer, uint32) {
	for {
		outBufs := bufs
		b := r.rx.Pull()
		if b == nil {
			return bufs, 0
		}
		if len(b) < sizeOfConsumedPacketHeader {
			log.Printf("Ignoring packet header: size (%v) is less than header size (%v)", len(b), sizeOfConsumedPacketHeader)
			r.rx.Flush()
			continue
		}
		totalDataSize := binary.LittleEndian.Uint32(b[consumedPacketSize:])
		count := (len(b) - sizeOfConsumedPacketHeader) / sizeOfConsumedBuffer
		offset := sizeOfConsumedPacketHeader
		buffersSize := uint32(0)
		for i := count; i > 0; i-- {
			s := binary.LittleEndian.Uint32(b[offset+consumedSize:])
			buffersSize += s
			if buffersSize < s {
				totalDataSize = 1
				buffersSize = 0
				break
			}
			outBufs = append(outBufs, RxBuffer{
				Offset: binary.LittleEndian.Uint64(b[offset+consumedOffset:]),
				Size:   s,
				ID:     binary.LittleEndian.Uint64(b[offset+consumedID:]),
			})
			offset += sizeOfConsumedBuffer
		}
		r.rx.Flush()
		if buffersSize < totalDataSize {
			log.Printf("Ignoring packet: actual data size (%v) less than expected size (%v)", buffersSize, totalDataSize)
			continue
		}
		return outBufs, totalDataSize
	}
}
Dequeue receives buffers that have been previously posted by PostBuffers()
and that have been filled by the peer and posted back.

This is similar to append() in that new buffers are appended to "bufs", with
reallocation only if "bufs" doesn't have enough capacity.
package queue
import (
	"encoding/binary"
	"sync/atomic"
	"github.com/google/netstack/tcpip/link/sharedmem/pipe"
	"log"
)
const (
	postedOffset           = 0
	postedSize             = 8
	postedRemainingInGroup = 12
	postedUserData         = 16
	postedID               = 24
	sizeOfPostedBuffer = 32
	consumedPacketSize     = 0
	consumedPacketReserved = 4
	sizeOfConsumedPacketHeader = 8
	consumedOffset   = 0
	consumedSize     = 8
	consumedUserData = 12
	consumedID       = 20
	sizeOfConsumedBuffer = 28
	eventFDUninitialized = 0
	eventFDDisabled      = 1
	eventFDEnabled       = 2
)
type RxBuffer struct {
	Offset   uint64
	Size     uint32
	ID       uint64
	UserData uint64
}
type Rx struct {
	tx                 pipe.Tx
	rx                 pipe.Rx
	sharedEventFDState *uint32
}
func (r *Rx) Init(tx, rx []byte, sharedEventFDState *uint32) {
	r.sharedEventFDState = sharedEventFDState
	r.tx.Init(tx)
	r.rx.Init(rx)
}
func (r *Rx) EnableNotification() {
	atomic.StoreUint32(r.sharedEventFDState, eventFDEnabled)
}
func (r *Rx) DisableNotification() {
	atomic.StoreUint32(r.sharedEventFDState, eventFDDisabled)
}
func (r *Rx) PostedBuffersLimit() uint64 {
	return r.tx.Capacity(sizeOfPostedBuffer)
}
func (r *Rx) PostBuffers(buffers []RxBuffer) bool {
	for i := range buffers {
		b := r.tx.Push(sizeOfPostedBuffer)
		if b == nil {
			r.tx.Abort()
			return false
		}
		pb := &buffers[i]
		binary.LittleEndian.PutUint64(b[postedOffset:], pb.Offset)
		binary.LittleEndian.PutUint32(b[postedSize:], pb.Size)
		binary.LittleEndian.PutUint32(b[postedRemainingInGroup:], 0)
		binary.LittleEndian.PutUint64(b[postedUserData:], pb.UserData)
		binary.LittleEndian.PutUint64(b[postedID:], pb.ID)
	}
	r.tx.Flush()
	return true
}