diff --git a/x/xnet/stack-berkeley.go b/x/xnet/stack-berkeley.go new file mode 100644 index 0000000..d499c8d --- /dev/null +++ b/x/xnet/stack-berkeley.go @@ -0,0 +1,167 @@ +package xnet + +import ( + "context" + "errors" + "net" + "net/netip" + "syscall" + "time" + + "github.com/soypat/lneto/tcp" +) + +type BerkeleyConfig struct { + ListenerPoolConfig TCPPoolConfig +} + +func (s StackBlocking) StackBerkeley(cfg BerkeleyConfig) StackBerkeley { + return StackBerkeley{ + blk: s, + plcfg: cfg.ListenerPoolConfig, + } +} + +type StackBerkeley struct { + blk StackBlocking + plcfg TCPPoolConfig +} + +func (s StackBerkeley) Socket(ctx context.Context, network string, family, sotype int, laddr, raddr net.Addr) (c interface{}, err error) { + switch family { + case syscall.AF_INET: + default: + return nil, errors.New("unsupported address family") + } + var local, remote netip.AddrPort + if laddr != nil { + local, err = netip.ParseAddrPort(local.String()) + if err != nil { + return nil, err + } + } + if raddr != nil { + remote, err = netip.ParseAddrPort(remote.String()) + if err != nil { + return nil, err + } + } + + switch network { + case "udp", "udp4": + return nil, errors.New("udp not yet supported") + case "tcp", "tcp4": + if sotype != syscall.SOCK_STREAM { + return nil, errors.New("unsupported socket type") + } + + if raddr != nil { + var conn tcp.Conn + // DIAL TCP: active connection a.k.a TCP Client branch. + err = s.blk.async.DialTCP(&conn, local.Port(), remote) + if err != nil { + return nil, err + } + for { + time.Sleep(s.blk.loopSleep) + state := conn.State() + if state == tcp.StateEstablished { + tc := tcpconn{ + Conn: &conn, + localAddr: laddr, + } + return tc, nil + } else if state == tcp.StateSynSent || state == tcp.StateSynRcvd || conn.InternalHandler().AwaitingSynSend() { + if err = ctx.Err(); err != nil { + conn.Abort() + return nil, err + } + } else { + // Unexpected state, abort and terminate connection. + conn.Abort() + return errTCPFailedToConnect, nil + } + } + } else { + // LISTEN TCP: passive connection. fulfills net.Listener interface. + pool, err := NewTCPPool(s.plcfg) + if err != nil { + return nil, err + } + var l tcplistener + l.sleep = s.blk.loopSleep + err = l.l.Reset(local.Port(), pool) + if err != nil { + return nil, err + } + err = s.blk.async.RegisterListener(&l.l) + if err != nil { + return nil, err + } + return &l, nil + } + } + return nil, errors.New("unsupported network") +} + +type tcplistener struct { + l tcp.Listener + closed bool + sleep time.Duration + localAddr net.Addr +} + +var _ net.Listener = (*tcplistener)(nil) + +func (l *tcplistener) Addr() net.Addr { + return l.localAddr +} + +func (l *tcplistener) Accept() (net.Conn, error) { + if l.closed { + return nil, net.ErrClosed + } + for { + n := l.l.NumberOfReadyToAccept() + if n == 0 { + time.Sleep(l.sleep) + continue + } + c, _, err := l.l.TryAccept() + if err != nil { + return nil, err + } + cc := tcpconn{ + Conn: c, + localAddr: l.localAddr, + } + return cc, nil + } +} + +func (l *tcplistener) Close() error { + if l.closed { + return net.ErrClosed + } + err := l.l.Close() + l.closed = true + return err +} + +type tcpconn struct { + *tcp.Conn + localAddr net.Addr +} + +var _ net.Conn = tcpconn{} + +func (c tcpconn) LocalAddr() net.Addr { + return c.localAddr +} + +func (c tcpconn) RemoteAddr() net.Addr { + return &net.TCPAddr{ + IP: c.Conn.RemoteAddr(), + Port: int(c.Conn.RemotePort()), + } +}