removed dependency on fiber and solved memory leaks

This commit is contained in:
Michał Adamski
2025-01-29 23:44:48 +01:00
parent 57ba81575d
commit 0d411de4d4
9 changed files with 216 additions and 202 deletions

View File

@@ -1,5 +1,5 @@
build: build:
CGO_ENABLED=0 go build -ldflags="-s -w" -trimpath CGO_ENABLED=0 go build -ldflags="-s -w -buildid=" -trimpath
deploy: build deploy: build
rsync -avzL --exclude '*.fiber.gz' docs privtracker privtracker:web/ rsync -avzL --exclude '*.fiber.gz' docs privtracker privtracker:web/

View File

@@ -2,32 +2,16 @@ package main
import ( import (
"crypto/sha1" "crypto/sha1"
"fmt"
"net" "net"
"net/http"
"strconv"
"strings"
"time" "time"
"github.com/gofiber/fiber/v2"
"github.com/jackpal/bencode-go" "github.com/jackpal/bencode-go"
) )
type AnnounceRequest struct {
InfoHash string `query:"info_hash"`
PeerID string `query:"peer_id"`
IP string `query:"ip"`
Port uint16 `query:"port"`
Uploaded uint `query:"uploaded"`
Downloaded uint `query:"downloaded"`
Left uint `query:"left"`
Numwant uint `query:"numwant"`
Key string `query:"key"`
Compact bool `query:"compact"`
SupportCrypto bool `query:"supportcrypto"`
Event string `query:"event"`
}
func (req *AnnounceRequest) IsSeeding() bool {
return req.Left == 0
}
type AnnounceResponse struct { type AnnounceResponse struct {
Interval int `bencode:"interval"` Interval int `bencode:"interval"`
Complete int `bencode:"complete"` Complete int `bencode:"complete"`
@@ -36,38 +20,44 @@ type AnnounceResponse struct {
PeersIPv6 []byte `bencode:"peers_ipv6"` PeersIPv6 []byte `bencode:"peers_ipv6"`
} }
func announce(c *fiber.Ctx) error { func announce(w http.ResponseWriter, r *http.Request) {
var req AnnounceRequest query := r.URL.Query()
err := c.QueryParser(&req) port, err := strconv.Atoi(query.Get("port"))
if err != nil { if err != nil {
return err http.Error(w, "port missing", 400)
return
} }
ip := net.ParseIP(c.IP()) ip := getRemoteIP(r)
if ip == nil { if ip == nil {
ip = c.Context().RemoteIP() http.Error(w, "can't parse IP", 400)
return
} }
if req.Numwant < 1 { numwant, err := strconv.Atoi(query.Get("numwant"))
req.Numwant = 30 if err != nil || numwant < 1 {
numwant = 30
} }
swarmHash := sha1.Sum([]byte(c.Params("room") + req.InfoHash)) swarmHash := sha1.Sum([]byte(r.PathValue("room") + query.Get("info_hash")))
peer := NewPeer(ip, req.Port) peer := NewPeer(ip, uint16(port))
switch req.Event { isSeeding := query.Get("left") == "0"
switch query.Get("event") {
case "stopped": case "stopped":
DeletePeer(swarmHash, peer) DeletePeer(swarmHash, peer)
case "completed": case "completed":
GraduateLeecher(swarmHash, peer) GraduateLeecher(swarmHash, peer)
default: default:
PutPeer(swarmHash, peer, req.IsSeeding()) PutPeer(swarmHash, peer, isSeeding)
} }
peersIPv4, peersIPv6, numSeeders, numLeechers := GetPeers(swarmHash, peer, req.IsSeeding(), req.Numwant) peersIPv4, peersIPv6, numSeeders, numLeechers := GetPeers(swarmHash, peer, isSeeding, numwant)
interval := int(time.Now().Unix()+int64(swarmHash[0]))%256 + 60 interval := 480 // must be smaller than cleanup interval
switch { switch {
// case numSeeders == 0: case numSeeders+numLeechers < 10:
// interval -= 30 // try to synchronize peer requests. Maybe it will help µTP with UDP port punching... not sure
case numLeechers == 0: interval = 240 - int(time.Now().Unix()+int64(swarmHash[0]))%240
interval += 240 if interval < 60 {
case numSeeders+numLeechers > 10: interval += 240
interval += 480 }
case numSeeders+numLeechers > 30:
interval = 900
} }
resp := AnnounceResponse{ resp := AnnounceResponse{
Interval: interval, Interval: interval,
@@ -76,5 +66,27 @@ func announce(c *fiber.Ctx) error {
Peers: peersIPv4, Peers: peersIPv4,
PeersIPv6: peersIPv6, PeersIPv6: peersIPv6,
} }
return bencode.Marshal(c, resp) w.Header().Add("X-PrivTracker", fmt.Sprintf("s:%d l:%d", numSeeders, numLeechers))
if err := bencode.Marshal(w, resp); err != nil {
http.Error(w, err.Error(), 400)
}
}
func getRemoteIP(r *http.Request) net.IP {
addr := r.RemoteAddr
if colonIndex := strings.LastIndex(addr, ":"); colonIndex != -1 {
addr = addr[:colonIndex]
}
addr = strings.Trim(addr, "[]")
ip := net.ParseIP(addr)
if ip.IsPrivate() {
ips := strings.Split(r.Header.Get("X-Forwarded-For"), ",")
if len(ips) > 0 {
ipForwarded := net.ParseIP(strings.TrimSpace(ips[0]))
if ipForwarded != nil {
ip = ipForwarded
}
}
}
return ip
} }

17
announce_test.go Normal file
View File

@@ -0,0 +1,17 @@
package main
import (
"net/http/httptest"
"testing"
)
func BenchmarkAnnounce(b *testing.B) {
server := httptest.NewServer(router())
client := server.Client()
for i := 0; i < b.N; i++ {
_, err := client.Get(server.URL + "/test/announce?port=1234")
if err != nil {
b.Fatal(err)
}
}
}

14
go.mod
View File

@@ -1,25 +1,13 @@
module github.com/meehow/privtracker module github.com/meehow/privtracker
go 1.21.4 go 1.22
require ( require (
github.com/gofiber/fiber/v2 v2.52.5
github.com/jackpal/bencode-go v1.0.2 github.com/jackpal/bencode-go v1.0.2
golang.org/x/crypto v0.32.0 golang.org/x/crypto v0.32.0
) )
require ( require (
github.com/andybalholm/brotli v1.1.0 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/klauspost/compress v1.17.9 // indirect
github.com/mattn/go-colorable v0.1.13 // indirect
github.com/mattn/go-isatty v0.0.20 // indirect
github.com/mattn/go-runewidth v0.0.16 // indirect
github.com/rivo/uniseg v0.2.0 // indirect
github.com/valyala/bytebufferpool v1.0.0 // indirect
github.com/valyala/fasthttp v1.51.0 // indirect
github.com/valyala/tcplisten v1.0.0 // indirect
golang.org/x/net v0.21.0 // indirect golang.org/x/net v0.21.0 // indirect
golang.org/x/sys v0.29.0 // indirect
golang.org/x/text v0.21.0 // indirect golang.org/x/text v0.21.0 // indirect
) )

27
go.sum
View File

@@ -1,35 +1,8 @@
github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
github.com/gofiber/fiber/v2 v2.52.5 h1:tWoP1MJQjGEe4GB5TUGOi7P2E0ZMMRx5ZTG4rT+yGMo=
github.com/gofiber/fiber/v2 v2.52.5/go.mod h1:KEOE+cXMhXG0zHc9d8+E38hoX+ZN7bhOtgeF2oT6jrQ=
github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
github.com/jackpal/bencode-go v1.0.2 h1:LcCNfZ344u0LpBPOZNjpCLps/wUOuN4r87Fy9+5yU8g= github.com/jackpal/bencode-go v1.0.2 h1:LcCNfZ344u0LpBPOZNjpCLps/wUOuN4r87Fy9+5yU8g=
github.com/jackpal/bencode-go v1.0.2/go.mod h1:6jI9mUjO3GQbZti3JizEfxTzRfWOM8oBBcwbwlTfceI= github.com/jackpal/bencode-go v1.0.2/go.mod h1:6jI9mUjO3GQbZti3JizEfxTzRfWOM8oBBcwbwlTfceI=
github.com/klauspost/compress v1.17.9 h1:6KIumPrER1LHsvBVuDa0r5xaG0Es51mhhB9BQB2qeMA=
github.com/klauspost/compress v1.17.9/go.mod h1:Di0epgTjJY877eYKx5yC51cX2A2Vl2ibi7bDH9ttBbw=
github.com/mattn/go-colorable v0.1.13 h1:fFA4WZxdEF4tXPZVKMLwD8oUnCTTo08duU7wxecdEvA=
github.com/mattn/go-colorable v0.1.13/go.mod h1:7S9/ev0klgBDR4GtXTXX8a3vIGJpMovkB8vQcUbaXHg=
github.com/mattn/go-isatty v0.0.16/go.mod h1:kYGgaQfpe5nmfYZH+SKPsOc2e4SrIfOl2e/yFXSvRLM=
github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
github.com/mattn/go-runewidth v0.0.16 h1:E5ScNMtiwvlvB5paMFdw9p4kSQzbXFikJ5SQO6TULQc=
github.com/mattn/go-runewidth v0.0.16/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY=
github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc=
github.com/valyala/bytebufferpool v1.0.0 h1:GqA5TC/0021Y/b9FG4Oi9Mr3q7XYx6KllzawFIhcdPw=
github.com/valyala/bytebufferpool v1.0.0/go.mod h1:6bBcMArwyJ5K/AmCkWv1jt77kVWyCJ6HpOuEn7z0Csc=
github.com/valyala/fasthttp v1.51.0 h1:8b30A5JlZ6C7AS81RsWjYMQmrZG6feChmgAolCl1SqA=
github.com/valyala/fasthttp v1.51.0/go.mod h1:oI2XroL+lI7vdXyYoQk03bXBThfFl2cVdIA3Xl7cH8g=
github.com/valyala/tcplisten v1.0.0 h1:rBHj/Xf+E1tRGZyWIWwJDiRY0zc1Js+CV5DqwacVSA8=
github.com/valyala/tcplisten v1.0.0/go.mod h1:T0xQ8SeCZGxckz9qRXTfG43PvQ/mcWh7FwZEA7Ioqkc=
golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc= golang.org/x/crypto v0.32.0 h1:euUpcYgM8WcP71gNpTqQCn6rC2t6ULUPiOzfWaXVVfc=
golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc= golang.org/x/crypto v0.32.0/go.mod h1:ZnnJkOaASj8g0AjIduWNlq2NRxL0PlBrbKVyZ6V/Ugc=
golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4= golang.org/x/net v0.21.0 h1:AQyQV4dYCvJ7vGmJyKki9+PBdyvhkSd8EIx/qb0AYv4=
golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
golang.org/x/sys v0.29.0 h1:TPYlXGxvx1MGTn2GiZDhnjPA9wZzZeGKHHmKhHYvgaU=
golang.org/x/sys v0.29.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo= golang.org/x/text v0.21.0 h1:zyQAAkrwaneQ066sspRyJaG9VNi/YJ1NfzcGB3hZ/qo=
golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ=

93
main.go
View File

@@ -5,64 +5,37 @@ import (
"fmt" "fmt"
"log" "log"
"net" "net"
"net/http"
"os" "os"
"path/filepath" "path/filepath"
"time"
"github.com/gofiber/fiber/v2"
"github.com/gofiber/fiber/v2/middleware/logger"
"github.com/gofiber/fiber/v2/middleware/monitor"
"github.com/gofiber/fiber/v2/middleware/recover"
"golang.org/x/crypto/acme/autocert" "golang.org/x/crypto/acme/autocert"
) )
func main() { func main() {
port := os.Getenv("PORT") port := os.Getenv("PORT")
tlsEnabled := port == "443"
if port == "" { if port == "" {
port = "1337" port = "1337"
} }
config := fiber.Config{ handler := router(recoveryMiddleware, headersMiddleware, logRequestMiddleware)
AppName: "PrivTracker", if port == "443" {
ServerHeader: "PrivTracker", go redirect80()
ReadTimeout: time.Second * 245, fmt.Println("PrivTracker listening on https://0.0.0.0/")
WriteTimeout: time.Second * 30, log.Fatal(http.Serve(autocertListener(), handler))
Network: fiber.NetworkTCP,
GETOnly: true,
DisableKeepalive: true,
Immutable: true,
}
// if you disable TLS, then I guess you want to use existing proxy
if !tlsEnabled {
config.EnableTrustedProxyCheck = true
config.TrustedProxies = []string{"127.0.0.1", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"}
config.ProxyHeader = fiber.HeaderXForwardedFor
}
go Cleanup(time.Minute * 16)
app := fiber.New(config)
app.Use(recover.New())
// app.Use(pprof.New())
app.Use(myLogger())
app.Use(hsts)
app.Static("/", "docs", fiber.Static{
MaxAge: 3600 * 24 * 7,
Compress: true,
CacheDuration: time.Hour,
})
app.Get("/dashboard", monitor.New())
app.Get("/:room/announce", announce)
app.Get("/:room/scrape", scrape)
app.Server().LogAllErrors = true
if tlsEnabled {
go redirect80(config)
log.Fatal(app.Listener(autocertListener()))
} else { } else {
log.Fatal(app.Listen(":" + port)) fmt.Printf("PrivTracker listening on http://0.0.0.0:%s/\n", port)
log.Fatal(http.ListenAndServe(":"+port, handler))
} }
} }
func router(middlewares ...Middleware) http.Handler {
mux := http.NewServeMux()
mux.Handle("/", http.FileServer(http.Dir("docs")))
mux.HandleFunc("GET /{room}/announce", announce)
mux.HandleFunc("GET /{room}/scrape", scrape)
return chainMiddleware(mux, middlewares...)
}
func autocertListener() net.Listener { func autocertListener() net.Listener {
homeDir, err := os.UserHomeDir() homeDir, err := os.UserHomeDir()
if err != nil { if err != nil {
@@ -75,33 +48,21 @@ func autocertListener() net.Listener {
} }
cfg := &tls.Config{ cfg := &tls.Config{
GetCertificate: m.GetCertificate, GetCertificate: m.GetCertificate,
NextProtos: []string{ NextProtos: []string{"h2", "http/1.1", "acme-tls/1"},
"http/1.1", "acme-tls/1",
},
} }
ln, err := tls.Listen("tcp", ":443", cfg) listener, err := tls.Listen("tcp", ":443", cfg)
if err != nil { if err != nil {
log.Fatal(err) log.Fatal(err)
} }
return ln return listener
} }
func redirect80(config fiber.Config) { func redirect80() {
config.DisableStartupMessage = true err := http.ListenAndServe(":80", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
app := fiber.New(config) url := fmt.Sprintf("https://%s/", r.Host)
app.Use(func(c *fiber.Ctx) error { http.Redirect(w, r, url, http.StatusMovedPermanently)
return c.Redirect(fmt.Sprintf("https://%s/", c.Hostname()), fiber.StatusMovedPermanently) }))
}) if err != nil {
log.Print(app.Listen(":80")) fmt.Println(err)
} }
func myLogger() fiber.Handler {
loggerConfig := logger.ConfigDefault
loggerConfig.Format = "${status} - ${latency} ${ip} ${method} ${path} ${bytesSent} - ${referer} - ${ua}\n"
return logger.New(loggerConfig)
}
func hsts(c *fiber.Ctx) error {
c.Set("Strict-Transport-Security", "max-age=31536000")
return c.Next()
} }

67
middleware.go Normal file
View File

@@ -0,0 +1,67 @@
package main
import (
"fmt"
"net/http"
"runtime/debug"
"time"
)
type Middleware func(http.Handler) http.Handler
type ResponseWriterWrapper struct {
http.ResponseWriter
StatusCode int
BytesSent int
}
func (rw *ResponseWriterWrapper) WriteHeader(code int) {
rw.StatusCode = code
rw.ResponseWriter.WriteHeader(code)
}
func (rw *ResponseWriterWrapper) Write(data []byte) (int, error) {
bytesWritten, err := rw.ResponseWriter.Write(data)
rw.BytesSent += bytesWritten
return bytesWritten, err
}
func logRequestMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
wrapper := &ResponseWriterWrapper{ResponseWriter: w, StatusCode: http.StatusOK}
timer := time.Now()
next.ServeHTTP(wrapper, r)
fmt.Printf("%d - %5dµs %47s %4s %42s %6d %11s - %s - %s\n",
wrapper.StatusCode, time.Since(timer).Microseconds(), r.RemoteAddr,
r.Method, r.URL.Path, wrapper.BytesSent,
wrapper.Header().Get("X-PrivTracker"), r.Referer(), r.UserAgent())
})
}
func headersMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Strict-Transport-Security", "max-age=31536000") // hsts
w.Header().Set("Server", "PrivTracker")
next.ServeHTTP(w, r)
})
}
func recoveryMiddleware(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
defer func() {
if err := recover(); err != nil {
fmt.Printf("Recovered from panic: %v\n%s\n", err, debug.Stack())
http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
}
}()
next.ServeHTTP(w, r)
})
}
func chainMiddleware(handler http.Handler, middlewares ...Middleware) http.Handler {
// Apply middlewares in reverse order (outermost to innermost)
for i := len(middlewares) - 1; i >= 0; i-- {
handler = middlewares[i](handler)
}
return handler
}

