Merge pull request #150 from loopholelabs/staging

v0.6.0

CHANGELOG.md

@@ -7,6 +7,20 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+## [v0.6.0] - 2022-08-07 (Beta)
+
+## Changes
+
+- **[BREAKING]** The `server` now concurrently process incoming packets from connections by calling handler functions in a goroutine.
+  This is done to avoid blocking the main packet processing loop when the handler for an incoming packet is slow.
+- The `UPDATE` Action has been completely removed from the `server` and the `client` - the context can no longer be
+  updated from a handler function.
+- The `SetConcurrency` function has been added to the `server` to set the concurrency of the packet processing
+  goroutines.
+- `io/ioutil.Discard` has been replaced with `io.Discard` because it was being deprecated
+- The `README.md` file has been updated to reflect the `frisbee-go` package name, and better direct users to the `frpc.io` website.
+- @jimmyaxod has been added as a maintainer for the `frisbee-go` package.
+
 ## [v0.5.4] - 2022-07-28 (Beta)
 
 ## Features
@@ -300,7 +314,8 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 Initial Release of Frisbee
 
-[unreleased]: https://github.com/loopholelabs/frisbee/compare/v0.5.4...HEAD
+[unreleased]: https://github.com/loopholelabs/frisbee/compare/v0.6.0...HEAD
+[v0.6.0]: https://github.com/loopholelabs/frisbee/compare/v0.5.4...v0.6.0
 [v0.5.4]: https://github.com/loopholelabs/frisbee/compare/v0.5.3...v0.5.4
 [v0.5.3]: https://github.com/loopholelabs/frisbee/compare/v0.5.2...v0.5.3
 [v0.5.2]: https://github.com/loopholelabs/frisbee/compare/v0.5.1...v0.5.2

MAINTAINERS.md

@@ -1,3 +1,4 @@
 - Shivansh Vij <shivanshvij@loopholelabs.io> @shivanshvij
 - Alex Sørlie Glomsaas <alexsorlieglomsaas@loopholelabs.io> @supermanifolds
 - Felicitas Pojtinger <felicitaspojtinger@loopholelabs.io> @pojntfx
+- Jimmy Moore <jimmy@loopholelabs.io> @jimmyaxod

README.md

@@ -21,8 +21,8 @@ same is true for selected other new features explicitly marked as
 
 ## Usage and Documentation
 
