Skip to content

wait for client conns to close #134

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
ingon wants to merge 1 commit into
base: main
Choose a base branch
from
Draft

wait for client conns to close #134

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion cmd/connet/control.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -194,7 +194,9 @@ func controlRun(ctx context.Context, cfg ControlConfig, logger *slog.Logger) err
if err != nil {
return fmt.Errorf("create control server: %w", err)
}
return runWithStatus(ctx, srv, statusAddr, logger)
err = runWithStatus(ctx, srv, statusAddr, logger)
srv.WaitDrainConns(ctx)
return err
}

func (cfg ControlIngress) parse() (control.Ingress, error) {
Expand Down
4 changes: 3 additions & 1 deletion cmd/connet/relay.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -185,7 +185,9 @@ func relayRun(ctx context.Context, cfg RelayConfig, logger *slog.Logger) error {
if err != nil {
return fmt.Errorf("create relay server: %w", err)
}
return runWithStatus(ctx, srv, statusAddr, logger)
err = runWithStatus(ctx, srv, statusAddr, logger)
srv.WaitDrainConns(ctx)
return err
}

func (cfg RelayIngress) parse() (relay.Ingress, error) {
Expand Down
7 changes: 6 additions & 1 deletion cmd/connet/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,12 @@ func serverRun(ctx context.Context, cfg ServerConfig, logger *slog.Logger) error
if err != nil {
return fmt.Errorf("create server: %w", err)
}
return runWithStatus(ctx, srv, statusAddr, logger)

err = runWithStatus(ctx, srv, statusAddr, logger)
logger.Info("server stopped, draining conns")
srv.WaitDrainConns(ctx)
logger.Info("server stopped, completed conns")
return err
}

func (c *ServerConfig) merge(o ServerConfig) {
Expand Down
15 changes: 13 additions & 2 deletions server/control/clients.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -163,6 +163,8 @@ type clientServer struct {

reactivate map[ClientID]reactivateValue
reactivateMu sync.RWMutex

connsWg sync.WaitGroup
}

type peerKey struct {
Expand Down Expand Up @@ -268,6 +270,7 @@ func (s *clientServer) listen(ctx context.Context, endpoint model.Endpoint, role
func (s *clientServer) run(ctx context.Context) error {
g := reliable.NewGroup(ctx)

s.connsWg.Add(len(s.ingresses))
for _, ingress := range s.ingresses {
g.Go(reliable.Bind(ingress, s.runListener))
}
Expand All @@ -281,6 +284,8 @@ func (s *clientServer) run(ctx context.Context) error {
}

func (s *clientServer) runListener(ctx context.Context, ingress Ingress) error {
defer s.connsWg.Done()

s.logger.Debug("start udp listener", "addr", ingress.Addr)
udpConn, err := net.ListenUDP("udp", ingress.Addr)
if err != nil {
Expand Down Expand Up @@ -339,7 +344,10 @@ func (s *clientServer) runListener(ctx context.Context, ingress Ingress) error {
conn: conn,
logger: s.logger,
}
go cc.run(ctx)
s.connsWg.Go(func() {
cc.run(ctx)
})
// go cc.run(ctx)
}
}

Expand Down Expand Up @@ -518,7 +526,10 @@ func (c *clientConn) runErr(ctx context.Context) error {
conn: c,
stream: stream,
}
go cs.run(ctx)
c.server.connsWg.Go(func() {
cs.run(ctx)
})
// go cs.run(ctx)
}
}

Expand Down
10 changes: 9 additions & 1 deletion server/control/relays.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -132,6 +132,8 @@ type relayServer struct {
connsCache map[RelayID]cachedRelay
connsOffset int64
connsMu sync.RWMutex

connsWg sync.WaitGroup
}

type cachedRelay struct {
Expand Down Expand Up @@ -223,6 +225,7 @@ func (s *relayServer) Relays(ctx context.Context, endpoint model.Endpoint, role
func (s *relayServer) run(ctx context.Context) error {
g := reliable.NewGroup(ctx)

s.connsWg.Add(len(s.ingresses))
for _, ingress := range s.ingresses {
g.Go(reliable.Bind(ingress, s.runListener))
}
Expand All @@ -234,6 +237,8 @@ func (s *relayServer) run(ctx context.Context) error {
}

func (s *relayServer) runListener(ctx context.Context, ingress Ingress) error {
defer s.connsWg.Done()

s.logger.Debug("start udp listener", "addr", ingress.Addr)
udpConn, err := net.ListenUDP("udp", ingress.Addr)
if err != nil {
Expand Down Expand Up @@ -292,7 +297,10 @@ func (s *relayServer) runListener(ctx context.Context, ingress Ingress) error {
conn: conn,
logger: s.logger,
}
go rc.run(ctx)
s.connsWg.Go(func() {
rc.run(ctx)
})
// go rc.run(ctx)
}
}

Expand Down
13 changes: 9 additions & 4 deletions server/control/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,15 +25,15 @@ type Config struct {
}

func NewServer(cfg Config) (*Server, error) {
if err := cfg.Stores.RemoveDeprecated(); err != nil {
cfg.Logger.Warn("could not remove deprecated stores", "err", err)
}

configStore, err := cfg.Stores.Config()
if err != nil {
return nil, fmt.Errorf("config store open: %w", err)
}

if err := cfg.Stores.RemoveDeprecated(); err != nil {
cfg.Logger.Warn("could not remove deprecated stores", "err", err)
}

relays, err := newRelayServer(cfg.RelaysIngress, cfg.RelaysAuth, configStore, cfg.Stores, cfg.Logger)
if err != nil {
return nil, fmt.Errorf("create relay server: %w", err)
Expand Down Expand Up @@ -67,6 +67,11 @@ func (s *Server) Run(ctx context.Context) error {
)
}

func (s *Server) WaitDrainConns(ctx context.Context) {
s.relays.connsWg.Wait()
s.clients.connsWg.Wait()
}

func (s *Server) Status(ctx context.Context) (Status, error) {
clients, err := s.getClients()
if err != nil {
Expand Down
10 changes: 8 additions & 2 deletions server/relay/clients.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,8 @@ type clientsServer struct {
endpoints map[model.Endpoint]*endpointClients
endpointsMu sync.RWMutex

logger *slog.Logger
connsWg sync.WaitGroup
logger *slog.Logger
}

func newClientsServer(cfg Config, cert *certc.Cert, auth ClientAuthenticator) (*clientsServer, error) {
Expand Down Expand Up @@ -190,6 +191,8 @@ type clientsServerCfg struct {
}

func (s *clientsServer) run(ctx context.Context, cfg clientsServerCfg) error {
defer s.connsWg.Done()

s.logger.Debug("start udp listener", "addr", cfg.ingress.Addr)
udpConn, err := net.ListenUDP("udp", cfg.ingress.Addr)
if err != nil {
Expand Down Expand Up @@ -246,7 +249,10 @@ func (s *clientsServer) run(ctx context.Context, cfg clientsServerCfg) error {
conn: conn,
logger: s.logger,
}
go rc.run(ctx)
s.connsWg.Go(func() {
rc.run(ctx)
})
// go rc.run(ctx)
}
}

Expand Down
13 changes: 9 additions & 4 deletions server/relay/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,15 +42,15 @@ func NewServer(cfg Config) (*Server, error) {
return nil, fmt.Errorf("relay server is missing ingresses")
}

if err := cfg.Stores.RemoveDeprecated(); err != nil {
cfg.Logger.Warn("could not remove deprecated stores", "err", err)
}

configStore, err := cfg.Stores.Config()
if err != nil {
return nil, fmt.Errorf("relay stores: %w", err)
}

if err := cfg.Stores.RemoveDeprecated(); err != nil {
cfg.Logger.Warn("could not remove deprecated stores", "err", err)
}

statelessResetVal, err := configStore.GetOrInit(configStatelessReset, func(ck ConfigKey) (ConfigValue, error) {
var key quic.StatelessResetKey
if _, err := io.ReadFull(rand.Reader, key[:]); err != nil {
Expand Down Expand Up @@ -112,6 +112,7 @@ func (s *Server) Run(ctx context.Context) error {

g := reliable.NewGroup(ctx)

s.clients.connsWg.Add(len(s.ingress))
for _, ingress := range s.ingress {
cfg := clientsServerCfg{
ingress: ingress,
Expand All @@ -131,6 +132,10 @@ func (s *Server) Run(ctx context.Context) error {
return g.Wait()
}

func (s *Server) WaitDrainConns(ctx context.Context) {
s.clients.connsWg.Wait()
}

type Status struct {
Status statusc.Status `json:"status"`
BuildVersion string `json:"build-version"`
Expand Down
5 changes: 5 additions & 0 deletions server/server.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,11 @@ func (s *Server) Run(ctx context.Context) error {
return g.Wait()
}

func (s *Server) WaitDrainConns(ctx context.Context) {
s.control.WaitDrainConns(ctx)
s.relay.WaitDrainConns(ctx)
}

func (s *Server) Status(ctx context.Context) (ServerStatus, error) {
control, err := s.control.Status(ctx)
if err != nil {
Expand Down