View File

@@ -2,15 +2,12 @@ package main
import ( import (
"crypto/sha1" "crypto/sha1"
"fmt"
"net/http"
"github.com/gofiber/fiber/v2"
"github.com/jackpal/bencode-go" "github.com/jackpal/bencode-go"
) )
type ScrapeRequest struct {
InfoHash string `query:"info_hash"`
}
type ScrapeResponse struct { type ScrapeResponse struct {
Files map[string]Stat `bencode:"files"` Files map[string]Stat `bencode:"files"`
} }
@@ -21,21 +18,22 @@ type Stat struct {
// Downloaded uint `bencode:"downloaded"` // Downloaded uint `bencode:"downloaded"`
} }
func scrape(c *fiber.Ctx) error { func scrape(w http.ResponseWriter, r *http.Request) {
var req ScrapeRequest query := r.URL.Query()
err := c.QueryParser(&req) infoHash := query.Get("info_hash")
if err != nil { swarmHash := sha1.Sum([]byte(r.PathValue("room") + infoHash))
return err
}
swarmHash := sha1.Sum([]byte(c.Params("room") + req.InfoHash))
numSeeders, numLeechers := GetStats(swarmHash) numSeeders, numLeechers := GetStats(swarmHash)
resp := ScrapeResponse{ resp := ScrapeResponse{
Files: map[string]Stat{ Files: map[string]Stat{
req.InfoHash: { infoHash: {
Complete: numSeeders, Complete: numSeeders,
Incomplete: numLeechers, Incomplete: numLeechers,
}, },
}, },
} }
return bencode.Marshal(c, resp) w.Header().Add("X-PrivTracker", fmt.Sprintf("s:%d l:%d", numSeeders, numLeechers))
if err := bencode.Marshal(w, resp); err != nil {
http.Error(w, err.Error(), 400)
}
} }

