merge with private mirror

.github/workflows/gosec.yml

@@ -15,10 +15,10 @@ jobs:
       - name: Checkout Source
         uses: actions/checkout@v3
       - name: Run Gosec Security Scanner
-        uses: securego/gosec@v2.15.0
+        uses: securego/gosec@v2.21.0
         with:
           args: '-no-fail -fmt=sarif -out=results.sarif -exclude-dir=examples -exclude-dir=stress_test -exclude-dir=other -exclude-dir=docs -exclude-dir=tests -exclude-dir=benchmark ./...'
       - name: Upload SARIF file
-        uses: github/codeql-action/upload-sarif@v2
+        uses: github/codeql-action/upload-sarif@v3
         with:
           sarif_file: results.sarif

async_adapter.go

@@ -202,7 +202,8 @@ func (a *AsyncAdapter) Close() error {
 		return io.EOF
 	}
 
-	_ = a.ioc.poller.Del(&a.slot)
+	_ = a.ioc.UnsetReadWrite(&a.slot)
+	a.ioc.Deregister(&a.slot)
 
 	return syscall.Close(a.slot.Fd)
 }
@@ -245,3 +246,7 @@ func (a *AsyncAdapter) cancelWrites() {
 func (a *AsyncAdapter) RawFd() int {
 	return a.slot.Fd
 }
+
+func (a *AsyncAdapter) Slot() *internal.Slot {
+	return &a.slot
+}

codec.go

@@ -2,64 +2,33 @@ package sonic
 
 import (
 	"errors"
-	"fmt"
-
-	"github.com/talostrading/sonic/internal"
 	"github.com/talostrading/sonic/sonicerrors"
 )
 
 type Encoder[Item any] interface {
-	// Encode encodes the given item into the `dst` byte stream.
+	// Encode the given `Item` into the given buffer.
 	//
 	// Implementations should:
-	// - Commit the bytes into the read area of `dst`.
-	// - ensure `dst` is big enough to hold the serialized item by
-	//   calling dst.Reserve(...)
+	// - Commit() the bytes into the given buffer if the encoding is successful.
+	// - Ensure the given buffer is big enough to hold the serialized `Item`s by calling `Reserve(...)`.
 	Encode(item Item, dst *ByteBuffer) error
 }
 
 type Decoder[Item any] interface {
-	// Decode decodes the given stream into an `Item`.
-	//
-	// An implementation of Codec takes a byte stream that has already
-	// been buffered in `src` and decodes the data into a stream of
-	// `Item` objects.
-	//
-	// Implementations should return an empty Item and ErrNeedMore if
-	// there are not enough bytes to decode into an Item.
+	// Decode the next `Item`, if any, from the provided buffer. If there are not enough bytes to decode an `Item`,
+	// implementations should return an empty `Item` along with `ErrNeedMore`. `CodecConn` will then know to read more
+	// bytes before calling `Decode(...)` again.
 	Decode(src *ByteBuffer) (Item, error)
 }
 
-// Codec defines a generic interface through which one can encode/decode
-// a raw stream of bytes.
-//
-// Implementations are optionally able to track their state which enables
-// writing both stateful and stateless parsers.
+// Codec groups together and Encoder and a Decoder for a CodecConn.
 type Codec[Enc, Dec any] interface {
 	Encoder[Enc]
 	Decoder[Dec]
 }
 
-type CodecConn[Enc, Dec any] interface {
-	AsyncReadNext(func(error, Dec))
-	ReadNext() (Dec, error)
-
-	AsyncWriteNext(Enc, AsyncCallback)
-	WriteNext(Enc) (int, error)
-
-	NextLayer() Stream
-
-	Close() error
-}
-
-var (
-	_ CodecConn[any, any] = &BlockingCodecConn[any, any]{}
-	_ CodecConn[any, any] = &NonblockingCodecConn[any, any]{}
-)
-
-// BlockingCodecConn handles the decoding/encoding of bytes funneled through a
-// provided blocking file descriptor.
-type BlockingCodecConn[Enc, Dec any] struct {
+// CodecConn reads/writes `Item`s through the provided `Codec`. For an example, see `codec/frame.go`.
+type CodecConn[Enc, Dec any] struct {
 	stream Stream
 	codec  Codec[Enc, Dec]
 	src    *ByteBuffer
@@ -69,14 +38,12 @@ type BlockingCodecConn[Enc, Dec any] struct {
 	emptyDec Dec
 }
 
-func NewBlockingCodecConn[Enc, Dec any](
+func NewCodecConn[Enc, Dec any](
 	stream Stream,
 	codec Codec[Enc, Dec],
 	src, dst *ByteBuffer,
-) (*BlockingCodecConn[Enc, Dec], error) {
-	// Works on both blocking and nonblocking fds.
-
-	c := &BlockingCodecConn[Enc, Dec]{
+) (*CodecConn[Enc, Dec], error) {
+	c := &CodecConn[Enc, Dec]{
 		stream: stream,
 		codec:  codec,
 		src:    src,
@@ -85,26 +52,22 @@ func NewBlockingCodecConn[Enc, Dec any](
 	return c, nil
 }
 
-func (c *BlockingCodecConn[Enc, Dec]) AsyncReadNext(cb func(error, Dec)) {
+func (c *CodecConn[Enc, Dec]) AsyncReadNext(cb func(error, Dec)) {
 	item, err := c.codec.Decode(c.src)
 	if errors.Is(err, sonicerrors.ErrNeedMore) {
-		c.scheduleAsyncRead(cb)
+		c.src.AsyncReadFrom(c.stream, func(err error, _ int) {
+			if err != nil {
+				cb(err, c.emptyDec)
+			} else {
+				c.AsyncReadNext(cb)
+			}
+		})
 	} else {
 		cb(err, item)
 	}
 }
 
-func (c *BlockingCodecConn[Enc, Dec]) scheduleAsyncRead(cb func(error, Dec)) {
-	c.src.AsyncReadFrom(c.stream, func(err error, _ int) {
-		if err != nil {
-			cb(err, c.emptyDec)
-		} else {
-			c.AsyncReadNext(cb)
-		}
-	})
-}
-
-func (c *BlockingCodecConn[Enc, Dec]) ReadNext() (Dec, error) {
+func (c *CodecConn[Enc, Dec]) ReadNext() (Dec, error) {
 	for {
 		item, err := c.codec.Decode(c.src)
 		if err == nil {
@@ -122,7 +85,7 @@ func (c *BlockingCodecConn[Enc, Dec]) ReadNext() (Dec, error) {
 	}
 }
 
-func (c *BlockingCodecConn[Enc, Dec]) WriteNext(item Enc) (n int, err error) {
+func (c *CodecConn[Enc, Dec]) WriteNext(item Enc) (n int, err error) {
 	err = c.codec.Encode(item, c.dst)
 	if err == nil {
 		var nn int64
@@ -132,7 +95,7 @@ func (c *BlockingCodecConn[Enc, Dec]) WriteNext(item Enc) (n int, err error) {
 	return
 }
 
-func (c *BlockingCodecConn[Enc, Dec]) AsyncWriteNext(item Enc, cb AsyncCallback) {
+func (c *CodecConn[Enc, Dec]) AsyncWriteNext(item Enc, cb AsyncCallback) {
 	err := c.codec.Encode(item, c.dst)
 	if err == nil {
 		c.dst.AsyncWriteTo(c.stream, cb)
@@ -141,105 +104,10 @@ func (c *BlockingCodecConn[Enc, Dec]) AsyncWriteNext(item Enc, cb AsyncCallback)
 	}
 }
 
-func (c *BlockingCodecConn[Enc, Dec]) NextLayer() Stream {
-	return c.stream
-}
-
-func (c *BlockingCodecConn[Enc, Dec]) Close() error {
-	return c.stream.Close()
-}
-
-type NonblockingCodecConn[Enc, Dec any] struct {
-	stream Stream
-	codec  Codec[Enc, Dec]
-	src    *ByteBuffer
-	dst    *ByteBuffer
-
-	dispatched int
-
-	emptyEnc Enc
-	emptyDec Dec
-}
-
-func NewNonblockingCodecConn[Enc, Dec any](
-	stream Stream,
-	codec Codec[Enc, Dec],
-	src, dst *ByteBuffer,
-) (*NonblockingCodecConn[Enc, Dec], error) {
-	nonblocking, err := internal.IsNonblocking(stream.RawFd())
-	if err != nil {
-		return nil, err
-	}
-	if !nonblocking {
-		return nil, fmt.Errorf("the provided Stream is blocking")
-	}
-
-	c := &NonblockingCodecConn[Enc, Dec]{
-		stream: stream,
-		codec:  codec,
-		src:    src,
-		dst:    dst,
-	}
-	return c, nil
-}
-
-func (c *NonblockingCodecConn[Enc, Dec]) AsyncReadNext(cb func(error, Dec)) {
-	item, err := c.codec.Decode(c.src)
-	if errors.Is(err, sonicerrors.ErrNeedMore) {
-		c.src.AsyncReadFrom(c.stream, func(err error, _ int) {
-			if err != nil {
-				cb(err, c.emptyDec)
-			} else {
-				c.AsyncReadNext(cb)
-			}
-		})
-	} else {
-		cb(err, item)
-	}
-}
-
-func (c *NonblockingCodecConn[Enc, Dec]) ReadNext() (Dec, error) {
-	for {
-		item, err := c.codec.Decode(c.src)
-		if err == nil {
-			return item, nil
-		}
-
-		if err != sonicerrors.ErrNeedMore {
-			return c.emptyDec, err
-		}
-
-		_, err = c.src.ReadFrom(c.stream)
-		if err != nil {
-			return c.emptyDec, err
-		}
-	}
-}
-
-func (c *NonblockingCodecConn[Enc, Dec]) AsyncWriteNext(item Enc, cb AsyncCallback) {
-	if err := c.codec.Encode(item, c.dst); err != nil {
-		cb(err, 0)
-		return
-	}
-
-	// write everything into `dst`
-	c.dst.AsyncWriteTo(c.stream, cb)
-}
-
-func (c *NonblockingCodecConn[Enc, Dec]) WriteNext(item Enc) (n int, err error) {
-	err = c.codec.Encode(item, c.dst)
-	if err == nil {
-		var nn int64
-		nn, err = c.dst.WriteTo(c.stream)
-		n = int(nn)
-	}
-	return
-}
-
-func (c *NonblockingCodecConn[Enc, Dec]) NextLayer() Stream {
+func (c *CodecConn[Enc, Dec]) NextLayer() Stream {
 	return c.stream
 }
 
-func (c *NonblockingCodecConn[Enc, Dec]) Close() error {
+func (c *CodecConn[Enc, Dec]) Close() error {
 	return c.stream.Close()
 }

codec/frame/stream_test.go

@@ -93,7 +93,7 @@ func runClient(port int, t *testing.T) {
 	src := sonic.NewByteBuffer()
 	dst := sonic.NewByteBuffer()
 	codec := NewCodec(src)
-	codecConn, err := sonic.NewNonblockingCodecConn[[]byte, []byte](
+	codecConn, err := sonic.NewCodecConn[[]byte, []byte](
 		conn, codec, src, dst,
 	)
 	if err != nil {