Mirror/mochi
1
0
Fork 0
mirror of https://github.com/sot-tech/mochi.git synced 2026-07-18 14:18:10 -07:00
Code Issues Packages Projects Releases Wiki Activity

Add workers config parameter to start concurrent listeners

Browse Source
This commit is contained in:
Lawrence, Rendall
2022-11-25 13:58:04 +03:00
parent ae0c006cd2
commit 234c65333e
12 changed files with 191 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mochi-e2e/e2e.go
+2
View File
@@ -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() {
cmd/mochi/config.go
+3
View File
@@ -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"
cmd/mochi/main.go
+1
View File
@@ -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)")
dist/example_config.yaml
Vendored
+12 -4
View File
@@ -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
dist/example_config_pg.yaml
Vendored
+2
View File
@@ -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
dist/example_config_redis.yaml
Vendored
+2
View File
@@ -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
frontend/frontend.go
+33
View File
@@ -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
}
frontend/http/frontend.go
+68 -32
View File
@@ -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) {
frontend/options.go
+12 -3
View File
@@ -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
frontend/udp/frontend.go
+52 -40
View File
@@ -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
pkg/randseed/rand_seed.go
+1 -1
View File
@@ -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 {
storage/storage.go
+3 -3
View File
@@ -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")