bufferio

Documentation

Index

• Variables
• func DrainChannel(arrayChan chan *BufferArray)
• func GetExtensionProperty(conn net.Conn) map[string]uint64
• func GrowBufferWithZeroPadding(b []byte, size int) []byte
• func IsConnClosed(conn net.Conn) bool
• func PushToChannelMayEviction(ctx context.Context, channel chan *BufferArray, data *BufferArray, ...) error
• func PutBufferArray(ba *BufferArray)
• func PutBufferEntity(b *BufferEntity)
• func Relay(RuntimeContext context.Context, peerA io.ReadWriteCloser, ...) error
• func SupportsKeepAlive(conn net.Conn) bool
• type BufferArray
  • func GetBufferArray() *BufferArray
  • func ReceiveBatch(ctx context.Context, channel chan *BufferArray, idealEntityNum int) *BufferArray
  • func (reader *BufferArray) Data() [][]byte
  • func (reader *BufferArray) Discard(n int) (int, error)
  • func (reader *BufferArray) MergeFrom(a *BufferArray)
  • func (reader *BufferArray) Peek(capacity int) ([]byte, error)
  • func (reader *BufferArray) Pop() *BufferEntity
  • func (reader *BufferArray) Push(buffer *BufferEntity)
  • func (reader *BufferArray) Read(b []byte) (int, error)
  • func (reader *BufferArray) ReadByte() (byte, error)
  • func (ba *BufferArray) Reset()
  • func (reader *BufferArray) Stat() (totalBuckets int, totalBytes uint64)
  • func (reader *BufferArray) Swap(b *BufferArray)
  • func (reader *BufferArray) Write(b []byte) (int, error)
• type BufferEntity
  • func GetBufferEntity() *BufferEntity
  • func GetBufferEntityWithAtLeastCapacity(capacity int) *BufferEntity
  • func ReserveStartOffset(packet *BufferEntity, startOffset int) *BufferEntity
  • func (buffer *BufferEntity) Append(b []byte)
  • func (buffer *BufferEntity) Data() []byte
  • func (buffer *BufferEntity) Dump(b []byte) (int, bool)
  • func (buffer *BufferEntity) Load(b []byte)
  • func (buffer *BufferEntity) Swap(b []byte) []byte
  • func (buffer *BufferEntity) Write(b []byte) (int, error)
• type BufferIO
  • func CreateAsyncChannel(Context context.Context, Raw io.ReadWriteCloser, ...) BufferIO
  • func GetBuiltinBufferIO(v any) BufferIO
• type ExtendConn
• type KeepAliveSupport
• type PerformanceCounterReporter

Variables

var BiCopyInflight = atomic.Int64{}

Functions

func DrainChannel

func DrainChannel(arrayChan chan *BufferArray)

func GetExtensionProperty

func GetExtensionProperty(conn net.Conn) map[string]uint64

func GrowBufferWithZeroPadding

func GrowBufferWithZeroPadding(b []byte, size int) []byte

func IsConnClosed

func IsConnClosed(conn net.Conn) bool

func PushToChannelMayEviction

func PushToChannelMayEviction(
	ctx context.Context,
	channel chan *BufferArray,
	data *BufferArray,
	whenEvict func(ba *BufferArray),
	allowBestEffortEvict bool,
) error

func PutBufferArray

func PutBufferArray(ba *BufferArray)

func PutBufferEntity

func PutBufferEntity(b *BufferEntity)

func Relay

func Relay(RuntimeContext context.Context, peerA io.ReadWriteCloser, peerB io.ReadWriteCloser) error

func SupportsKeepAlive

func SupportsKeepAlive(conn net.Conn) bool

Types

type BufferArray

type BufferArray struct {
	// contains filtered or unexported fields
}

func GetBufferArray

func GetBufferArray() *BufferArray

func ReceiveBatch

func ReceiveBatch(ctx context.Context, channel chan *BufferArray, idealEntityNum int) *BufferArray

Returns nil if ctx is done or channel is closed.

func (*BufferArray) Data

func (reader *BufferArray) Data() [][]byte

Data is a reference to the buffer array.

func (*BufferArray) Discard

func (reader *BufferArray) Discard(n int) (int, error)

func (*BufferArray) MergeFrom

func (reader *BufferArray) MergeFrom(a *BufferArray)

Takes the ownership of `a`.

func (*BufferArray) Peek

func (reader *BufferArray) Peek(capacity int) ([]byte, error)

Returns next `capacity` bytes if current segment has enough bytes.

func (*BufferArray) Pop

func (reader *BufferArray) Pop() *BufferEntity

func (*BufferArray) Push

func (reader *BufferArray) Push(buffer *BufferEntity)

func (*BufferArray) Read

func (reader *BufferArray) Read(b []byte) (int, error)

func (*BufferArray) ReadByte

func (reader *BufferArray) ReadByte() (byte, error)

func (*BufferArray) Reset

func (ba *BufferArray) Reset()

func (*BufferArray) Stat

func (reader *BufferArray) Stat() (totalBuckets int, totalBytes uint64)

func (*BufferArray) Swap

func (reader *BufferArray) Swap(b *BufferArray)

func (*BufferArray) Write

func (reader *BufferArray) Write(b []byte) (int, error)

type BufferEntity

type BufferEntity struct {
	Start, End int
	Memory     []byte
}

func GetBufferEntity

func GetBufferEntity() *BufferEntity

func GetBufferEntityWithAtLeastCapacity

func GetBufferEntityWithAtLeastCapacity(capacity int) *BufferEntity

func ReserveStartOffset

func ReserveStartOffset(packet *BufferEntity, startOffset int) *BufferEntity

func (*BufferEntity) Append

func (buffer *BufferEntity) Append(b []byte)

func (*BufferEntity) Data

func (buffer *BufferEntity) Data() []byte

func (*BufferEntity) Dump

func (buffer *BufferEntity) Dump(b []byte) (int, bool)

Writes data into `b`, returns if the buffer becomes empty.

func (*BufferEntity) Load

func (buffer *BufferEntity) Load(b []byte)

Performs a copy of `b` and reset read status.

func (*BufferEntity) Swap

func (buffer *BufferEntity) Swap(b []byte) []byte

func (*BufferEntity) Write

func (buffer *BufferEntity) Write(b []byte) (int, error)

type BufferIO

type BufferIO interface {
	io.Closer

	// ConsumeBuffer consumes a buffer array.
	// This does not take the ownership of the array.
	ConsumeBuffer(bs *BufferArray) error
	ProduceBuffer(bs *BufferArray, reservedOffset int) error
	SupportBufferIO() bool
}

func CreateAsyncChannel

func CreateAsyncChannel(Context context.Context, Raw io.ReadWriteCloser, MaxEntitySize, QueueLength, ReservedStartOffset int) BufferIO

func GetBuiltinBufferIO

func GetBuiltinBufferIO(v any) BufferIO

GetBuiltinBufferIO returns its BufferIO interface if supported. Returns nil if not supported.

type ExtendConn

type ExtendConn interface {
	net.Conn

	BufferIO
	KeepAliveSupport
	PerformanceCounterReporter
}

type KeepAliveSupport

type KeepAliveSupport interface {
	SupportKeepAlive() bool

	// Returns true if the underlying connection is confirmed to be closed.
	// A false does not necessary mean the connection is still active.
	IsClosed() bool
}

type PerformanceCounterReporter

type PerformanceCounterReporter interface {
	GetExtensionProperty() map[string]uint64
}