View File

@@ -5,9 +5,7 @@ import (
"crypto/sha1" "crypto/sha1"
"encoding/binary" "encoding/binary"
"fmt" "fmt"
"log"
"net" "net"
"runtime"
"sync" "sync"
"time" "time"
) )
@@ -15,11 +13,11 @@ import (
type Hash [sha1.Size]byte // we use sha1 and we are not affraid of hash collisions type Hash [sha1.Size]byte // we use sha1 and we are not affraid of hash collisions
type Peer [18]byte // 16 bytes for IP and 2 bytes for port number type Peer [18]byte // 16 bytes for IP and 2 bytes for port number
var shards = NewShards(256)
var v4InV6Prefix = []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff} var v4InV6Prefix = []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff}
var shards [256]*shard
type shard struct { type shard struct {
swarms map[Hash]Swarm swarms map[Hash]*Swarm
sync.RWMutex sync.RWMutex
} }
@@ -28,8 +26,17 @@ type Swarm struct {
leechers map[Peer]int64 leechers map[Peer]int64
} }
func NewSwarm() Swarm { func init() {
return Swarm{ for i := range shards {
shards[i] = &shard{
swarms: make(map[Hash]*Swarm),
}
}
go Cleanup(time.Minute * 16) // needs to be bigget than biggest interval in announce.go
}
func NewSwarm() *Swarm {
return &Swarm{
seeders: make(map[Peer]int64), seeders: make(map[Peer]int64),
leechers: make(map[Peer]int64), leechers: make(map[Peer]int64),
} }
@@ -47,23 +54,15 @@ func (peer Peer) String() string {
return fmt.Sprintf("%s:%d", ip, port) return fmt.Sprintf("%s:%d", ip, port)
} }
func shardIndex(hash Hash) int { func shardIndex(hash Hash) uint8 {
return int(binary.BigEndian.Uint16(hash[:2])) % len(shards) // return int(binary.BigEndian.Uint16(hash[:2])) % len(shards)
} return hash[0] // works only with 256 shards
func NewShards(size int) []*shard {
shards := make([]*shard, size)
for i := 0; i < size; i++ {
shards[i] = &shard{
swarms: make(map[Hash]Swarm),
}
}
return shards
} }
func PutPeer(h Hash, peer Peer, seeding bool) { func PutPeer(h Hash, peer Peer, seeding bool) {
shard := shards[shardIndex(h)] shard := shards[shardIndex(h)]
shard.Lock() shard.Lock()
defer shard.Unlock()
if _, ok := shard.swarms[h]; !ok { if _, ok := shard.swarms[h]; !ok {
shard.swarms[h] = NewSwarm() shard.swarms[h] = NewSwarm()
} }
@@ -72,34 +71,40 @@ func PutPeer(h Hash, peer Peer, seeding bool) {
} else { } else {
shard.swarms[h].leechers[peer] = time.Now().Unix() shard.swarms[h].leechers[peer] = time.Now().Unix()
} }
shard.Unlock()
} }
func DeletePeer(h Hash, peer Peer) { func DeletePeer(h Hash, peer Peer) {
shard := shards[shardIndex(h)] shard := shards[shardIndex(h)]
shard.Lock() shard.Lock()
defer shard.Unlock()
if _, ok := shard.swarms[h]; !ok { if _, ok := shard.swarms[h]; !ok {
return return
} }
delete(shard.swarms[h].seeders, peer) delete(shard.swarms[h].seeders, peer)
delete(shard.swarms[h].leechers, peer) delete(shard.swarms[h].leechers, peer)
shard.Unlock()
} }
func GraduateLeecher(h Hash, peer Peer) { func GraduateLeecher(h Hash, peer Peer) {
shard := shards[shardIndex(h)] shard := shards[shardIndex(h)]
shard.Lock() shard.Lock()
defer shard.Unlock()
if _, ok := shard.swarms[h]; !ok { if _, ok := shard.swarms[h]; !ok {
shard.swarms[h] = NewSwarm() shard.swarms[h] = NewSwarm()
} }
shard.swarms[h].seeders[peer] = time.Now().Unix() shard.swarms[h].seeders[peer] = time.Now().Unix()
delete(shard.swarms[h].leechers, peer) delete(shard.swarms[h].leechers, peer)
shard.Unlock()
} }
func GetPeers(h Hash, client Peer, seeding bool, numWant uint) (peersIPv4, peersIPv6 []byte, numSeeders, numLeechers int) { func GetPeers(h Hash, client Peer, seeding bool, numWant int) (peersIPv4, peersIPv6 []byte, numSeeders, numLeechers int) {
shard := shards[shardIndex(h)] shard := shards[shardIndex(h)]
shard.RLock() shard.RLock()
defer shard.RUnlock()
if _, ok := shard.swarms[h]; !ok {
return
}
numSeeders = len(shard.swarms[h].seeders)
numLeechers = len(shard.swarms[h].leechers)
// seeders don't need other seeders // seeders don't need other seeders
if !seeding { if !seeding {
@@ -107,6 +112,9 @@ func GetPeers(h Hash, client Peer, seeding bool, numWant uint) (peersIPv4, peers
if numWant == 0 { if numWant == 0 {
break break
} }
if peer == client {
continue
}
if bytes.HasPrefix(peer[:], v4InV6Prefix) { if bytes.HasPrefix(peer[:], v4InV6Prefix) {
peersIPv4 = append(peersIPv4, peer[len(v4InV6Prefix):]...) peersIPv4 = append(peersIPv4, peer[len(v4InV6Prefix):]...)
} else { } else {
@@ -116,66 +124,56 @@ func GetPeers(h Hash, client Peer, seeding bool, numWant uint) (peersIPv4, peers
} }
} }
for peer := range shard.swarms[h].leechers { for peer := range shard.swarms[h].leechers {
if peer == client {
continue
}
if numWant == 0 { if numWant == 0 {
break break
} }
if peer == client {
continue
}
if bytes.HasPrefix(peer[:], v4InV6Prefix) { if bytes.HasPrefix(peer[:], v4InV6Prefix) {
peersIPv4 = append(peersIPv4, peer[12:]...) peersIPv4 = append(peersIPv4, peer[len(v4InV6Prefix):]...)
} else { } else {
peersIPv6 = append(peersIPv6, peer[:]...) peersIPv6 = append(peersIPv6, peer[:]...)
} }
numWant-- numWant--
} }
numSeeders = len(shard.swarms[h].seeders)
numLeechers = len(shard.swarms[h].leechers)
shard.RUnlock()
return return
} }
func GetStats(h Hash) (numSeeders, numLeechers int) { func GetStats(h Hash) (numSeeders, numLeechers int) {
shard := shards[shardIndex(h)] shard := shards[shardIndex(h)]
shard.RLock() shard.RLock()
defer shard.RUnlock()
if _, ok := shard.swarms[h]; !ok {
return
}
numSeeders = len(shard.swarms[h].seeders) numSeeders = len(shard.swarms[h].seeders)
numLeechers = len(shard.swarms[h].leechers) numLeechers = len(shard.swarms[h].leechers)
shard.RUnlock()
return return
} }
func Cleanup(duration time.Duration) { func Cleanup(duration time.Duration) {
ticker := time.NewTicker(duration) ticker := time.NewTicker(duration)
for range ticker.C { for range ticker.C {
var seeders, leechers, swarms, seedersDeleted, leechersDeleted, swarmsDeleted int
expiration := time.Now().Unix() - int64(duration.Seconds()) expiration := time.Now().Unix() - int64(duration.Seconds())
for _, shard := range shards { for _, shard := range shards {
shard.Lock() shard.Lock()
swarms += len(shard.swarms)
for h, swarm := range shard.swarms { for h, swarm := range shard.swarms {
seeders += len(swarm.seeders)
leechers += len(swarm.leechers)
for peer, lastSeen := range swarm.seeders { for peer, lastSeen := range swarm.seeders {
if lastSeen < expiration { if lastSeen < expiration {
seedersDeleted++
delete(swarm.seeders, peer) delete(swarm.seeders, peer)
} }
} }
for peer, lastSeen := range swarm.leechers { for peer, lastSeen := range swarm.leechers {
if lastSeen < expiration { if lastSeen < expiration {
leechersDeleted++
delete(swarm.leechers, peer) delete(swarm.leechers, peer)
} }
} }
if len(swarm.leechers) == 0 && len(swarm.seeders) == 0 { if len(swarm.leechers) == 0 && len(swarm.seeders) == 0 {
swarmsDeleted++
delete(shard.swarms, h) delete(shard.swarms, h)
} }
} }
shard.Unlock() shard.Unlock()
} }
log.Printf("seeders: %d (%d deleted), leechers: %d (%d deleted), swarms: %d (%d deleted)",
seeders, seedersDeleted, leechers, leechersDeleted, swarms, swarmsDeleted)
runtime.GC()
} }
} }