-Usage instructions and documentation for Frisbee is available
-at [https://frpc.io/frisbee](https://frpc.io/frisbee). The Frisbee framework also has great
+Usage instructions and documentation for `frisbee-go` are available
+at [https://frpc.io/frisbee](https://frpc.io/frisbee). This library also has great
 documentation coverage using [GoDoc](https://godoc.org/github.com/loopholelabs/frisbee-go).
 
 ## Contributing

async_test.go

@@ -25,7 +25,6 @@ import (
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"io"
-	"io/ioutil"
 	"net"
 	"runtime"
 	"sync"
@@ -38,7 +37,7 @@ func TestNewAsync(t *testing.T) {
 
 	const packetSize = 512
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	reader, writer := net.Pipe()
 
@@ -95,7 +94,7 @@ func TestAsyncLargeWrite(t *testing.T) {
 	const testSize = 100000
 	const packetSize = 512
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	reader, writer := net.Pipe()
 
@@ -142,7 +141,7 @@ func TestAsyncRawConn(t *testing.T) {
 	const testSize = 100000
 	const packetSize = 32
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	reader, writer, err := pair.New()
 	require.NoError(t, err)
@@ -204,7 +203,7 @@ func TestAsyncReadClose(t *testing.T) {
 
 	reader, writer := net.Pipe()
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	readerConn := NewAsync(reader, &emptyLogger)
 	writerConn := NewAsync(writer, &emptyLogger)
@@ -252,7 +251,7 @@ func TestAsyncReadAvailableClose(t *testing.T) {
 
 	reader, writer := net.Pipe()
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	readerConn := NewAsync(reader, &emptyLogger)
 	writerConn := NewAsync(writer, &emptyLogger)
@@ -302,7 +301,7 @@ func TestAsyncWriteClose(t *testing.T) {
 
 	reader, writer := net.Pipe()
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	readerConn := NewAsync(reader, &emptyLogger)
 	writerConn := NewAsync(writer, &emptyLogger)
@@ -352,7 +351,7 @@ func TestAsyncWriteClose(t *testing.T) {
 func TestAsyncTimeout(t *testing.T) {
 	t.Parallel()
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	reader, writer, err := pair.New()
 	require.NoError(t, err)
@@ -380,7 +379,7 @@ func TestAsyncTimeout(t *testing.T) {
 	assert.Equal(t, uint32(0), p.Metadata.ContentLength)
 	assert.Equal(t, 0, len(*p.Content))
 
-	time.Sleep(DefaultDeadline * 5)
+	time.Sleep(DefaultDeadline * 2)
 
 	err = writerConn.Error()
 	require.NoError(t, err)
@@ -400,7 +399,7 @@ func TestAsyncTimeout(t *testing.T) {
 	require.NoError(t, err)
 
 	runtime.Gosched()
-	time.Sleep(DefaultDeadline * 5)
+	time.Sleep(DefaultDeadline * 2)
 	runtime.Gosched()
 
 	p, err = readerConn.ReadPacket()
@@ -417,7 +416,7 @@ func TestAsyncTimeout(t *testing.T) {
 	err = readerConn.Error()
 	if err == nil {
 		runtime.Gosched()
-		time.Sleep(DefaultDeadline * 10)
+		time.Sleep(DefaultDeadline * 3)
 		runtime.Gosched()
 	}
 	require.Error(t, readerConn.Error())
@@ -431,7 +430,7 @@ func TestAsyncTimeout(t *testing.T) {
 func BenchmarkAsyncThroughputPipe(b *testing.B) {
 	const testSize = 100
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	reader, writer := net.Pipe()
 
@@ -451,7 +450,7 @@ func BenchmarkAsyncThroughputPipe(b *testing.B) {
 func BenchmarkAsyncThroughputNetwork(b *testing.B) {
 	const testSize = 100
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 
 	reader, writer, err := pair.New()
 	if err != nil {
@@ -530,7 +529,7 @@ func BenchmarkAsyncThroughputNetworkMultiple(b *testing.B) {
 			b.ReportAllocs()
 			for i := 0; i < numClients; i++ {
 				go func() {
-					emptyLogger := zerolog.New(ioutil.Discard)
+					emptyLogger := zerolog.New(io.Discard)
 
 					reader, writer, err := pair.New()
 					if err != nil {

client.go

@@ -179,56 +179,52 @@ func (c *Client) handleConn() {
 	var action Action
 	var err error
 	var handlerFunc Handler
-LOOP:
-	if c.closed.Load() {
-		c.wg.Done()
-		return
-	}
-	p, err = c.conn.ReadPacket()
-	if err != nil {
-		c.Logger().Debug().Err(err).Msg("error while getting packet frisbee connection")
-		c.wg.Done()
-		_ = c.Close()
-		return
-	}
-	handlerFunc = c.handlerTable[p.Metadata.Operation]
-	if handlerFunc != nil {
-		packetCtx := c.ctx
-		if c.PacketContext != nil {
-			packetCtx = c.PacketContext(packetCtx, p)
+	for {
+		if c.closed.Load() {
+			c.wg.Done()
+			return
 		}
-		outgoing, action = handlerFunc(packetCtx, p)
-		if outgoing != nil && outgoing.Metadata.ContentLength == uint32(len(*outgoing.Content)) {
-			err = c.conn.WritePacket(outgoing)
-			if outgoing != p {
-				packet.Put(outgoing)
+		p, err = c.conn.ReadPacket()
+		if err != nil {
+			c.Logger().Debug().Err(err).Msg("error while getting packet frisbee connection")
+			c.wg.Done()
+			_ = c.Close()
+			return
+		}
+		handlerFunc = c.handlerTable[p.Metadata.Operation]
+		if handlerFunc != nil {
+			packetCtx := c.ctx
+			if c.PacketContext != nil {
+				packetCtx = c.PacketContext(packetCtx, p)
 			}
-			packet.Put(p)
-			if err != nil {
-				c.Logger().Error().Err(err).Msg("error while writing to frisbee conn")
+			outgoing, action = handlerFunc(packetCtx, p)
+			if outgoing != nil && outgoing.Metadata.ContentLength == uint32(len(*outgoing.Content)) {
+				err = c.conn.WritePacket(outgoing)
+				if outgoing != p {
+					packet.Put(outgoing)
+				}
+				packet.Put(p)
+				if err != nil {
+					c.Logger().Error().Err(err).Msg("error while writing to frisbee conn")
+					c.wg.Done()
+					_ = c.Close()
+					return
+				}
+			} else {
+				packet.Put(p)
+			}
+			switch action {
+			case NONE:
+			case CLOSE:
+				c.Logger().Debug().Msgf("Closing connection %s because of CLOSE action", c.conn.RemoteAddr())
 				c.wg.Done()
 				_ = c.Close()
 				return
 			}
 		} else {
 			packet.Put(p)
 		}
-		switch action {
-		case NONE:
-		case UPDATE:
-			if c.UpdateContext != nil {
-				c.ctx = c.UpdateContext(c.ctx, c.conn)
-			}
-		case CLOSE:
-			c.Logger().Debug().Msgf("Closing connection %s because of CLOSE action", c.conn.RemoteAddr())
-			c.wg.Done()
-			_ = c.Close()
-			return
-		}
-	} else {
-		packet.Put(p)
 	}
-	goto LOOP
 }
 
 func (c *Client) heartbeat() {

client_test.go

@@ -25,7 +25,7 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-	"io/ioutil"
+	"io"
 	"net"
 	"testing"
 )
@@ -63,10 +63,12 @@ func TestClientRaw(t *testing.T) {
 		return
 	}
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 	s, err := NewServer(serverHandlerTable, WithLogger(&emptyLogger))
 	require.NoError(t, err)
 
+	s.SetConcurrency(1)
+
 	s.ConnContext = func(ctx context.Context, c *Async) context.Context {
 		return context.WithValue(ctx, clientConnContextKey, c)
 	}
@@ -156,10 +158,12 @@ func TestClientStaleClose(t *testing.T) {
 		return
 	}
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 	s, err := NewServer(serverHandlerTable, WithLogger(&emptyLogger))
 	require.NoError(t, err)
 
+	s.SetConcurrency(1)
+
 	serverConn, clientConn, err := pair.New()
 	require.NoError(t, err)
 
@@ -214,12 +218,14 @@ func BenchmarkThroughputClient(b *testing.B) {
 		return
 	}
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 	s, err := NewServer(serverHandlerTable, WithLogger(&emptyLogger))
 	if err != nil {
 		b.Fatal(err)
 	}
 
+	s.SetConcurrency(1)
+
 	serverConn, clientConn, err := pair.New()
 	if err != nil {
 		b.Fatal(err)
@@ -297,12 +303,14 @@ func BenchmarkThroughputResponseClient(b *testing.B) {
 		return
 	}
 
-	emptyLogger := zerolog.New(ioutil.Discard)
+	emptyLogger := zerolog.New(io.Discard)
 	s, err := NewServer(serverHandlerTable, WithLogger(&emptyLogger))
 	if err != nil {
 		b.Fatal(err)
 	}
 
+	s.SetConcurrency(1)
+
 	serverConn, clientConn, err := pair.New()
 	if err != nil {
 		b.Fatal(err)

conn.go

@@ -22,7 +22,7 @@ import (
 	"github.com/loopholelabs/frisbee-go/pkg/packet"
 	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
-	"io/ioutil"
+	"io"
 	"net"
 	"time"
 )
@@ -31,7 +31,7 @@ import (
 const DefaultBufferSize = 1 << 16
 
 var (
-	defaultLogger = zerolog.New(ioutil.Discard)
+	defaultLogger = zerolog.New(io.Discard)
 
 	DefaultDeadline = time.Second * 5