feat: add EventLoop interface and implementation (#700)

Fixes #689
Fixes #703

Author: panjf2000

README.md

@@ -49,6 +49,7 @@ English | [中文](README_ZH.md)
 - [x] Running on `Linux`, `macOS`, `Windows`, and *BSD: `Darwin`/`DragonFlyBSD`/`FreeBSD`/`NetBSD`/`OpenBSD`
 - [x] **Edge-triggered** I/O support
 - [x] Multiple network addresses binding
+- [x] Support registering new connections to event-loops
 
 ## 🕊 Roadmap
 

README_ZH.md

@@ -49,6 +49,7 @@
 - [x] 支持 `Linux`, `macOS`, `Windows` 和 *BSD 操作系统: `Darwin`/`DragonFlyBSD`/`FreeBSD`/`NetBSD`/`OpenBSD`
 - [x] **Edge-triggered** I/O 支持
 - [x] 多网络地址绑定
+- [x] 支持注册新的连接到事件循环
 
 ## 🕊 未来计划
 

acceptor_windows.go

@@ -20,6 +20,7 @@ import (
 	"runtime"
 
 	errorx "github.com/panjf2000/gnet/v2/pkg/errors"
+	"github.com/panjf2000/gnet/v2/pkg/pool/goroutine"
 )
 
 func (eng *engine) listenStream(ln net.Listener) (err error) {
@@ -43,9 +44,9 @@ func (eng *engine) listenStream(ln net.Listener) (err error) {
 			return
 		}
 		el := eng.eventLoops.next(tc.RemoteAddr())
-		c := newTCPConn(tc, el)
+		c := newTCPConn(el, tc, nil)
 		el.ch <- &openConn{c: c}
-		go func(c *conn, tc net.Conn, el *eventloop) {
+		goroutine.DefaultWorkerPool.Submit(func() {
 			var buffer [0x10000]byte
 			for {
 				n, err := tc.Read(buffer[:])
@@ -55,7 +56,7 @@ func (eng *engine) listenStream(ln net.Listener) (err error) {
 				}
 				el.ch <- packTCPConn(c, buffer[:n])
 			}
-		}(c, tc, el)
+		})
 	}
 }
 
@@ -81,7 +82,7 @@ func (eng *engine) ListenUDP(pc net.PacketConn) (err error) {
 			return
 		}
 		el := eng.eventLoops.next(addr)
-		c := newUDPConn(el, pc, pc.LocalAddr(), addr)
+		c := newUDPConn(el, pc, nil, pc.LocalAddr(), addr, nil)
 		el.ch <- packUDPConn(c, buffer[:n])
 	}
 }

client_test.go

@@ -375,7 +375,6 @@ type testClient struct {
 	connected     int32
 	clientActive  int32
 	disconnected  int32
-	workerPool    *goPool.Pool
 	udpReadHeader int32
 }
 
