Skip to content

Commit 1b23e28

Browse files
improve write deadline for p2p data upload (#173)
1 parent bab32bc commit 1b23e28

File tree

2 files changed

+41
-15
lines changed

2 files changed

+41
-15
lines changed

p2p/kademlia/network.go

Lines changed: 40 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -627,20 +627,7 @@ func (s *Network) Call(ctx context.Context, request *Message, isLong bool) (*Mes
627627
// ---- retryable RPC helpers -------------------------------------------------
628628

629629
func (s *Network) rpcOnceWrapper(ctx context.Context, cw *connWrapper, remoteAddr string, data []byte, timeout time.Duration, msgType int) (*Message, error) {
630-
631-
sizeMB := float64(len(data)) / (1024.0 * 1024.0) // data is your gob-encoded message
632-
throughputFloor := 8.0 // MB/s (~64 Mbps)
633-
est := time.Duration(sizeMB / throughputFloor * float64(time.Second))
634-
base := 1 * time.Second
635-
cushion := 5 * time.Second
636-
637-
writeDL := base + est + cushion
638-
if writeDL < 5*time.Second {
639-
writeDL = 5 * time.Second
640-
}
641-
if writeDL > timeout-1*time.Second {
642-
writeDL = timeout - 1*time.Second
643-
}
630+
writeDL := calcWriteDeadline(timeout, len(data), 2.0) // target ~2 MB/s
644631

645632
retried := false
646633
for {
@@ -1430,3 +1417,42 @@ func readDeadlineFor(msgType int, overall time.Duration) time.Duration {
14301417
return overall // Bulk responses keep full budget
14311418
}
14321419
}
1420+
1421+
// calcWriteDeadline returns a conservative write deadline based on payload size.
1422+
// - targetMBps: assumed sustained throughput (lower = more lenient).
1423+
// - We reserve some headroom from overall timeout for server processing/response.
1424+
func calcWriteDeadline(timeout time.Duration, sizeBytes int, targetMBps float64) time.Duration {
1425+
if timeout <= 0 {
1426+
timeout = 30 * time.Second
1427+
}
1428+
// Leave headroom for server processing + response
1429+
const reserve = 8 * time.Second
1430+
maxBudget := timeout - reserve
1431+
if maxBudget < 5*time.Second {
1432+
maxBudget = timeout - 1*time.Second
1433+
if maxBudget < 3*time.Second {
1434+
maxBudget = 3 * time.Second
1435+
}
1436+
}
1437+
1438+
sizeMB := float64(sizeBytes) / (1024.0 * 1024.0)
1439+
base := 2 * time.Second
1440+
cushion := 5 * time.Second
1441+
1442+
// Softer floor: assume ~2 MB/s; increase if you like.
1443+
if targetMBps <= 0 {
1444+
targetMBps = 2.0
1445+
}
1446+
est := time.Duration(sizeMB / targetMBps * float64(time.Second))
1447+
1448+
writeDL := base + est + cushion
1449+
1450+
// Ensure a more generous minimum for big-ish payloads
1451+
if writeDL < 10*time.Second {
1452+
writeDL = 10 * time.Second
1453+
}
1454+
if writeDL > maxBudget {
1455+
writeDL = maxBudget
1456+
}
1457+
return writeDL
1458+
}

supernode/services/cascade/adaptors/p2p.go

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@ import (
2020
)
2121

2222
const (
23-
loadSymbolsBatchSize = 5000
23+
loadSymbolsBatchSize = 3000
2424
// Minimum first-pass coverage to store before returning from Register (percent)
2525
storeSymbolsPercent = 18
2626

0 commit comments

Comments
 (0)