diff --git a/cmd/mochi-e2e/e2e.go b/cmd/mochi-e2e/e2e.go index 0d68274..9841c72 100644 --- a/cmd/mochi-e2e/e2e.go +++ b/cmd/mochi-e2e/e2e.go @@ -12,6 +12,8 @@ import ( "github.com/anacrolix/torrent/tracker" "github.com/sot-tech/mochi/bittorrent" + + _ "github.com/sot-tech/mochi/pkg/randseed" ) func main() { diff --git a/cmd/mochi/config.go b/cmd/mochi/config.go index aac39bd..d51ebaf 100644 --- a/cmd/mochi/config.go +++ b/cmd/mochi/config.go @@ -11,6 +11,9 @@ import ( fu "github.com/sot-tech/mochi/frontend/udp" "github.com/sot-tech/mochi/pkg/conf" + // Seed math random + _ "github.com/sot-tech/mochi/pkg/randseed" + // Imports to register middleware hooks. _ "github.com/sot-tech/mochi/middleware/clientapproval" _ "github.com/sot-tech/mochi/middleware/jwt" diff --git a/cmd/mochi/main.go b/cmd/mochi/main.go index 393e28b..e5bea9f 100644 --- a/cmd/mochi/main.go +++ b/cmd/mochi/main.go @@ -27,6 +27,7 @@ func main() { logOut := flag.String(logOutArg, "stderr", "output for logging, might be 'stderr', 'stdout' or file path") logLevel := flag.String(logLevelArg, "warn", "logging level: trace, debug, info, warn, error, fatal, panic") logPretty := flag.Bool(logPrettyArg, false, "enable log pretty print. used only if 'logOut' set to 'stdout' or 'stderr'. if not set, log outputs json") + //goland:noinspection GoBoolExpressions logColored := flag.Bool(logColorsArg, runtime.GOOS == "windows", "enable log coloring. used only if set 'logPretty'") configPath := flag.String(configArg, "/etc/mochi.yaml", "location of configuration file") quickStart := flag.Bool(quickArg, false, "start tracker with default configuration (all frontends, in-memory store, no hooks)") diff --git a/dist/example_config.yaml b/dist/example_config.yaml index 3e92b1c..558373c 100644 --- a/dist/example_config.yaml +++ b/dist/example_config.yaml @@ -34,10 +34,14 @@ frontends: tls_key_path: "" # Enable SO_REUSEPORT to allow starting multiple mochi instances with the same HTTP(S) port. - # You can also use this parameter to define two or mote listeners for the same address and port, - # and (possibly) increase throughput. + # You can also use this parameter to define two or more listeners or separate processes + # for the same address and port, and (possibly) increase throughput (faster queue processing + # because of multiple 'workers'). reuse_port: true + # Number of connection listeners to start. See reuse_port, default is 1. + workers: 1 + # The timeout durations for HTTP requests. read_timeout: 5s write_timeout: 5s @@ -110,10 +114,14 @@ frontends: addr: "0.0.0.0:6969" # Enable SO_REUSEPORT to allow starting multiple mochi instances with the same UDP port. - # You can also use this parameter to define two or mote listeners for the same address and port, - # and (a little) increase throughput (faster queue processing because of multiple 'workers'). + # You can also use this parameter to define two or more listeners or separate processes + # for the same address and port, and (a little) increase throughput (faster queue processing + # because of multiple 'workers'). reuse_port: true + # Number of connection listeners to start. See reuse_port, default is 1. + workers: 1 + # The leeway for a timestamp on a connection ID. max_clock_skew: 10s diff --git a/dist/example_config_pg.yaml b/dist/example_config_pg.yaml index 26647f7..e830291 100644 --- a/dist/example_config_pg.yaml +++ b/dist/example_config_pg.yaml @@ -14,6 +14,7 @@ frontends: tls_cert_path: "" tls_key_path: "" reuse_port: true + workers: 1 read_timeout: 5s write_timeout: 5s enable_keepalive: false @@ -36,6 +37,7 @@ frontends: config: addr: "0.0.0.0:6969" reuse_port: true + workers: 1 max_clock_skew: 10s private_key: "paste a random string here that will be used to hmac connection IDs" enable_request_timing: false diff --git a/dist/example_config_redis.yaml b/dist/example_config_redis.yaml index 792e047..adbbde9 100644 --- a/dist/example_config_redis.yaml +++ b/dist/example_config_redis.yaml @@ -14,6 +14,7 @@ frontends: tls_cert_path: "" tls_key_path: "" reuse_port: true + workers: 1 read_timeout: 5s write_timeout: 5s enable_keepalive: false @@ -36,6 +37,7 @@ frontends: config: addr: "0.0.0.0:6969" reuse_port: true + workers: 1 max_clock_skew: 10s private_key: "paste a random string here that will be used to hmac connection IDs" enable_request_timing: false diff --git a/frontend/frontend.go b/frontend/frontend.go index b00c1e1..9276ed3 100644 --- a/frontend/frontend.go +++ b/frontend/frontend.go @@ -3,8 +3,10 @@ package frontend import ( + "errors" "fmt" "io" + "strings" "sync" "github.com/sot-tech/mochi/middleware" @@ -71,3 +73,34 @@ func NewFrontends(configs []conf.NamedMapConfig, logic *middleware.Logic) (fs [] } return } + +// CloseGroup simultaneously calls Close for each non-nil +// array element and combines non-nil errors into one +func CloseGroup(cls []io.Closer) (err error) { + l := len(cls) + errs := make([]error, l) + wg := sync.WaitGroup{} + wg.Add(l) + for i, c := range cls { + if c != nil { + go func(i int, c io.Closer) { + defer wg.Done() + if e := c.Close(); e != nil { + errs[i] = e + } + }(i, c) + } + } + wg.Wait() + sb := strings.Builder{} + for _, e := range errs { + if e != nil { + sb.WriteString(e.Error()) + sb.WriteString("; ") + } + } + if sb.Len() > 0 { + err = errors.New(sb.String()) + } + return +} diff --git a/frontend/http/frontend.go b/frontend/http/frontend.go index c2c970c..3eea926 100644 --- a/frontend/http/frontend.go +++ b/frontend/http/frontend.go @@ -6,8 +6,10 @@ import ( "context" "crypto/tls" "errors" + "io" "net/http" "net/netip" + "sync" "time" "github.com/julienschmidt/httprouter" @@ -56,7 +58,7 @@ const ( // default values replacing anything that is invalid. func (cfg Config) Validate() (validCfg Config, err error) { validCfg = cfg - validCfg.ListenOptions = cfg.ListenOptions.Validate(false) + validCfg.ListenOptions = cfg.ListenOptions.Validate(false, logger) if cfg.UseTLS && (len(cfg.TLSCertPath) == 0 || len(cfg.TLSKeyPath) == 0) { err = errTLSNotProvided return @@ -88,14 +90,28 @@ func (cfg Config) Validate() (validCfg Config, err error) { Strs("default", validCfg.ScrapeRoutes). Msg("falling back to default configuration") } - validCfg.ParseOptions.ParseOptions = cfg.ParseOptions.ParseOptions.Validate() + validCfg.ParseOptions.ParseOptions = cfg.ParseOptions.ParseOptions.Validate(logger) + return +} + +// httpServer replaces http.Close method with http.Shutdown +type httpServer struct { + *http.Server +} + +func (c httpServer) Close() (err error) { + if c.Server != nil { + err = c.Shutdown(context.Background()) + } return } type httpFE struct { - srv *http.Server + servers []httpServer logic *middleware.Logic collectTimings bool + onceCloser sync.Once + ParseOptions } @@ -111,26 +127,35 @@ func NewFrontend(c conf.MapConfig, logic *middleware.Logic) (frontend.Frontend, } f := &httpFE{ - logic: logic, - srv: &http.Server{ - ReadTimeout: cfg.ReadTimeout, - ReadHeaderTimeout: cfg.ReadTimeout, - WriteTimeout: cfg.WriteTimeout, - IdleTimeout: cfg.IdleTimeout, - }, + logic: logic, + servers: make([]httpServer, cfg.Workers), collectTimings: cfg.EnableRequestTiming, ParseOptions: cfg.ParseOptions, } + for i := range f.servers { + f.servers[i] = httpServer{ + &http.Server{ + ReadTimeout: cfg.ReadTimeout, + ReadHeaderTimeout: cfg.ReadTimeout, + WriteTimeout: cfg.WriteTimeout, + IdleTimeout: cfg.IdleTimeout, + }, + } + } + // If TLS is enabled, create a key pair. if cfg.UseTLS { var cert tls.Certificate if cert, err = tls.LoadX509KeyPair(cfg.TLSCertPath, cfg.TLSKeyPath); err != nil { return nil, err } - f.srv.TLSConfig = &tls.Config{ - Certificates: []tls.Certificate{cert}, - MinVersion: tls.VersionTLS12, + certs := []tls.Certificate{cert} + for i := range f.servers { + f.servers[i].TLSConfig = &tls.Config{ + Certificates: certs, + MinVersion: tls.VersionTLS12, + } } } @@ -145,30 +170,41 @@ func NewFrontend(c conf.MapConfig, logic *middleware.Logic) (frontend.Frontend, router.GET(route, f.ping) router.HEAD(route, f.ping) } - f.srv.Handler = router - f.srv.SetKeepAlivesEnabled(cfg.EnableKeepAlive) - - go func() { - ln, err := cfg.ListenTCP() - if err == nil { - if f.srv.TLSConfig == nil { - err = f.srv.Serve(ln) - } else { - err = f.srv.ServeTLS(ln, "", "") - } - } - if !errors.Is(err, http.ErrServerClosed) { - logger.Fatal().Err(err).Msg("server failed") - } - }() + for _, srv := range f.servers { + srv.Handler = router + srv.SetKeepAlivesEnabled(cfg.EnableKeepAlive) + go runServer(srv, &cfg) + } return f, nil } -// Close provides a thread-safe way to shut down a currently running Frontend. -func (f *httpFE) Close() error { - return f.srv.Shutdown(context.Background()) +func runServer(s httpServer, cfg *Config) { + ln, err := cfg.ListenTCP() + if err == nil { + if s.TLSConfig == nil { + err = s.Serve(ln) + } else { + err = s.ServeTLS(ln, "", "") + } + } + if !errors.Is(err, http.ErrServerClosed) { + logger.Fatal().Err(err).Msg("server failed") + } +} + +// Close provides a thread-safe way to gracefully shut down a currently running Frontend. +func (f *httpFE) Close() (err error) { + f.onceCloser.Do(func() { + cls := make([]io.Closer, len(f.servers)) + for i, s := range f.servers { + cls[i] = s + } + err = frontend.CloseGroup(cls) + }) + + return } func httpParamsToRouteParams(in httprouter.Params) (out bittorrent.RouteParams) { diff --git a/frontend/options.go b/frontend/options.go index 6a4d064..c6ca37a 100644 --- a/frontend/options.go +++ b/frontend/options.go @@ -6,6 +6,7 @@ import ( "time" "github.com/libp2p/go-reuseport" + "github.com/sot-tech/mochi/pkg/log" ) const ( @@ -19,7 +20,8 @@ var errUnexpectedListenerType = errors.New("unexpected listener type") // ListenOptions is the base configuration which may be used in net listeners type ListenOptions struct { Addr string - ReusePort bool `cfg:"reuse_port"` + ReusePort bool `cfg:"reuse_port"` + Workers uint ReadTimeout time.Duration `cfg:"read_timeout"` WriteTimeout time.Duration `cfg:"write_timeout"` EnableRequestTiming bool `cfg:"enable_request_timing"` @@ -27,7 +29,7 @@ type ListenOptions struct { // Validate checks if listen address provided and sets default // timeout options if needed -func (lo ListenOptions) Validate(ignoreTimeouts bool) (validOptions ListenOptions) { +func (lo ListenOptions) Validate(ignoreTimeouts bool, logger *log.Logger) (validOptions ListenOptions) { validOptions = lo if len(lo.Addr) == 0 { validOptions.Addr = defaultListenAddress @@ -37,6 +39,13 @@ func (lo ListenOptions) Validate(ignoreTimeouts bool) (validOptions ListenOption Str("default", validOptions.Addr). Msg("falling back to default configuration") } + if lo.Workers == 0 { + validOptions.Workers = 1 + } + if lo.Workers > 1 && !lo.ReusePort { + validOptions.ReusePort = true + logger.Warn().Msg("forcibly enabling ReusePort because Workers > 1") + } if !ignoreTimeouts { if lo.ReadTimeout <= 0 { validOptions.ReadTimeout = defaultReadTimeout @@ -114,7 +123,7 @@ type ParseOptions struct { // Validate sanity checks values set in a config and returns a new config with // default values replacing anything that is invalid. -func (op ParseOptions) Validate() ParseOptions { +func (op ParseOptions) Validate(logger *log.Logger) ParseOptions { valid := op if op.MaxNumWant <= 0 { valid.MaxNumWant = defaultMaxNumWant diff --git a/frontend/udp/frontend.go b/frontend/udp/frontend.go index c8c6689..61c1a3e 100644 --- a/frontend/udp/frontend.go +++ b/frontend/udp/frontend.go @@ -7,6 +7,7 @@ import ( "context" "encoding/binary" "errors" + "io" "math/rand" "net" "net/netip" @@ -48,7 +49,7 @@ type Config struct { // default values replacing anything that is invalid. func (cfg Config) Validate() (validCfg Config) { validCfg = cfg - validCfg.ListenOptions = cfg.ListenOptions.Validate(true) + validCfg.ListenOptions = cfg.ListenOptions.Validate(true, logger) // Generate a private key if one isn't provided by the user. if cfg.PrivateKey == "" { @@ -65,14 +66,14 @@ func (cfg Config) Validate() (validCfg Config) { Msg("falling back to default configuration") } - validCfg.ParseOptions = cfg.ParseOptions.Validate() + validCfg.ParseOptions = cfg.ParseOptions.Validate(logger) return } // udpFE holds the state of a UDP BitTorrent Frontend. type udpFE struct { - socket *net.UDPConn + sockets []*net.UDPConn closing chan any wg sync.WaitGroup genPool *sync.Pool @@ -94,6 +95,7 @@ func NewFrontend(c conf.MapConfig, logic *middleware.Logic) (frontend.Frontend, cfg = cfg.Validate() f := &udpFE{ + sockets: make([]*net.UDPConn, cfg.Workers), closing: make(chan any), logic: logic, maxClockSkew: cfg.MaxClockSkew, @@ -106,30 +108,40 @@ func NewFrontend(c conf.MapConfig, logic *middleware.Logic) (frontend.Frontend, }, } - if f.socket, err = cfg.ListenUDP(); err == nil { - var ctx context.Context - ctx, f.ctxCancel = context.WithCancel(context.Background()) - f.wg.Add(1) - go func(ctx context.Context) { - if err := f.serve(ctx); err != nil { - logger.Fatal().Err(err).Msg("server failed") - } - }(ctx) + var ctx context.Context + ctx, f.ctxCancel = context.WithCancel(context.Background()) + for i := range f.sockets { + if f.sockets[i], err = cfg.ListenUDP(); err == nil { + f.wg.Add(1) + go func(socket *net.UDPConn, ctx context.Context) { + if err := f.serve(ctx, socket); err != nil { + logger.Fatal().Err(err).Msg("server failed") + } + }(f.sockets[i], ctx) + } + } + if err != nil { + _ = f.Close() } return f, err } // Close provides a thread-safe way to shut down a currently running Frontend. -func (t *udpFE) Close() (err error) { - t.onceCloser.Do(func() { - close(t.closing) - if t.socket != nil { - t.ctxCancel() - _ = t.socket.SetReadDeadline(time.Now()) - t.wg.Wait() - err = t.socket.Close() +func (f *udpFE) Close() (err error) { + f.onceCloser.Do(func() { + close(f.closing) + f.ctxCancel() + cls := make([]io.Closer, 0, len(f.sockets)) + now := time.Now() + for _, s := range f.sockets { + if s != nil { + _ = s.SetDeadline(now) + cls = append(cls, s) + } } + f.wg.Wait() + err = frontend.CloseGroup(cls) }) return @@ -137,14 +149,14 @@ func (t *udpFE) Close() (err error) { // serve blocks while listening and serving UDP BitTorrent requests // until Stop() is called or an error is returned. -func (t *udpFE) serve(ctx context.Context) error { +func (f *udpFE) serve(ctx context.Context, socket *net.UDPConn) error { pool := bytepool.NewBytePool(2048) - defer t.wg.Done() + defer f.wg.Done() for { // Check to see if we need shutdown. select { - case <-t.closing: + case <-f.closing: log.Debug().Msg("serve received shutdown signal") return nil default: @@ -152,7 +164,7 @@ func (t *udpFE) serve(ctx context.Context) error { // Read a UDP packet into a reusable buffer. buffer := pool.Get() - n, addrPort, err := t.socket.ReadFromUDPAddrPort(*buffer) + n, addrPort, err := socket.ReadFromUDPAddrPort(*buffer) if err != nil { pool.Put(buffer) var netErr net.Error @@ -169,22 +181,22 @@ func (t *udpFE) serve(ctx context.Context) error { continue } - t.wg.Add(1) + f.wg.Add(1) go func() { - defer t.wg.Done() + defer f.wg.Done() defer pool.Put(buffer) // Handle the request. addr := addrPort.Addr().Unmap() var start time.Time - if t.collectTimings && metrics.Enabled() { + if f.collectTimings && metrics.Enabled() { start = time.Now() } - action, err := t.handleRequest(ctx, + action, err := f.handleRequest(ctx, Request{(*buffer)[:n], addr}, - ResponseWriter{t.socket, addrPort}, + ResponseWriter{socket, addrPort}, ) - if t.collectTimings && metrics.Enabled() { + if f.collectTimings && metrics.Enabled() { recordResponseDuration(action, addr, err, time.Since(start)) } }() @@ -210,7 +222,7 @@ func (w ResponseWriter) Write(b []byte) (int, error) { } // handleRequest parses and responds to a UDP Request. -func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) (actionName string, err error) { +func (f *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) (actionName string, err error) { if len(r.Packet) < 16 { // Malformed, no client packets are less than 16 bytes. // We explicitly return nothing in case this is a DoS attempt. @@ -224,12 +236,12 @@ func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) txID := r.Packet[12:16] // get a connection ID generator/validator from the pool. - gen := t.genPool.Get().(*ConnectionIDGenerator) - defer t.genPool.Put(gen) + gen := f.genPool.Get().(*ConnectionIDGenerator) + defer f.genPool.Put(gen) // If this isn't requesting a new connection ID and the connection ID is // invalid, then fail. - if actionID != connectActionID && !gen.Validate(connID, r.IP, timecache.Now(), t.maxClockSkew) { + if actionID != connectActionID && !gen.Validate(connID, r.IP, timecache.Now(), f.maxClockSkew) { err = errBadConnectionID WriteError(w, txID, err) return @@ -251,7 +263,7 @@ func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) actionName = "announce" var req *bittorrent.AnnounceRequest - req, err = ParseAnnounce(r, actionID == announceV6ActionID, t.ParseOptions) + req, err = ParseAnnounce(r, actionID == announceV6ActionID, f.ParseOptions) if err != nil { WriteError(w, txID, err) return @@ -259,7 +271,7 @@ func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) var resp *bittorrent.AnnounceResponse ctx := bittorrent.InjectRouteParamsToContext(ctx, bittorrent.RouteParams{}) - ctx, resp, err = t.logic.HandleAnnounce(ctx, req) + ctx, resp, err = f.logic.HandleAnnounce(ctx, req) if err != nil { WriteError(w, txID, err) return @@ -268,13 +280,13 @@ func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) WriteAnnounce(w, txID, resp, actionID == announceV6ActionID, r.IP.Is6()) ctx = bittorrent.RemapRouteParamsToBgContext(ctx) - go t.logic.AfterAnnounce(ctx, req, resp) + go f.logic.AfterAnnounce(ctx, req, resp) case scrapeActionID: actionName = "scrape" var req *bittorrent.ScrapeRequest - req, err = ParseScrape(r, t.ParseOptions) + req, err = ParseScrape(r, f.ParseOptions) if err != nil { WriteError(w, txID, err) return @@ -282,7 +294,7 @@ func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) var resp *bittorrent.ScrapeResponse ctx := bittorrent.InjectRouteParamsToContext(ctx, bittorrent.RouteParams{}) - ctx, resp, err = t.logic.HandleScrape(ctx, req) + ctx, resp, err = f.logic.HandleScrape(ctx, req) if err != nil { WriteError(w, txID, err) return @@ -291,7 +303,7 @@ func (t *udpFE) handleRequest(ctx context.Context, r Request, w ResponseWriter) WriteScrape(w, txID, resp) ctx = bittorrent.RemapRouteParamsToBgContext(ctx) - go t.logic.AfterScrape(ctx, req, resp) + go f.logic.AfterScrape(ctx, req, resp) default: err = errUnknownAction diff --git a/pkg/randseed/rand_seed.go b/pkg/randseed/rand_seed.go index f752756..6f46c67 100644 --- a/pkg/randseed/rand_seed.go +++ b/pkg/randseed/rand_seed.go @@ -13,7 +13,7 @@ func init() { } // GenSeed returns 64bit seed from crypto/rand source or -// from current time, if crypto random error occurred +// from current time, if crypto random read error occurred func GenSeed() (seed int64) { r := make([]byte, 8) if _, err := cr.Read(r); err == nil { diff --git a/storage/storage.go b/storage/storage.go index 5b3aa73..7a4d0ae 100644 --- a/storage/storage.go +++ b/storage/storage.go @@ -44,7 +44,7 @@ func (c Config) sanitizeGCConfig() (gcInterval, peerTTL time.Duration) { if c.GarbageCollectionInterval <= 0 { gcInterval = DefaultGarbageCollectionInterval logger.Warn(). - Str("name", "garbageCollectionInterval"). + Str("name", "GarbageCollectionInterval"). Dur("provided", c.GarbageCollectionInterval). Dur("default", DefaultGarbageCollectionInterval). Msg("falling back to default configuration") @@ -54,7 +54,7 @@ func (c Config) sanitizeGCConfig() (gcInterval, peerTTL time.Duration) { if c.PeerLifetime <= 0 { peerTTL = DefaultPeerLifetime logger.Warn(). - Str("name", "peerLifetime"). + Str("name", "PeerLifetime"). Dur("provided", c.PeerLifetime). Dur("default", DefaultPeerLifetime). Msg("falling back to default configuration") @@ -68,7 +68,7 @@ func (c Config) sanitizeStatisticsConfig() (statInterval time.Duration) { if c.PrometheusReportingInterval < 0 { statInterval = DefaultPrometheusReportingInterval logger.Warn(). - Str("name", "prometheusReportingInterval"). + Str("name", "PrometheusReportingInterval"). Dur("provided", c.PrometheusReportingInterval). Dur("default", DefaultPrometheusReportingInterval). Msg("falling back to default configuration")