@@ -404,7 +403,6 @@ func (s *testClient) OnClose(c Conn, err error) (action Action) {
 	if atomic.LoadInt32(&s.connected) == atomic.LoadInt32(&s.disconnected) &&
 		atomic.LoadInt32(&s.disconnected) == int32(s.nclients) {
 		action = Shutdown
-		s.workerPool.Release()
 	}
 
 	return
@@ -431,7 +429,8 @@ func (s *testClient) OnTraffic(c Conn) (action Action) {
 
 	if s.async {
 		buf := bbPool.Get()
-		_, _ = c.WriteTo(buf)
+		_, err := c.WriteTo(buf)
+		assert.NoError(s.tester, err, "WriteTo error")
 
 		if s.network == "tcp" || s.network == "unix" {
 			// just for test
@@ -443,17 +442,19 @@ func (s *testClient) OnTraffic(c Conn) (action Action) {
 			atomic.AddInt32(&s.udpReadHeader, 1)
 			buf.Reset()
 		}
-		_ = s.workerPool.Submit(
+		err = goPool.DefaultWorkerPool.Submit(
 			func() {
 				if buf.Len() > 0 {
 					err := c.AsyncWrite(buf.Bytes(), nil)
 					assert.NoError(s.tester, err)
 				}
 			})
+		assert.NoError(s.tester, err)
 		return
 	}
 
-	buf, _ := c.Next(-1)
+	buf, err := c.Next(-1)
+	assert.NoError(s.tester, err, "Reading data error")
 	if v == nil && bytes.Equal(buf, pingMsg) {
 		atomic.AddInt32(&s.udpReadHeader, 1)
 		buf = nil
@@ -475,7 +476,12 @@ func (s *testClient) OnTick() (delay time.Duration, action Action) {
 			if i%2 == 0 {
 				netConn = true
 			}
-			go startGnetClient(s.tester, s.client, s.network, s.addr, s.multicore, s.async, netConn)
+			err := goPool.DefaultWorkerPool.Submit(
+				func() {
+					startGnetClient(s.tester, s.client, s.network, s.addr, s.multicore, s.async, netConn)
+				})
+			assert.NoError(s.tester, err)
+
 		}
 	}
 	if s.network == "udp" && atomic.LoadInt32(&s.clientActive) == 0 {
@@ -487,13 +493,12 @@ func (s *testClient) OnTick() (delay time.Duration, action Action) {
 
 func runClient(t *testing.T, network, addr string, conf *testConf) {
 	ts := &testClient{
-		tester:     t,
-		network:    network,
-		addr:       addr,
-		multicore:  conf.multicore,
-		async:      conf.async,
-		nclients:   conf.clients,
-		workerPool: goPool.Default(),
+		tester:    t,
+		network:   network,
+		addr:      addr,
+		multicore: conf.multicore,
+		async:     conf.async,
+		nclients:  conf.clients,
 	}
 	var err error
 	clientEV := &clientEvents{tester: t, packetLen: streamLen, svr: ts}
@@ -646,7 +651,10 @@ func (ev *serverEventsForWake) OnClose(_ Conn, _ error) Action {
 
 func (ev *serverEventsForWake) OnTick() (time.Duration, Action) {
 	if atomic.CompareAndSwapInt32(&ev.started, 0, 1) {
-		go testConnWakeImmediately(ev.tester, ev.client, ev.clientEV, ev.network, ev.addr)
+		err := goPool.DefaultWorkerPool.Submit(func() {
+			testConnWakeImmediately(ev.tester, ev.client, ev.clientEV, ev.network, ev.addr)
+		})
+		assert.NoError(ev.tester, err)
 	}
 	return 100 * time.Millisecond, None
 }
@@ -699,15 +707,19 @@ func TestClientReadOnEOF(t *testing.T) {
 	assert.NoError(t, err)
 	defer ln.Close() //nolint:errcheck
 
-	go func() {
+	err = goPool.DefaultWorkerPool.Submit(func() {
 		for {
 			conn, err := ln.Accept()
 			if err != nil {
 				break
 			}
-			go process(conn)
+			err = goPool.DefaultWorkerPool.Submit(func() {
+				process(conn)
+			})
+			assert.NoError(t, err)
 		}
-	}()
+	})
+	assert.NoError(t, err)
 
 	ev := &clientReadOnEOF{
 		result: make(chan struct {

client_unix.go

@@ -242,7 +242,7 @@ func (cli *Client) EnrollContext(c net.Conn, ctx any) (Conn, error) {
 	}}
 	err = cli.el.poller.Trigger(queue.HighPriority, cli.el.register, ccb)
 	if err != nil {
-		_ = gc.Close()
+		gc.Close() //nolint:errcheck
 		return nil, err
 	}
 

client_windows.go

@@ -25,6 +25,7 @@ import (
 
 	errorx "github.com/panjf2000/gnet/v2/pkg/errors"
 	"github.com/panjf2000/gnet/v2/pkg/logging"
+	"github.com/panjf2000/gnet/v2/pkg/pool/goroutine"
 )
 
 type Client struct {
@@ -161,61 +162,56 @@ func (cli *Client) EnrollContext(nc net.Conn, ctx any) (gc Conn, err error) {
 			}
 		}
 
-		c := newTCPConn(nc, cli.el)
-		c.SetContext(ctx)
+		c := newTCPConn(cli.el, nc, ctx)
 		cli.el.ch <- &openConn{c: c, cb: func() { close(connOpened) }}
-		go func(c *conn, tc net.Conn, el *eventloop) {
+		goroutine.DefaultWorkerPool.Submit(func() {
 			var buffer [0x10000]byte
 			for {
-				n, err := tc.Read(buffer[:])
+				n, err := nc.Read(buffer[:])
 				if err != nil {
-					el.ch <- &netErr{c, err}
+					cli.el.ch <- &netErr{c, err}
 					return
 				}
-				el.ch <- packTCPConn(c, buffer[:n])
+				cli.el.ch <- packTCPConn(c, buffer[:n])
 			}
-		}(c, nc, cli.el)
+		})
 		gc = c
 	case *net.UnixConn:
-		c := newTCPConn(nc, cli.el)
-		c.SetContext(ctx)
+		c := newTCPConn(cli.el, nc, ctx)
 		cli.el.ch <- &openConn{c: c, cb: func() { close(connOpened) }}
-		go func(c *conn, uc net.Conn, el *eventloop) {
+		goroutine.DefaultWorkerPool.Submit(func() {
 			var buffer [0x10000]byte
 			for {
-				n, err := uc.Read(buffer[:])
+				n, err := nc.Read(buffer[:])
 				if err != nil {
-					el.ch <- &netErr{c, err}
+					cli.el.ch <- &netErr{c, err}
 					mu.RLock()
-					tmpDir := unixAddrDirs[uc.LocalAddr().String()]
+					tmpDir := unixAddrDirs[nc.LocalAddr().String()]
 					mu.RUnlock()
 					if err := os.RemoveAll(tmpDir); err != nil {
 						logging.Errorf("failed to remove temporary directory for unix local address: %v", err)
 					}
 					return
 				}
-				el.ch <- packTCPConn(c, buffer[:n])
+				cli.el.ch <- packTCPConn(c, buffer[:n])
 			}
-		}(c, nc, cli.el)
+		})
 		gc = c
 	case *net.UDPConn:
-		c := newUDPConn(cli.el, nil, nc.LocalAddr(), nc.RemoteAddr())
-		c.SetContext(ctx)
-		c.rawConn = nc
+		c := newUDPConn(cli.el, nil, nc, nc.LocalAddr(), nc.RemoteAddr(), ctx)
 		cli.el.ch <- &openConn{c: c, cb: func() { close(connOpened) }}
-		go func(uc net.Conn, el *eventloop) {
+		goroutine.DefaultWorkerPool.Submit(func() {
 			var buffer [0x10000]byte
 			for {
-				n, err := uc.Read(buffer[:])
+				n, err := nc.Read(buffer[:])
 				if err != nil {
+					cli.el.ch <- &netErr{c, err}
 					return
 				}
-				c := newUDPConn(cli.el, nil, uc.LocalAddr(), uc.RemoteAddr())
-				c.SetContext(ctx)
-				c.rawConn = uc
-				el.ch <- packUDPConn(c, buffer[:n])
+				c := newUDPConn(cli.el, nil, nc, nc.LocalAddr(), nc.RemoteAddr(), ctx)
+				cli.el.ch <- packUDPConn(c, buffer[:n])
 			}
-		}(nc, cli.el)
+		})
 		gc = c
 	default:
 		return nil, errorx.ErrUnsupportedProtocol

connection_unix.go

@@ -519,6 +519,10 @@ func (c *conn) Close() error {
 	}, nil)
 }
 
+func (c *conn) EventLoop() EventLoop {
+	return c.loop
+}
+
 func (*conn) SetDeadline(_ time.Time) error {
 	return errorx.ErrUnsupportedOp
 }

connection_windows.go

@@ -21,14 +21,13 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/panjf2000/ants/v2"
 	"golang.org/x/sys/windows"
 
 	"github.com/panjf2000/gnet/v2/pkg/buffer/elastic"
 	errorx "github.com/panjf2000/gnet/v2/pkg/errors"
 	bbPool "github.com/panjf2000/gnet/v2/pkg/pool/bytebuffer"
 	bsPool "github.com/panjf2000/gnet/v2/pkg/pool/byteslice"
-	goPool "github.com/panjf2000/gnet/v2/pkg/pool/goroutine"
+	"github.com/panjf2000/gnet/v2/pkg/pool/goroutine"
 )
 
 type netErr struct {
@@ -83,8 +82,9 @@ func packUDPConn(c *conn, buf []byte) *udpConn {
 	return &udpConn{c}
 }
 
-func newTCPConn(nc net.Conn, el *eventloop) (c *conn) {
+func newTCPConn(el *eventloop, nc net.Conn, ctx any) (c *conn) {
 	return &conn{
+		ctx:        ctx,
 		loop:       el,
 		buffer:     bbPool.Get(),
 		rawConn:    nc,
@@ -105,9 +105,11 @@ func (c *conn) release() {
 	c.buffer = nil
 }
 
-func newUDPConn(el *eventloop, pc net.PacketConn, localAddr, remoteAddr net.Addr) *conn {
+func newUDPConn(el *eventloop, pc net.PacketConn, rc net.Conn, localAddr, remoteAddr net.Addr, ctx any) *conn {
 	return &conn{
+		ctx:        ctx,
 		pc:         pc,
+		rawConn:    rc,
 		loop:       el,
 		buffer:     bbPool.Get(),
 		localAddr:  localAddr,
@@ -419,19 +421,6 @@ func (c *conn) SetKeepAlivePeriod(d time.Duration) error {
 	return nil
 }
 
-type nonBlockingPool struct {
-	*goPool.Pool
-}
-
-func (np *nonBlockingPool) Go(task func()) (err error) {
-	if err = np.Submit(task); err == ants.ErrPoolOverload {
-		go task()
-	}
-	return
-}
-
-var workerPool = nonBlockingPool{Pool: goPool.Default()}
-
 // Gfd return an uninitialized GFD which is not valid,
 // this method is only implemented for compatibility, don't use it on Windows.
 // func (c *conn) Gfd() gfd.GFD { return gfd.GFD{} }
@@ -450,7 +439,7 @@ func (c *conn) AsyncWrite(buf []byte, cb AsyncCallback) error {
 	case c.loop.ch <- fn:
 	default:
 		// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.
-		err = workerPool.Go(func() {
+		err = goroutine.DefaultWorkerPool.Submit(func() {
 			c.loop.ch <- fn
 		})
 	}
@@ -489,7 +478,7 @@ func (c *conn) Wake(cb AsyncCallback) (err error) {
 	case c.loop.ch <- wakeFn:
 	default:
 		// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.
-		err = workerPool.Go(func() {
+		err = goroutine.DefaultWorkerPool.Submit(func() {
 			c.loop.ch <- wakeFn
 		})
 	}
@@ -506,7 +495,7 @@ func (c *conn) Close() (err error) {
 	case c.loop.ch <- closeFn:
 	default:
 		// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.
-		err = workerPool.Go(func() {
+		err = goroutine.DefaultWorkerPool.Submit(func() {
 			c.loop.ch <- closeFn
 		})
 	}
@@ -527,14 +516,18 @@ func (c *conn) CloseWithCallback(cb AsyncCallback) (err error) {
 	case c.loop.ch <- closeFn:
 	default:
 		// If the event-loop channel is full, asynchronize this operation to avoid blocking the eventloop.
-		err = workerPool.Go(func() {
+		err = goroutine.DefaultWorkerPool.Submit(func() {
 			c.loop.ch <- closeFn
 		})
 	}
 
 	return
 }
 
+func (c *conn) EventLoop() EventLoop {
+	return c.loop
+}
+
 func (*conn) SetDeadline(_ time.Time) error {
 	return errorx.ErrUnsupportedOp
 }