aheh/MasterHttpRelayVPN
1
0
Fork 0
mirror of https://github.com/masterking32/MasterHttpRelayVPN.git synced 2026-05-18 23:54:37 +03:00
Code Issues Packages Projects Releases Wiki Activity

Add config-driven transport randomization for mux, ping timing, and HTTP reuse

Browse Source
This commit is contained in:
Amin.MasterkinG
2026-04-21 15:51:20 +03:30
parent 74ff27ac78
commit ab7365e35d
5 changed files with 414 additions and 98 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/client/client.go
+2 -2
View File
@@ -196,11 +196,11 @@ func (c *Client) idleIntervalForStreak(streak int64) time.Duration {
if interval > c.cfg.PingMaxIntervalMS {
interval = c.cfg.PingMaxIntervalMS
}
return time.Duration(interval) * time.Millisecond
return c.pingIntervalWithJitter(time.Duration(interval) * time.Millisecond)
}
func (c *Client) scheduleAggressivePing(now time.Time) {
c.nextPingDueUnixMS.Store(now.Add(time.Duration(c.cfg.IdlePollIntervalMS) * time.Millisecond).UnixMilli())
c.nextPingDueUnixMS.Store(now.Add(c.pingIntervalWithJitter(time.Duration(c.cfg.IdlePollIntervalMS) * time.Millisecond)).UnixMilli())
}
func (c *Client) setPingState(state int32) {
internal/client/sender_workers.go
+105 -9
View File
@@ -18,12 +18,16 @@ import (
"sync"
"time"
"masterhttprelayvpn/internal/config"
"masterhttprelayvpn/internal/protocol"
)
type sendWorker struct {
id int
httpClient *http.Client
id int
httpClient *http.Client
httpTransport *http.Transport
transportUseCount int
transportReuseLimit int
}
type dequeuedPacket struct {
@@ -35,10 +39,8 @@ func (c *Client) startSendWorkers(ctx context.Context, wg *sync.WaitGroup) {
for i := 0; i < c.cfg.WorkerCount; i++ {
worker := &sendWorker{
id: i + 1,
httpClient: &http.Client{
Timeout: time.Duration(c.cfg.HTTPRequestTimeoutMS) * time.Millisecond,
},
}
worker.resetHTTPClient(c.cfg)
wg.Add(1)
go func(w *sendWorker) {
@@ -137,12 +139,15 @@ func (c *Client) buildNextBatch(connections []*SOCKSConnection, totalQueuedBytes
})
}
if len(connections) > 1 {
rotationEvery := c.cfg.MuxRotateEveryBatches
rotationEvery := c.effectiveMuxRotateEveryBatches()
if rotationEvery < 1 {
rotationEvery = 1
}
turn := c.batchCursor.Add(1) - 1
start = int((turn / uint64(rotationEvery)) % uint64(len(connections)))
if offset := c.randomMuxStartOffset(len(connections)); offset > 0 {
start = (start + offset) % len(connections)
}
}
maxPackets, maxBatchBytes := c.effectiveBatchLimits(totalQueuedBytes)
maxPerSOCKS := c.cfg.MaxPacketsPerSOCKSPerBatch
@@ -245,7 +250,7 @@ func (c *Client) shouldSendPing(connections []*SOCKSConnection, totalQueuedBytes
func (c *Client) effectiveBatchLimits(totalQueuedBytes int) (int, int) {
maxPackets := c.cfg.MaxPacketsPerBatch
maxBatchBytes := c.cfg.MaxBatchBytes
if totalQueuedBytes < c.cfg.MuxBurstThresholdBytes {
if totalQueuedBytes < c.effectiveBurstThresholdBytes() {
if reducedPackets := maxPackets / 2; reducedPackets >= 1 {
maxPackets = reducedPackets
}
@@ -276,7 +281,7 @@ func (c *Client) effectiveBatchLimits(totalQueuedBytes int) (int, int) {
func (c *Client) effectiveWaitInterval(totalQueuedBytes int) time.Duration {
interval := time.Duration(c.cfg.WorkerPollIntervalMS) * time.Millisecond
if totalQueuedBytes >= c.cfg.MuxBurstThresholdBytes {
if totalQueuedBytes >= c.effectiveBurstThresholdBytes() {
if burst := interval / 2; burst >= 25*time.Millisecond {
return burst
}
@@ -286,7 +291,7 @@ func (c *Client) effectiveWaitInterval(totalQueuedBytes int) time.Duration {
}
func (c *Client) effectiveConcurrentBatches(totalQueuedBytes int) int {
if totalQueuedBytes >= c.cfg.MuxBurstThresholdBytes {
if totalQueuedBytes >= c.effectiveBurstThresholdBytes() {
return c.cfg.MaxConcurrentBatches
}
return 1
@@ -362,6 +367,46 @@ func (c *Client) jitterDuration(base time.Duration) time.Duration {
return base + jitter
}
func (c *Client) pingIntervalWithJitter(base time.Duration) time.Duration {
if base <= 0 || !c.cfg.HTTPRandomizeTransport || c.cfg.PingIntervalJitterMS <= 0 {
return base
}
jitter := time.Duration(randomIndex(c.cfg.PingIntervalJitterMS+1)) * time.Millisecond
return base + jitter
}
func (c *Client) effectiveBurstThresholdBytes() int {
threshold := c.cfg.MuxBurstThresholdBytes
if !c.cfg.HTTPRandomizeTransport || c.cfg.MuxBurstThresholdJitterBytes <= 0 {
return threshold
}
delta := randomIndex(c.cfg.MuxBurstThresholdJitterBytes + 1)
if randomIndex(2) == 0 {
if adjusted := threshold - delta; adjusted >= c.cfg.MaxChunkSize {
return adjusted
}
return c.cfg.MaxChunkSize
}
return threshold + delta
}
func (c *Client) effectiveMuxRotateEveryBatches() int {
rotationEvery := c.cfg.MuxRotateEveryBatches
if !c.cfg.HTTPRandomizeTransport || c.cfg.MuxRotateJitterBatches <= 0 {
return rotationEvery
}
return rotationEvery + randomIndex(c.cfg.MuxRotateJitterBatches+1)
}
func (c *Client) randomMuxStartOffset(connectionCount int) int {
if !c.cfg.HTTPRandomizeTransport || connectionCount <= 1 {
return 0
}
return randomIndex(connectionCount)
}
func (c *Client) requeueSelected(selected []dequeuedPacket) {
grouped := make(map[*SOCKSConnection][]string)
for _, entry := range selected {
@@ -450,6 +495,7 @@ func (w *sendWorker) postBatch(ctx context.Context, c *Client, batch protocol.Ba
}
resp, err := w.httpClient.Do(req)
w.recordTransportUse(c.cfg)
if err != nil {
if pingOnly {
c.failPing()
@@ -512,6 +558,56 @@ func (w *sendWorker) postBatch(ctx context.Context, c *Client, batch protocol.Ba
return nil
}
func (w *sendWorker) resetHTTPClient(cfg config.Config) {
transport := &http.Transport{
Proxy: http.ProxyFromEnvironment,
ForceAttemptHTTP2: true,
MaxIdleConns: 32,
MaxIdleConnsPerHost: 8,
IdleConnTimeout: w.randomizedIdleConnTimeout(cfg),
}
w.httpTransport = transport
w.httpClient = &http.Client{
Timeout: time.Duration(cfg.HTTPRequestTimeoutMS) * time.Millisecond,
Transport: transport,
}
w.transportUseCount = 0
w.transportReuseLimit = w.nextTransportReuseLimit(cfg)
}
func (w *sendWorker) recordTransportUse(cfg config.Config) {
if !cfg.HTTPRandomizeTransport {
return
}
w.transportUseCount++
if w.transportReuseLimit > 0 && w.transportUseCount >= w.transportReuseLimit {
if w.httpTransport != nil {
w.httpTransport.CloseIdleConnections()
}
w.resetHTTPClient(cfg)
}
}
func (w *sendWorker) randomizedIdleConnTimeout(cfg config.Config) time.Duration {
minTimeout := cfg.HTTPIdleConnTimeoutMinMS
maxTimeout := cfg.HTTPIdleConnTimeoutMaxMS
if !cfg.HTTPRandomizeTransport || maxTimeout <= minTimeout {
return time.Duration(minTimeout) * time.Millisecond
}
return time.Duration(minTimeout+randomIndex(maxTimeout-minTimeout+1)) * time.Millisecond
}
func (w *sendWorker) nextTransportReuseLimit(cfg config.Config) int {
if !cfg.HTTPRandomizeTransport || cfg.HTTPTransportReuseMax <= cfg.HTTPTransportReuseMin {
return cfg.HTTPTransportReuseMin
}
return cfg.HTTPTransportReuseMin + randomIndex(cfg.HTTPTransportReuseMax-cfg.HTTPTransportReuseMin+1)
}
func (c *Client) applyResponseBatch(batch protocol.Batch) error {
for _, packet := range batch.Packets {
if packet.Type == protocol.PacketTypePong {
internal/client/sender_workers_test.go
+54
View File
@@ -27,6 +27,10 @@ func testClientConfig() config.Config {
PingMaxIntervalMS: 60000,
MaxQueueBytesPerSOCKS: 4096,
HTTPBatchRandomize: false,
HTTPIdleConnTimeoutMinMS: 15000,
HTTPIdleConnTimeoutMaxMS: 45000,
HTTPTransportReuseMin: 8,
HTTPTransportReuseMax: 24,
}
}
@@ -254,6 +258,56 @@ func TestEffectiveConcurrentBatchesUsesBurstThreshold(t *testing.T) {
}
}
func TestEffectiveBurstThresholdBytesStaysWithinConfiguredJitterRange(t *testing.T) {
cfg := testClientConfig()
cfg.HTTPRandomizeTransport = true
cfg.MuxBurstThresholdBytes = 4096
cfg.MuxBurstThresholdJitterBytes = 512
client := New(cfg, nil)
for i := 0; i < 50; i++ {
got := client.effectiveBurstThresholdBytes()
if got < cfg.MaxChunkSize {
t.Fatalf("expected threshold >= max chunk size, got %d", got)
}
if got < cfg.MuxBurstThresholdBytes-cfg.MuxBurstThresholdJitterBytes || got > cfg.MuxBurstThresholdBytes+cfg.MuxBurstThresholdJitterBytes {
t.Fatalf("threshold %d outside jitter range", got)
}
}
}
func TestPingIntervalWithJitterStaysWithinConfiguredRange(t *testing.T) {
cfg := testClientConfig()
cfg.HTTPRandomizeTransport = true
cfg.PingIntervalJitterMS = 250
client := New(cfg, nil)
base := 2 * time.Second
for i := 0; i < 50; i++ {
got := client.pingIntervalWithJitter(base)
if got < base || got > base+250*time.Millisecond {
t.Fatalf("ping interval %v outside expected jitter range", got)
}
}
}
func TestSendWorkerTransportReuseLimitStaysWithinConfiguredRange(t *testing.T) {
cfg := testClientConfig()
cfg.HTTPRandomizeTransport = true
cfg.HTTPTransportReuseMin = 3
cfg.HTTPTransportReuseMax = 7
cfg.HTTPIdleConnTimeoutMinMS = 1000
cfg.HTTPIdleConnTimeoutMaxMS = 2000
worker := &sendWorker{id: 1}
for i := 0; i < 50; i++ {
limit := worker.nextTransportReuseLimit(cfg)
if limit < cfg.HTTPTransportReuseMin || limit > cfg.HTTPTransportReuseMax {
t.Fatalf("reuse limit %d outside expected range", limit)
}
}
}
func TestBuildPollBatchSkipsWhenTransportBusy(t *testing.T) {
cfg := testClientConfig()
client := New(cfg, nil)