Skip to content

Commit 2b2a1fb

Browse files
byPixelTVbyPixelTV
committed
performance fixes
1 parent f91a183 commit 2b2a1fb

5 files changed

Lines changed: 439 additions & 24 deletions

File tree

‎main.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -63,7 +63,7 @@ func main() {
6363
return
6464
}
6565

66-
util.Logger.Info().Msg("Loaded " + strconv.Itoa(int(rune(len(Servers)))) + " servers from servers.json")
66+
util.Logger.Info().Msg("Loaded " + strconv.Itoa(len(Servers)) + " servers from servers.json")
6767

6868
go func() {
6969
if os.Getenv("DEPLOYMENT_MODE") == "production" || os.Getenv("DEPLOYMENT_MODE") == "release" {

‎task/pingJob.go‎

Lines changed: 56 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,8 @@ import (
66
"MineTracker/util"
77
"MineTracker/websocket"
88
"context"
9+
"net"
10+
"os"
911
"strconv"
1012
"strings"
1113
"sync"
@@ -22,11 +24,6 @@ const (
2224
influxQueueSize = 500
2325
bulkFlushSize = 300
2426
dbWriteQueueSize = 1000
25-
26-
// maxConcurrentPings caps simultaneous active ping calls.
27-
// Each call allocates a TCP connection + bufio buffer; keeping a hard cap
28-
// prevents a latency spike on slow servers from stacking up goroutines.
29-
maxConcurrentPings = 20
3027
)
3128

3229
type dbWriteOp struct {
@@ -35,6 +32,8 @@ type dbWriteOp struct {
3532

3633
var dbWriteQueue = make(chan dbWriteOp, dbWriteQueueSize)
3734

35+
var maxConcurrentPings = loadMaxConcurrentPings()
36+
3837
// pingLimit is a counting semaphore that limits concurrent TCP ping calls.
3938
var pingLimit = make(chan struct{}, maxConcurrentPings)
4039

@@ -69,6 +68,18 @@ func isServerActive(ip string) bool {
6968
}
7069

7170
func parseAddress(addr string) (host string, port *uint16) {
71+
if strings.Count(addr, ":") > 1 {
72+
if h, p, err := net.SplitHostPort(addr); err == nil {
73+
host = strings.Trim(h, "[]")
74+
parsed, convErr := strconv.ParseUint(p, 10, 16)
75+
if convErr == nil {
76+
pp := uint16(parsed)
77+
return host, &pp
78+
}
79+
}
80+
return addr, nil
81+
}
82+
7283
parts := strings.Split(addr, ":")
7384
host = parts[0]
7485

@@ -83,6 +94,30 @@ func parseAddress(addr string) (host string, port *uint16) {
8394
return
8495
}
8596

97+
func loadMaxConcurrentPings() int {
98+
const defaultMax = 96
99+
const minMax = 1
100+
const maxMax = 1000
101+
102+
raw := strings.TrimSpace(os.Getenv("PING_MAX_CONCURRENT"))
103+
if raw == "" {
104+
return defaultMax
105+
}
106+
107+
v, err := strconv.Atoi(raw)
108+
if err != nil {
109+
util.Logger.Warn().Str("PING_MAX_CONCURRENT", raw).Msg("Invalid PING_MAX_CONCURRENT, using default")
110+
return defaultMax
111+
}
112+
if v < minMax {
113+
return minMax
114+
}
115+
if v > maxMax {
116+
return maxMax
117+
}
118+
return v
119+
}
120+
86121
func portOrDefault(port *uint16, def uint16) uint16 {
87122
if port == nil {
88123
return def
@@ -177,12 +212,13 @@ func (j *PingJob) runServerLoop(ctx context.Context, server data.PingableServer)
177212
if !ok {
178213
return
179214
}
215+
prevInterval := lastInterval
180216
newInterval := getCurrentInterval()
181217
if newInterval != lastInterval {
182218
ticker.Reset(newInterval)
183219
lastInterval = newInterval
184220

185-
if newInterval < lastInterval && isServerActive(server.IP) {
221+
if newInterval < prevInterval && isServerActive(server.IP) {
186222
j.pingServer(server, pinger)
187223
}
188224
}
@@ -398,17 +434,22 @@ func (j *PingJob) pingServer(server data.PingableServer, pinger serverPinger) {
398434
if err != nil {
399435
serverCacheMu.Lock()
400436
existing, ok := serverCacheMap[server.IP]
401-
if ok {
402-
existing.Online = false
403-
serverCacheMap[server.IP] = existing
437+
if !ok {
438+
existing = data.Server{
439+
Name: server.Name,
440+
IP: server.IP,
441+
Type: server.Type,
442+
Active: true,
443+
}
404444
}
445+
existing.Online = false
446+
existing.PlayerCount = 0
447+
serverCacheMap[server.IP] = existing
405448
serverCacheMu.Unlock()
406449

407-
if ok {
408-
select {
409-
case dbWriteQueue <- dbWriteOp{server: existing}:
410-
default:
411-
}
450+
select {
451+
case dbWriteQueue <- dbWriteOp{server: existing}:
452+
default:
412453
}
413454
return
414455
}
@@ -437,9 +478,7 @@ func (j *PingJob) pingServer(server data.PingableServer, pinger serverPinger) {
437478
existing.Active = true
438479
}
439480

440-
if pc > 0 {
441-
existing.PlayerCount = pc
442-
}
481+
existing.PlayerCount = pc
443482
if pc > existing.Peak {
444483
existing.Peak = pc
445484
}

‎task/pingJob_test.go‎

Lines changed: 230 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,230 @@
1+
package task
2+
3+
import (
4+
"MineTracker/data"
5+
"errors"
6+
"os"
7+
"testing"
8+
"time"
9+
10+
"github.com/influxdata/influxdb-client-go/v2/api/write"
11+
)
12+
13+
type fakePinger struct {
14+
resp *mcPingResult
15+
err error
16+
called bool
17+
host string
18+
port uint16
19+
timeout time.Duration
20+
}
21+
22+
func (f *fakePinger) ping(host string, port uint16, timeout time.Duration) (*mcPingResult, error) {
23+
f.called = true
24+
f.host = host
25+
f.port = port
26+
f.timeout = timeout
27+
return f.resp, f.err
28+
}
29+
30+
func TestParseAddress(t *testing.T) {
31+
host, port := parseAddress("example.org:25570")
32+
if host != "example.org" {
33+
t.Fatalf("unexpected host: %s", host)
34+
}
35+
if port == nil || *port != 25570 {
36+
t.Fatalf("unexpected port: %v", port)
37+
}
38+
39+
host, port = parseAddress("example.org")
40+
if host != "example.org" {
41+
t.Fatalf("unexpected host without port: %s", host)
42+
}
43+
if port != nil {
44+
t.Fatalf("expected nil port for host without explicit port, got %v", *port)
45+
}
46+
47+
host, port = parseAddress("[2001:db8::1]:25565")
48+
if host != "2001:db8::1" {
49+
t.Fatalf("unexpected ipv6 host: %s", host)
50+
}
51+
if port == nil || *port != 25565 {
52+
t.Fatalf("unexpected ipv6 port: %v", port)
53+
}
54+
55+
host, port = parseAddress("2001:db8::1")
56+
if host != "2001:db8::1" {
57+
t.Fatalf("unexpected bare ipv6 host: %s", host)
58+
}
59+
if port != nil {
60+
t.Fatalf("expected nil port for bare ipv6, got %v", *port)
61+
}
62+
}
63+
64+
func TestLoadMaxConcurrentPings(t *testing.T) {
65+
old := os.Getenv("PING_MAX_CONCURRENT")
66+
defer func() {
67+
if old == "" {
68+
_ = os.Unsetenv("PING_MAX_CONCURRENT")
69+
return
70+
}
71+
_ = os.Setenv("PING_MAX_CONCURRENT", old)
72+
}()
73+
74+
_ = os.Unsetenv("PING_MAX_CONCURRENT")
75+
if got := loadMaxConcurrentPings(); got != 96 {
76+
t.Fatalf("default max concurrent should be 96, got %d", got)
77+
}
78+
79+
_ = os.Setenv("PING_MAX_CONCURRENT", "not-a-number")
80+
if got := loadMaxConcurrentPings(); got != 96 {
81+
t.Fatalf("invalid env should fallback to 96, got %d", got)
82+
}
83+
84+
_ = os.Setenv("PING_MAX_CONCURRENT", "0")
85+
if got := loadMaxConcurrentPings(); got != 1 {
86+
t.Fatalf("value below minimum should clamp to 1, got %d", got)
87+
}
88+
89+
_ = os.Setenv("PING_MAX_CONCURRENT", "5000")
90+
if got := loadMaxConcurrentPings(); got != 1000 {
91+
t.Fatalf("value above maximum should clamp to 1000, got %d", got)
92+
}
93+
94+
_ = os.Setenv("PING_MAX_CONCURRENT", "128")
95+
if got := loadMaxConcurrentPings(); got != 128 {
96+
t.Fatalf("expected explicit value 128, got %d", got)
97+
}
98+
}
99+
100+
func TestPortOrDefault(t *testing.T) {
101+
def := uint16(25565)
102+
if got := portOrDefault(nil, def); got != def {
103+
t.Fatalf("expected default port %d, got %d", def, got)
104+
}
105+
106+
p := uint16(25570)
107+
if got := portOrDefault(&p, def); got != 25570 {
108+
t.Fatalf("expected explicit port 25570, got %d", got)
109+
}
110+
}
111+
112+
func TestPingServerFailureMarksOfflineAndQueuesWrite(t *testing.T) {
113+
oldCache := serverCacheMap
114+
oldDB := dbWriteQueue
115+
oldInflux := influxQueue
116+
oldLimit := pingLimit
117+
oldDropped := droppedInfluxPoints
118+
defer func() {
119+
serverCacheMap = oldCache
120+
dbWriteQueue = oldDB
121+
influxQueue = oldInflux
122+
pingLimit = oldLimit
123+
droppedInfluxPoints = oldDropped
124+
}()
125+
126+
serverCacheMap = map[string]data.Server{}
127+
dbWriteQueue = make(chan dbWriteOp, 1)
128+
influxQueue = make(chan *write.Point, 1)
129+
pingLimit = make(chan struct{}, 1)
130+
droppedInfluxPoints = 0
131+
132+
job := &PingJob{}
133+
server := data.PingableServer{Name: "A", IP: "example.org:25565", Type: "java"}
134+
fake := &fakePinger{err: errors.New("dial timeout")}
135+
136+
job.pingServer(server, fake)
137+
138+
if !fake.called {
139+
t.Fatal("expected pinger to be called")
140+
}
141+
if fake.host != "example.org" || fake.port != 25565 {
142+
t.Fatalf("unexpected ping target %s:%d", fake.host, fake.port)
143+
}
144+
145+
entry, ok := serverCacheMap[server.IP]
146+
if !ok {
147+
t.Fatal("expected server cache entry")
148+
}
149+
if entry.Online {
150+
t.Fatal("expected offline server state")
151+
}
152+
if entry.PlayerCount != 0 {
153+
t.Fatalf("expected player count 0, got %d", entry.PlayerCount)
154+
}
155+
if !entry.Active {
156+
t.Fatal("expected newly created entry to default active=true")
157+
}
158+
159+
select {
160+
case op := <-dbWriteQueue:
161+
if op.server.IP != server.IP || op.server.Online {
162+
t.Fatalf("unexpected db write payload: %+v", op.server)
163+
}
164+
default:
165+
t.Fatal("expected db write operation to be queued")
166+
}
167+
}
168+
169+
func TestPingServerSuccessWritesZeroPlayerCount(t *testing.T) {
170+
oldCache := serverCacheMap
171+
oldDB := dbWriteQueue
172+
oldInflux := influxQueue
173+
oldLimit := pingLimit
174+
oldDropped := droppedInfluxPoints
175+
defer func() {
176+
serverCacheMap = oldCache
177+
dbWriteQueue = oldDB
178+
influxQueue = oldInflux
179+
pingLimit = oldLimit
180+
droppedInfluxPoints = oldDropped
181+
}()
182+
183+
serverIP := "example.org:25565"
184+
serverCacheMap = map[string]data.Server{
185+
serverIP: {
186+
Name: "Old",
187+
IP: serverIP,
188+
Type: "java",
189+
Online: false,
190+
PlayerCount: 5,
191+
Peak: 7,
192+
Active: true,
193+
},
194+
}
195+
dbWriteQueue = make(chan dbWriteOp, 1)
196+
influxQueue = make(chan *write.Point, 1)
197+
pingLimit = make(chan struct{}, 1)
198+
droppedInfluxPoints = 0
199+
200+
job := &PingJob{}
201+
server := data.PingableServer{Name: "New", IP: serverIP, Type: "java"}
202+
fake := &fakePinger{
203+
resp: &mcPingResult{PlayerCount: 0, Favicon: ""},
204+
}
205+
206+
job.pingServer(server, fake)
207+
208+
entry := serverCacheMap[serverIP]
209+
if !entry.Online {
210+
t.Fatal("expected online server state")
211+
}
212+
if entry.PlayerCount != 0 {
213+
t.Fatalf("expected player count to update to 0, got %d", entry.PlayerCount)
214+
}
215+
if entry.Peak != 7 {
216+
t.Fatalf("expected peak to stay 7, got %d", entry.Peak)
217+
}
218+
219+
select {
220+
case <-dbWriteQueue:
221+
default:
222+
t.Fatal("expected db write operation to be queued")
223+
}
224+
225+
select {
226+
case <-influxQueue:
227+
default:
228+
t.Fatal("expected influx point to be queued")
229+
}
230+
}

0 commit comments

Comments
 (0)