mirror of
https://github.com/XTLS/Xray-core.git
synced 2026-11-06 10:12:40 +00:00
Compare commits
10
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 RPRX
|
bb335c6b0c | ||
|
|
043d2e47fc | ||
|
|
36756ad672 | ||
|
MeowandRPRX
|
abecd7d9f1 | ||
|
Wu ShuwenandGitHub
|
a52db8f6ed | ||
|
MinyandGitHub
|
858c44f410 | ||
|
Hossin AsaadiandGitHub
|
fff4acac52 | ||
|
|
ad98ffd42b | ||
|
MeowandGitHub
|
63ca2e0c1d | ||
|
|
ab88e0118f |
@@ -20,7 +20,7 @@ RUN cat <<EOF >/tmp/usr/local/etc/xray/00_log.json
|
||||
{
|
||||
"log": {
|
||||
"error": "/var/log/xray/error.log",
|
||||
"loglevel": "warning",
|
||||
"logLevel": "warning",
|
||||
"access": "none",
|
||||
"dnsLog": false
|
||||
}
|
||||
|
||||
@@ -20,7 +20,7 @@ RUN cat <<EOF >/tmp/usr/local/etc/xray/00_log.json
|
||||
{
|
||||
"log": {
|
||||
"error": "/var/log/xray/error.log",
|
||||
"loglevel": "warning",
|
||||
"logLevel": "warning",
|
||||
"access": "none",
|
||||
"dnsLog": false
|
||||
}
|
||||
|
||||
@@ -122,8 +122,8 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
|
||||
if config.Concurrency < 0 {
|
||||
h.mux = &mux.ClientManager{Enabled: false}
|
||||
}
|
||||
if config.Concurrency == 0 {
|
||||
config.Concurrency = 8 // same as before
|
||||
if config.Concurrency >= 0 && config.Concurrency <= 128 {
|
||||
config.Concurrency = 128
|
||||
}
|
||||
if config.Concurrency > 0 {
|
||||
h.mux = &mux.ClientManager{
|
||||
@@ -134,7 +134,7 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
|
||||
Dialer: h,
|
||||
Strategy: mux.ClientStrategy{
|
||||
MaxConcurrency: uint32(config.Concurrency),
|
||||
MaxConnection: 128,
|
||||
MaxConnection: 1024,
|
||||
},
|
||||
},
|
||||
},
|
||||
@@ -155,7 +155,7 @@ func NewHandler(ctx context.Context, config *core.OutboundHandlerConfig) (outbou
|
||||
Dialer: h,
|
||||
Strategy: mux.ClientStrategy{
|
||||
MaxConcurrency: uint32(config.XudpConcurrency),
|
||||
MaxConnection: 128,
|
||||
MaxConnection: 1024,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
||||
+1
-1
@@ -73,7 +73,7 @@ func (r *TimeoutWrapperReader) ReadMultiBufferTimeout(duration time.Duration) (M
|
||||
}
|
||||
return r.mb, r.err
|
||||
case <-timeout:
|
||||
return nil, nil
|
||||
return nil, ErrReadTimeout
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -134,7 +134,7 @@ func TestMarshalConfigJson(t *testing.T) {
|
||||
func getConfig() string {
|
||||
return `{
|
||||
"log": {
|
||||
"loglevel": "debug"
|
||||
"logLevel": "debug"
|
||||
},
|
||||
"stats": {},
|
||||
"policy": {
|
||||
|
||||
@@ -10,7 +10,7 @@ require (
|
||||
github.com/google/go-cmp v0.7.0
|
||||
github.com/google/uuid v1.6.0
|
||||
github.com/gorilla/websocket v1.5.3
|
||||
github.com/klauspost/cpuid/v2 v2.4.1
|
||||
github.com/klauspost/cpuid/v2 v2.4.2
|
||||
github.com/libp2p/go-nat v1.0.1-0.20250821073202-01afc089f138
|
||||
github.com/miekg/dns v1.1.73
|
||||
github.com/pelletier/go-toml v1.9.5
|
||||
|
||||
@@ -35,8 +35,8 @@ github.com/juju/ratelimit v1.0.2 h1:sRxmtRiajbvrcLQT7S+JbqU0ntsb9W2yhSdNN8tWfaI=
|
||||
github.com/juju/ratelimit v1.0.2/go.mod h1:qapgC/Gy+xNh9UxzV13HGGl/6UXNN+ct+vwSgWNm/qk=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/klauspost/cpuid/v2 v2.4.1 h1:I9KxadAkvo+QjtcEE2f2tyfEvmIla/T0m+14PrZNl0U=
|
||||
github.com/klauspost/cpuid/v2 v2.4.1/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU=
|
||||
github.com/klauspost/cpuid/v2 v2.4.2 h1:xRw7IZEsAgVCp6trS/DgWbTAqpixhmywgVgBjYGg1bQ=
|
||||
github.com/klauspost/cpuid/v2 v2.4.2/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU=
|
||||
github.com/koron/go-ssdp v0.0.4 h1:1IDwrghSKYM7yLf7XCzbByg2sJ/JcNOZRXS2jczTwz0=
|
||||
github.com/koron/go-ssdp v0.0.4/go.mod h1:oDXq+E5IL5q0U8uSBcoAXzTzInwy5lEgC91HoKtbmZk=
|
||||
github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
|
||||
|
||||
+33
-3
@@ -9,6 +9,7 @@ import (
|
||||
|
||||
"github.com/xtls/xray-core/common/errors"
|
||||
"github.com/xtls/xray-core/common/geodata"
|
||||
"github.com/xtls/xray-core/common/log"
|
||||
xnet "github.com/xtls/xray-core/common/net"
|
||||
"github.com/xtls/xray-core/common/protocol"
|
||||
"github.com/xtls/xray-core/proxy/freedom"
|
||||
@@ -48,6 +49,8 @@ type FreedomFinalRuleConfig struct {
|
||||
Network *NetworkList `json:"network"`
|
||||
Port *PortList `json:"port"`
|
||||
IP *StringList `json:"ip"`
|
||||
Delay *Int32Range `json:"delay"`
|
||||
LogLevel string `json:"logLevel"`
|
||||
BlockDelay *Int32Range `json:"blockDelay"`
|
||||
}
|
||||
|
||||
@@ -280,10 +283,37 @@ func (c *FreedomFinalRuleConfig) Build() (*freedom.FinalRuleConfig, error) {
|
||||
}
|
||||
|
||||
if c.BlockDelay != nil {
|
||||
rule.BlockDelay = &freedom.Range{
|
||||
Min: uint64(c.BlockDelay.From),
|
||||
Max: uint64(c.BlockDelay.To),
|
||||
errors.PrintDeprecatedFeatureWarning("finalRules.blockDelay", "finalRules.delay")
|
||||
c.Delay = c.BlockDelay
|
||||
}
|
||||
|
||||
if rule.Action != freedom.RuleAction_Block {
|
||||
if c.Delay != nil || c.LogLevel != "" {
|
||||
return nil, errors.New("delay and logLevel are only supported for block finalRules")
|
||||
}
|
||||
return rule, nil
|
||||
}
|
||||
|
||||
if c.Delay != nil {
|
||||
rule.Delay = &freedom.Range{
|
||||
Min: uint64(c.Delay.From),
|
||||
Max: uint64(c.Delay.To),
|
||||
}
|
||||
}
|
||||
|
||||
switch strings.ToLower(c.LogLevel) {
|
||||
case "", "info":
|
||||
rule.LogLevel = log.Severity_Info.Enum()
|
||||
case "none":
|
||||
rule.LogLevel = log.Severity_Unknown.Enum()
|
||||
case "debug":
|
||||
rule.LogLevel = log.Severity_Debug.Enum()
|
||||
case "warning":
|
||||
rule.LogLevel = log.Severity_Warning.Enum()
|
||||
case "error":
|
||||
rule.LogLevel = log.Severity_Error.Enum()
|
||||
default:
|
||||
return nil, errors.New("unknown finalRules.logLevel: ", c.LogLevel)
|
||||
}
|
||||
|
||||
return rule, nil
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"testing"
|
||||
|
||||
"github.com/xtls/xray-core/common/geodata"
|
||||
"github.com/xtls/xray-core/common/log"
|
||||
"github.com/xtls/xray-core/common/net"
|
||||
"github.com/xtls/xray-core/common/protocol"
|
||||
. "github.com/xtls/xray-core/infra/conf"
|
||||
@@ -46,7 +47,8 @@ func TestFreedomConfig(t *testing.T) {
|
||||
"network": "tcp,udp",
|
||||
"port": "53,443",
|
||||
"ip": ["10.0.0.0/8", "2001:db8::/32"],
|
||||
"blockDelay": "30-60"
|
||||
"delay": "30-60",
|
||||
"logLevel": "debug"
|
||||
}, {
|
||||
"action": "allow",
|
||||
"network": ["udp"]
|
||||
@@ -86,10 +88,8 @@ func TestFreedomConfig(t *testing.T) {
|
||||
},
|
||||
},
|
||||
},
|
||||
BlockDelay: &freedom.Range{
|
||||
Min: 30,
|
||||
Max: 60,
|
||||
},
|
||||
Delay: &freedom.Range{Min: 30, Max: 60},
|
||||
LogLevel: log.Severity_Debug.Enum(),
|
||||
},
|
||||
{
|
||||
Action: freedom.RuleAction_Allow,
|
||||
|
||||
+1
-1
@@ -18,7 +18,7 @@ func DefaultLogConfig() *log.Config {
|
||||
type LogConfig struct {
|
||||
AccessLog string `json:"access"`
|
||||
ErrorLog string `json:"error"`
|
||||
LogLevel string `json:"loglevel"`
|
||||
LogLevel string `json:"logLevel"`
|
||||
DNSLog bool `json:"dnsLog"`
|
||||
MaskAddress string `json:"maskAddress"`
|
||||
}
|
||||
|
||||
@@ -18,7 +18,7 @@ func TestLoaderError(t *testing.T) {
|
||||
"log": {
|
||||
// abcd
|
||||
0,
|
||||
"loglevel": "info"
|
||||
"logLevel": "info"
|
||||
}
|
||||
}`,
|
||||
Output: "line 4 char 6",
|
||||
@@ -27,7 +27,7 @@ func TestLoaderError(t *testing.T) {
|
||||
Input: `{
|
||||
"log": {
|
||||
// abcd
|
||||
"loglevel": "info",
|
||||
"logLevel": "info",
|
||||
}
|
||||
}`,
|
||||
Output: "line 5 char 5",
|
||||
|
||||
@@ -42,7 +42,7 @@ func TestXrayConfig(t *testing.T) {
|
||||
Input: `{
|
||||
"log": {
|
||||
"access": "/var/log/xray/access.log",
|
||||
"loglevel": "error",
|
||||
"logLevel": "error",
|
||||
"error": "/var/log/xray/error.log"
|
||||
},
|
||||
"inbounds": [{
|
||||
|
||||
+34
-20
@@ -8,6 +8,7 @@ package freedom
|
||||
|
||||
import (
|
||||
geodata "github.com/xtls/xray-core/common/geodata"
|
||||
log "github.com/xtls/xray-core/common/log"
|
||||
net "github.com/xtls/xray-core/common/net"
|
||||
protocol "github.com/xtls/xray-core/common/protocol"
|
||||
internet "github.com/xtls/xray-core/transport/internet"
|
||||
@@ -357,7 +358,8 @@ type FinalRuleConfig struct {
|
||||
Networks []net.Network `protobuf:"varint,2,rep,packed,name=networks,proto3,enum=xray.common.net.Network" json:"networks,omitempty"`
|
||||
PortList *net.PortList `protobuf:"bytes,3,opt,name=port_list,json=portList,proto3" json:"port_list,omitempty"`
|
||||
Ip []*geodata.IPRule `protobuf:"bytes,4,rep,name=ip,proto3" json:"ip,omitempty"`
|
||||
BlockDelay *Range `protobuf:"bytes,5,opt,name=block_delay,json=blockDelay,proto3" json:"block_delay,omitempty"`
|
||||
Delay *Range `protobuf:"bytes,5,opt,name=delay,proto3" json:"delay,omitempty"`
|
||||
LogLevel *log.Severity `protobuf:"varint,6,opt,name=log_level,json=logLevel,proto3,enum=xray.common.log.Severity,oneof" json:"log_level,omitempty"`
|
||||
unknownFields protoimpl.UnknownFields
|
||||
sizeCache protoimpl.SizeCache
|
||||
}
|
||||
@@ -420,13 +422,20 @@ func (x *FinalRuleConfig) GetIp() []*geodata.IPRule {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *FinalRuleConfig) GetBlockDelay() *Range {
|
||||
func (x *FinalRuleConfig) GetDelay() *Range {
|
||||
if x != nil {
|
||||
return x.BlockDelay
|
||||
return x.Delay
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *FinalRuleConfig) GetLogLevel() log.Severity {
|
||||
if x != nil && x.LogLevel != nil {
|
||||
return *x.LogLevel
|
||||
}
|
||||
return log.Severity(0)
|
||||
}
|
||||
|
||||
type Config struct {
|
||||
state protoimpl.MessageState `protogen:"open.v1"`
|
||||
DomainStrategy internet.DomainStrategy `protobuf:"varint,1,opt,name=domain_strategy,json=domainStrategy,proto3,enum=xray.transport.internet.DomainStrategy" json:"domain_strategy,omitempty"`
|
||||
@@ -523,7 +532,7 @@ var File_proxy_freedom_config_proto protoreflect.FileDescriptor
|
||||
|
||||
const file_proxy_freedom_config_proto_rawDesc = "" +
|
||||
"\n" +
|
||||
"\x1aproxy/freedom/config.proto\x12\x12xray.proxy.freedom\x1a!common/protocol/server_spec.proto\x1a\x1ftransport/internet/config.proto\x1a\x15common/net/port.proto\x1a\x18common/net/network.proto\x1a\x1bcommon/geodata/geodat.proto\"S\n" +
|
||||
"\x1aproxy/freedom/config.proto\x12\x12xray.proxy.freedom\x1a!common/protocol/server_spec.proto\x1a\x1ftransport/internet/config.proto\x1a\x15common/net/port.proto\x1a\x18common/net/network.proto\x1a\x1bcommon/geodata/geodat.proto\x1a\x14common/log/log.proto\"S\n" +
|
||||
"\x13DestinationOverride\x12<\n" +
|
||||
"\x06server\x18\x01 \x01(\v2$.xray.common.protocol.ServerEndpointR\x06server\"\x98\x02\n" +
|
||||
"\bFragment\x12!\n" +
|
||||
@@ -549,14 +558,16 @@ const file_proxy_freedom_config_proto_rawDesc = "" +
|
||||
"\bapply_to\x18\x06 \x01(\tR\aapplyTo\"+\n" +
|
||||
"\x05Range\x12\x10\n" +
|
||||
"\x03min\x18\x01 \x01(\x04R\x03min\x12\x10\n" +
|
||||
"\x03max\x18\x02 \x01(\x04R\x03max\"\xa0\x02\n" +
|
||||
"\x03max\x18\x02 \x01(\x04R\x03max\"\xe0\x02\n" +
|
||||
"\x0fFinalRuleConfig\x126\n" +
|
||||
"\x06action\x18\x01 \x01(\x0e2\x1e.xray.proxy.freedom.RuleActionR\x06action\x124\n" +
|
||||
"\bnetworks\x18\x02 \x03(\x0e2\x18.xray.common.net.NetworkR\bnetworks\x126\n" +
|
||||
"\tport_list\x18\x03 \x01(\v2\x19.xray.common.net.PortListR\bportList\x12+\n" +
|
||||
"\x02ip\x18\x04 \x03(\v2\x1b.xray.common.geodata.IPRuleR\x02ip\x12:\n" +
|
||||
"\vblock_delay\x18\x05 \x01(\v2\x19.xray.proxy.freedom.RangeR\n" +
|
||||
"blockDelay\"\xaf\x03\n" +
|
||||
"\x02ip\x18\x04 \x03(\v2\x1b.xray.common.geodata.IPRuleR\x02ip\x12/\n" +
|
||||
"\x05delay\x18\x05 \x01(\v2\x19.xray.proxy.freedom.RangeR\x05delay\x12;\n" +
|
||||
"\tlog_level\x18\x06 \x01(\x0e2\x19.xray.common.log.SeverityH\x00R\blogLevel\x88\x01\x01B\f\n" +
|
||||
"\n" +
|
||||
"_log_level\"\xaf\x03\n" +
|
||||
"\x06Config\x12P\n" +
|
||||
"\x0fdomain_strategy\x18\x01 \x01(\x0e2'.xray.transport.internet.DomainStrategyR\x0edomainStrategy\x12Z\n" +
|
||||
"\x14destination_override\x18\x03 \x01(\v2'.xray.proxy.freedom.DestinationOverrideR\x13destinationOverride\x12\x1d\n" +
|
||||
@@ -599,7 +610,8 @@ var file_proxy_freedom_config_proto_goTypes = []any{
|
||||
(net.Network)(0), // 8: xray.common.net.Network
|
||||
(*net.PortList)(nil), // 9: xray.common.net.PortList
|
||||
(*geodata.IPRule)(nil), // 10: xray.common.geodata.IPRule
|
||||
(internet.DomainStrategy)(0), // 11: xray.transport.internet.DomainStrategy
|
||||
(log.Severity)(0), // 11: xray.common.log.Severity
|
||||
(internet.DomainStrategy)(0), // 12: xray.transport.internet.DomainStrategy
|
||||
}
|
||||
var file_proxy_freedom_config_proto_depIdxs = []int32{
|
||||
7, // 0: xray.proxy.freedom.DestinationOverride.server:type_name -> xray.common.protocol.ServerEndpoint
|
||||
@@ -607,17 +619,18 @@ var file_proxy_freedom_config_proto_depIdxs = []int32{
|
||||
8, // 2: xray.proxy.freedom.FinalRuleConfig.networks:type_name -> xray.common.net.Network
|
||||
9, // 3: xray.proxy.freedom.FinalRuleConfig.port_list:type_name -> xray.common.net.PortList
|
||||
10, // 4: xray.proxy.freedom.FinalRuleConfig.ip:type_name -> xray.common.geodata.IPRule
|
||||
4, // 5: xray.proxy.freedom.FinalRuleConfig.block_delay:type_name -> xray.proxy.freedom.Range
|
||||
11, // 6: xray.proxy.freedom.Config.domain_strategy:type_name -> xray.transport.internet.DomainStrategy
|
||||
1, // 7: xray.proxy.freedom.Config.destination_override:type_name -> xray.proxy.freedom.DestinationOverride
|
||||
2, // 8: xray.proxy.freedom.Config.fragment:type_name -> xray.proxy.freedom.Fragment
|
||||
3, // 9: xray.proxy.freedom.Config.noises:type_name -> xray.proxy.freedom.Noise
|
||||
5, // 10: xray.proxy.freedom.Config.final_rules:type_name -> xray.proxy.freedom.FinalRuleConfig
|
||||
11, // [11:11] is the sub-list for method output_type
|
||||
11, // [11:11] is the sub-list for method input_type
|
||||
11, // [11:11] is the sub-list for extension type_name
|
||||
11, // [11:11] is the sub-list for extension extendee
|
||||
0, // [0:11] is the sub-list for field type_name
|
||||
4, // 5: xray.proxy.freedom.FinalRuleConfig.delay:type_name -> xray.proxy.freedom.Range
|
||||
11, // 6: xray.proxy.freedom.FinalRuleConfig.log_level:type_name -> xray.common.log.Severity
|
||||
12, // 7: xray.proxy.freedom.Config.domain_strategy:type_name -> xray.transport.internet.DomainStrategy
|
||||
1, // 8: xray.proxy.freedom.Config.destination_override:type_name -> xray.proxy.freedom.DestinationOverride
|
||||
2, // 9: xray.proxy.freedom.Config.fragment:type_name -> xray.proxy.freedom.Fragment
|
||||
3, // 10: xray.proxy.freedom.Config.noises:type_name -> xray.proxy.freedom.Noise
|
||||
5, // 11: xray.proxy.freedom.Config.final_rules:type_name -> xray.proxy.freedom.FinalRuleConfig
|
||||
12, // [12:12] is the sub-list for method output_type
|
||||
12, // [12:12] is the sub-list for method input_type
|
||||
12, // [12:12] is the sub-list for extension type_name
|
||||
12, // [12:12] is the sub-list for extension extendee
|
||||
0, // [0:12] is the sub-list for field type_name
|
||||
}
|
||||
|
||||
func init() { file_proxy_freedom_config_proto_init() }
|
||||
@@ -625,6 +638,7 @@ func file_proxy_freedom_config_proto_init() {
|
||||
if File_proxy_freedom_config_proto != nil {
|
||||
return
|
||||
}
|
||||
file_proxy_freedom_config_proto_msgTypes[4].OneofWrappers = []any{}
|
||||
type x struct{}
|
||||
out := protoimpl.TypeBuilder{
|
||||
File: protoimpl.DescBuilder{
|
||||
|
||||
@@ -11,6 +11,7 @@ import "transport/internet/config.proto";
|
||||
import "common/net/port.proto";
|
||||
import "common/net/network.proto";
|
||||
import "common/geodata/geodat.proto";
|
||||
import "common/log/log.proto";
|
||||
|
||||
message DestinationOverride {
|
||||
xray.common.protocol.ServerEndpoint server = 1;
|
||||
@@ -51,7 +52,8 @@ message FinalRuleConfig {
|
||||
repeated xray.common.net.Network networks = 2;
|
||||
xray.common.net.PortList port_list = 3;
|
||||
repeated xray.common.geodata.IPRule ip = 4;
|
||||
Range block_delay = 5;
|
||||
Range delay = 5;
|
||||
optional xray.common.log.Severity log_level = 6;
|
||||
}
|
||||
|
||||
message Config {
|
||||
|
||||
+32
-10
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/xtls/xray-core/common/dice"
|
||||
"github.com/xtls/xray-core/common/errors"
|
||||
"github.com/xtls/xray-core/common/geodata"
|
||||
"github.com/xtls/xray-core/common/log"
|
||||
"github.com/xtls/xray-core/common/net"
|
||||
"github.com/xtls/xray-core/common/platform"
|
||||
"github.com/xtls/xray-core/common/retry"
|
||||
@@ -75,11 +76,13 @@ func init() {
|
||||
action: RuleAction_Block,
|
||||
network: allNetworks,
|
||||
ip: geodata.GetPrivateIPMatcher(),
|
||||
logLevel: log.Severity_Warning.Enum(),
|
||||
}
|
||||
|
||||
defaultBlockAllRule = &FinalRule{
|
||||
action: RuleAction_Block,
|
||||
network: allNetworks,
|
||||
logLevel: log.Severity_Warning.Enum(),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -88,7 +91,8 @@ type FinalRule struct {
|
||||
network [8]bool
|
||||
port net.MemoryPortList
|
||||
ip geodata.IPMatcher
|
||||
blockDelay *Range
|
||||
delay *Range
|
||||
logLevel *log.Severity
|
||||
}
|
||||
|
||||
// Handler handles Freedom connections.
|
||||
@@ -103,7 +107,8 @@ type Handler struct {
|
||||
func buildFinalRule(config *FinalRuleConfig) (*FinalRule, error) {
|
||||
rule := &FinalRule{
|
||||
action: config.GetAction(),
|
||||
blockDelay: config.GetBlockDelay(),
|
||||
delay: config.GetDelay(),
|
||||
logLevel: config.LogLevel,
|
||||
}
|
||||
|
||||
if len(config.Networks) == 0 {
|
||||
@@ -164,7 +169,7 @@ func getDefaultFinalRule(inbound *session.Inbound) *FinalRule {
|
||||
switch inbound.Name {
|
||||
case "vless-reverse":
|
||||
return defaultBlockAllRule
|
||||
case "vless", "vmess", "trojan", "hysteria", "wireguard":
|
||||
case "vless", "vmess", "trojan", "hysteria", "wireguard", "masque":
|
||||
return defaultBlockPrivateRule
|
||||
default:
|
||||
if strings.HasPrefix(inbound.Name, "shadowsocks") {
|
||||
@@ -212,12 +217,12 @@ func (h *Handler) policy() policy.Session {
|
||||
return p
|
||||
}
|
||||
|
||||
func (h *Handler) blockDelay(rule *FinalRule) time.Duration {
|
||||
func (r *FinalRule) blockDuration() time.Duration {
|
||||
min := uint64(30)
|
||||
max := uint64(90)
|
||||
if rule.blockDelay != nil {
|
||||
min = rule.blockDelay.Min
|
||||
max = rule.blockDelay.Max
|
||||
if r.delay != nil {
|
||||
min = r.delay.Min
|
||||
max = r.delay.Max
|
||||
}
|
||||
span := max - min
|
||||
if max < min {
|
||||
@@ -226,9 +231,26 @@ func (h *Handler) blockDelay(rule *FinalRule) time.Duration {
|
||||
return time.Duration(min+uint64(dice.Roll(int(span+1)))) * time.Second
|
||||
}
|
||||
|
||||
func (r *FinalRule) log(ctx context.Context, msg ...interface{}) {
|
||||
level := log.Severity_Info
|
||||
if r.logLevel != nil {
|
||||
level = *r.logLevel
|
||||
}
|
||||
switch level {
|
||||
case log.Severity_Debug:
|
||||
errors.LogDebug(ctx, msg...)
|
||||
case log.Severity_Info:
|
||||
errors.LogInfo(ctx, msg...)
|
||||
case log.Severity_Warning:
|
||||
errors.LogWarning(ctx, msg...)
|
||||
case log.Severity_Error:
|
||||
errors.LogError(ctx, msg...)
|
||||
}
|
||||
}
|
||||
|
||||
func (h *Handler) blackhole(ctx context.Context, input buf.Reader, output buf.Writer, rule *FinalRule, dest *net.Destination) error {
|
||||
delay := h.blockDelay(rule)
|
||||
errors.LogInfo(ctx, "blocked target: ", *dest, ", blackholing connection for ", delay)
|
||||
delay := rule.blockDuration()
|
||||
rule.log(ctx, "target blocked by a finalrule: ", *dest, ", blackholing connection for ", delay)
|
||||
defer common.Close(output)
|
||||
done := make(chan struct{})
|
||||
go func() { // a DispatchLink inbound's reader ignores Interrupt, so don't wait for this copy
|
||||
@@ -240,7 +262,7 @@ func (h *Handler) blackhole(ctx context.Context, input buf.Reader, output buf.Wr
|
||||
case <-time.After(delay):
|
||||
common.Interrupt(input)
|
||||
common.Interrupt(output)
|
||||
errors.LogInfo(ctx, "closed blackholed connection to blocked target: ", *dest)
|
||||
rule.log(ctx, "closed blackholed connection to blocked target: ", *dest)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
+1
-1
@@ -223,7 +223,7 @@ To make it start, wintun.dll specific for your Windows/arch must be present next
|
||||
|
||||
After the start network adapter with the name you chose in the config will be created in the system, and exist while Xray is running. A Wintun adapter that already has that name is reused.
|
||||
|
||||
With IPv6 disabled system-wide (the `DisabledComponents` registry value), the adapter has no IPv6, and the TUN starts with IPv4 only, unless `gateway` or `autoSystemRoutingTable` has IPv6 entries, which then keep it from starting.
|
||||
With IPv6 disabled system-wide (the `DisabledComponents` registry value), the adapter has no IPv6, and the TUN starts with IPv4 only, unless `gateway`, `dns` or `autoSystemRoutingTable` has IPv6 entries, which then keep it from starting.
|
||||
|
||||
When `dns` is set, those servers are applied to the adapter. Windows is kept from registering the TUN's addresses in DNS, and its DNS cache is flushed when the TUN starts and stops. Xray's own lookups through the system's DNS servers, like those of the `localhost` DNS server with `autoOutboundsInterface`, skip them, unless another interface uses them as well, as they would lead back into the TUN.
|
||||
|
||||
|
||||
@@ -115,6 +115,7 @@ func (t *WindowsTun) Start() (err error) {
|
||||
for _, ip := range t.options.DNS {
|
||||
dns = append(dns, netip.MustParseAddr(ip))
|
||||
}
|
||||
dns6 := slices.ContainsFunc(dns, netip.Addr.Is6)
|
||||
|
||||
var route4, route6 bool
|
||||
routesMap := make(map[winipcfg.RouteData]struct{})
|
||||
@@ -175,7 +176,7 @@ startOver:
|
||||
if err != nil {
|
||||
// With IPv6 disabled system-wide (DisabledComponents), the adapter has no
|
||||
// IPv6 interface at all. Skip the family unless the config asks for it.
|
||||
if err == windows.ERROR_NOT_FOUND && family == windows.AF_INET6 && !address6 && !route6 {
|
||||
if err == windows.ERROR_NOT_FOUND && family == windows.AF_INET6 && !address6 && !route6 && !dns6 {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
@@ -201,6 +202,12 @@ startOver:
|
||||
}
|
||||
err = t.luid.SetDNS(family, dns, nil)
|
||||
if err != nil {
|
||||
// Skip IPv6 the config does not ask for also when setting its DNS
|
||||
// servers fails, as reported on Windows 11 24H2 with IPv6 disabled
|
||||
// (Access is denied).
|
||||
if family == windows.AF_INET6 && !address6 && !route6 && !dns6 {
|
||||
continue
|
||||
}
|
||||
firstErr = errors.New("unable to set DNS").Base(err)
|
||||
if err == windows.ERROR_NOT_FOUND {
|
||||
goto startOver
|
||||
|
||||
@@ -17,7 +17,7 @@ import (
|
||||
|
||||
type bind struct {
|
||||
ctx context.Context
|
||||
resolveFunc func(host string) (net.IP, error)
|
||||
resolveFunc func(ctx context.Context, host string) (net.IP, error)
|
||||
listenFunc func(ctx context.Context) (net.PacketConn, error)
|
||||
downFunc func() error
|
||||
reserved []byte
|
||||
@@ -143,7 +143,11 @@ func (b *bind) ParseEndpoint(s string) (conn.Endpoint, error) {
|
||||
if port < 0 || port > 65535 {
|
||||
return nil, errors.New("invalid port " + sport)
|
||||
}
|
||||
ip, err := b.resolveFunc(host)
|
||||
// up sets the connection's context right before it hands over the endpoint
|
||||
b.mu.Lock()
|
||||
ctx := b.ctx
|
||||
b.mu.Unlock()
|
||||
ip, err := b.resolveFunc(ctx, host)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
@@ -323,12 +323,12 @@ func (h *Handler) up(ctx context.Context) error {
|
||||
return h.dev.Up()
|
||||
}
|
||||
|
||||
func MakeResolveLocal(d dns.Client, streamSettings *internet.MemoryStreamConfig) func(host string) (net.IP, error) {
|
||||
return func(host string) (net.IP, error) {
|
||||
func MakeResolveLocal(d dns.Client, streamSettings *internet.MemoryStreamConfig) func(ctx context.Context, host string) (net.IP, error) {
|
||||
return func(ctx context.Context, host string) (net.IP, error) {
|
||||
if ip := net.ParseIP(host); ip != nil {
|
||||
return ip, nil
|
||||
}
|
||||
ips, _, err := d.LookupIP(host, dns.IPOption{IPv4Enable: true, IPv6Enable: true})
|
||||
ips, err := lookupIPContext(ctx, d, host)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -368,6 +368,27 @@ func MakeResolveLocal(d dns.Client, streamSettings *internet.MemoryStreamConfig)
|
||||
}
|
||||
}
|
||||
|
||||
func lookupIPContext(ctx context.Context, d dns.Client, host string) ([]net.IP, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
type result struct {
|
||||
ips []net.IP
|
||||
err error
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
go func() {
|
||||
ips, _, err := d.LookupIP(host, dns.IPOption{IPv4Enable: true, IPv6Enable: true})
|
||||
done <- result{ips, err}
|
||||
}()
|
||||
select {
|
||||
case r := <-done:
|
||||
return r.ips, r.err
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
type UDPConnClient struct {
|
||||
net.PacketConn
|
||||
Dest *net.UDPAddr
|
||||
|
||||
@@ -661,7 +661,7 @@ func TestVlessRealityFingerprints(t *testing.T) {
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// A blocked connection must be closed after blockDelay even while the client keeps it open and sends nothing.
|
||||
// A blocked connection must close after finalRules.delay even while the client keeps it open and sends nothing.
|
||||
func TestVlessBlockDelay(t *testing.T) {
|
||||
tcpServer := tcp.Server{
|
||||
MsgProcessor: xor,
|
||||
@@ -671,7 +671,7 @@ func TestVlessBlockDelay(t *testing.T) {
|
||||
defer tcpServer.Close()
|
||||
|
||||
for _, delay := range []uint64{0, 1} {
|
||||
t.Run(fmt.Sprint("blockDelay=", delay), func(t *testing.T) {
|
||||
t.Run(fmt.Sprint("delay=", delay), func(t *testing.T) {
|
||||
userID := protocol.NewID(uuid.New())
|
||||
serverPort := tcp.PickPort()
|
||||
serverConfig := &core.Config{
|
||||
@@ -698,7 +698,7 @@ func TestVlessBlockDelay(t *testing.T) {
|
||||
FinalRules: []*freedom.FinalRuleConfig{
|
||||
{
|
||||
Action: freedom.RuleAction_Block,
|
||||
BlockDelay: &freedom.Range{Min: delay, Max: delay},
|
||||
Delay: &freedom.Range{Min: delay, Max: delay},
|
||||
},
|
||||
},
|
||||
}),
|
||||
@@ -726,15 +726,15 @@ func TestVlessBlockDelay(t *testing.T) {
|
||||
Port: dest.Port,
|
||||
}, &encoding.Addons{}))
|
||||
|
||||
blockDelay := time.Duration(delay) * time.Second
|
||||
common.Must(conn.SetReadDeadline(time.Now().Add(blockDelay + 5*time.Second)))
|
||||
blockDuration := time.Duration(delay) * time.Second
|
||||
common.Must(conn.SetReadDeadline(time.Now().Add(blockDuration + 5*time.Second)))
|
||||
_, err = io.Copy(io.Discard, conn)
|
||||
elapsed := time.Since(start)
|
||||
if err, ok := err.(net.Error); ok && err.Timeout() {
|
||||
t.Fatalf("connection is still open after %v", elapsed)
|
||||
}
|
||||
if elapsed < blockDelay {
|
||||
t.Fatalf("connection is closed after %v, before blockDelay", elapsed)
|
||||
if elapsed < blockDuration {
|
||||
t.Fatalf("connection is closed after %v, before finalRules.delay", elapsed)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
goerrors "errors"
|
||||
"io"
|
||||
mrand "math/rand"
|
||||
stdnet "net"
|
||||
"net/netip"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -164,7 +165,7 @@ func (c *udpHopConn) recv(conn net.PacketConn) {
|
||||
n, addr, err := conn.ReadFrom(p)
|
||||
if err != nil {
|
||||
pool.Put(p[:cap(p)])
|
||||
if c.closed() {
|
||||
if c.closed() || goerrors.Is(err, stdnet.ErrClosed) {
|
||||
return
|
||||
}
|
||||
var netErr net.Error
|
||||
|
||||
@@ -441,6 +441,8 @@ type sessionTransport struct {
|
||||
mu sync.Mutex
|
||||
client *splithttp.XmuxClient
|
||||
closed bool
|
||||
slowAt, retiredAt time.Time
|
||||
movedAt time.Time
|
||||
}
|
||||
|
||||
func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
@@ -450,7 +452,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
return nil, errSessionClosed
|
||||
}
|
||||
client := t.client
|
||||
if client == nil || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) {
|
||||
if client == nil || client.XmuxConn.IsClosed() || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) {
|
||||
t.pool.mu.Lock()
|
||||
t.client = t.pool.xmux.GetXmuxClient(r.Context())
|
||||
t.pool.mu.Unlock()
|
||||
@@ -458,6 +460,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
t.client.LeftRequests.Add(-1)
|
||||
if client != nil {
|
||||
client.DoneRunning()
|
||||
t.movedAt = time.Now()
|
||||
}
|
||||
client = t.client
|
||||
}
|
||||
@@ -471,6 +474,22 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
return conn.roundTrip(r)
|
||||
}
|
||||
|
||||
func (t *sessionTransport) stalled(began time.Time) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if began.Before(t.movedAt) {
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
slow := now.Sub(t.slowAt) < 3*time.Second
|
||||
t.slowAt = now
|
||||
if !slow || t.client == nil || now.Sub(t.retiredAt) < 2*time.Second {
|
||||
return
|
||||
}
|
||||
t.retiredAt = now
|
||||
t.client.XmuxConn.(*poolConn).Close()
|
||||
}
|
||||
|
||||
func (t *sessionTransport) Close() {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
|
||||
@@ -987,3 +987,103 @@ func TestDriveXmuxOverH3(t *testing.T) {
|
||||
}
|
||||
waitFor(t, "the first connection of XMUX to close", func() bool { return server.closed.Load() == 1 })
|
||||
}
|
||||
|
||||
type stallConn struct{}
|
||||
|
||||
func (stallConn) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
<-r.Context().Done()
|
||||
return nil, r.Context().Err()
|
||||
}
|
||||
|
||||
type okConn struct{}
|
||||
|
||||
func (okConn) RoundTrip(r *http.Request) (*http.Response, error) {
|
||||
return &http.Response{
|
||||
StatusCode: http.StatusOK,
|
||||
Header: make(http.Header),
|
||||
Body: io.NopCloser(strings.NewReader(`{"id":"file-1"}`)),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func TestSessionRetiresStalledConn(t *testing.T) {
|
||||
made := atomic.Int32{}
|
||||
transport := &sessionTransport{pool: newConnPool(nil, func(connCtx context.Context) *http.Client {
|
||||
if made.Add(1) == 1 {
|
||||
return &http.Client{Transport: stallConn{}}
|
||||
}
|
||||
return &http.Client{Transport: okConn{}}
|
||||
})}
|
||||
storage := &driveStorage{
|
||||
driveShared: &driveShared{
|
||||
token: "token",
|
||||
tokenExpiry: time.Now().Add(time.Hour),
|
||||
backoff: time.Millisecond,
|
||||
},
|
||||
client: &http.Client{Transport: transport},
|
||||
inflight: make(chan struct{}, 1),
|
||||
reserved: make(chan struct{}, 1),
|
||||
}
|
||||
|
||||
ctx := context.WithValue(context.Background(), attemptKey{}, 100*time.Millisecond)
|
||||
code, body, err := storage.do(ctx, http.MethodPost, "http://drive.invalid/files", "", nil)
|
||||
if err != nil {
|
||||
t.Fatalf("the first request over a stalled connection: %v", err)
|
||||
}
|
||||
if code != http.StatusOK || string(body) != `{"id":"file-1"}` {
|
||||
t.Fatalf("the stalled connection gave %d %q", code, body)
|
||||
}
|
||||
if n := made.Load(); n != 2 {
|
||||
t.Fatalf("a stalled connection took %d connections, want it retired and one made after it", n)
|
||||
}
|
||||
if _, _, err := storage.do(ctx, http.MethodPost, "http://drive.invalid/files", "", nil); err != nil {
|
||||
t.Fatalf("the request after the retirement: %v", err)
|
||||
}
|
||||
if n := made.Load(); n != 2 {
|
||||
t.Fatalf("%d connections were made, want the healthy one kept", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSessionIgnoresStallOfRetiredConn(t *testing.T) {
|
||||
var made atomic.Int32
|
||||
transport := &sessionTransport{pool: newConnPool(nil, func(connCtx context.Context) *http.Client {
|
||||
made.Add(1)
|
||||
return &http.Client{Transport: okConn{}}
|
||||
})}
|
||||
send := func() {
|
||||
req, err := http.NewRequest(http.MethodGet, "http://drive.invalid/files", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resp, err := transport.RoundTrip(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resp.Body.Close()
|
||||
}
|
||||
send()
|
||||
old := time.Now()
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
transport.stalled(old)
|
||||
transport.stalled(old)
|
||||
send()
|
||||
if n := made.Load(); n != 2 {
|
||||
t.Fatalf("a stalled connection took %d connections, want it retired and one made after it", n)
|
||||
}
|
||||
transport.mu.Lock()
|
||||
transport.slowAt, transport.retiredAt = time.Time{}, time.Time{}
|
||||
transport.mu.Unlock()
|
||||
transport.stalled(old)
|
||||
transport.stalled(old)
|
||||
send()
|
||||
if n := made.Load(); n != 2 {
|
||||
t.Fatalf("the stall of a retired connection took the healthy one too, %d connections", n)
|
||||
}
|
||||
time.Sleep(2 * time.Millisecond)
|
||||
fresh := time.Now()
|
||||
transport.stalled(fresh)
|
||||
transport.stalled(fresh)
|
||||
send()
|
||||
if n := made.Load(); n != 3 {
|
||||
t.Fatalf("a stall of the current connection made %d connections, want it retired", n)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24,7 +24,8 @@ const (
|
||||
driveBoundary = "xdrive-boundary"
|
||||
drivePageSize = 1000
|
||||
driveMaxAttempts = 8
|
||||
driveMaxInflight = 32
|
||||
driveMaxInflight = 64
|
||||
driveReservedInflight = 8
|
||||
driveInlineLimit = 12000
|
||||
// driveTimeout ends a request that makes no progress for that long
|
||||
driveTimeout = 60 * time.Second
|
||||
@@ -36,13 +37,21 @@ var (
|
||||
driveFilesURL = "https://www.googleapis.com/drive/v3/files"
|
||||
driveUploadURL = "https://www.googleapis.com/upload/drive/v3/files?uploadType=multipart&fields=id,name"
|
||||
driveInitialBackoff = 200 * time.Millisecond
|
||||
driveCooldown = 2 * time.Second
|
||||
driveSlotRead = 3 * time.Second
|
||||
driveSlotWrite = 5 * time.Second
|
||||
)
|
||||
|
||||
type laneKey struct{}
|
||||
|
||||
type attemptKey struct{}
|
||||
|
||||
type driveStorage struct {
|
||||
*driveShared
|
||||
// a session has a client and inflight of its own, see session
|
||||
client *http.Client
|
||||
inflight chan struct{}
|
||||
reserved chan struct{}
|
||||
release func()
|
||||
}
|
||||
|
||||
@@ -65,6 +74,9 @@ type driveShared struct {
|
||||
|
||||
idMu sync.Mutex
|
||||
ids map[string]string
|
||||
|
||||
coolMu sync.Mutex
|
||||
coolUntil time.Time
|
||||
}
|
||||
|
||||
func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config) (*driveStorage, error) {
|
||||
@@ -100,6 +112,7 @@ func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config
|
||||
},
|
||||
client: client,
|
||||
inflight: make(chan struct{}, driveMaxInflight),
|
||||
reserved: make(chan struct{}, driveReservedInflight),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -111,6 +124,7 @@ func (s *driveStorage) session() Storage {
|
||||
driveShared: s.driveShared,
|
||||
client: &http.Client{Transport: transport},
|
||||
inflight: make(chan struct{}, driveMaxInflight),
|
||||
reserved: make(chan struct{}, driveReservedInflight),
|
||||
release: transport.Close,
|
||||
}
|
||||
}
|
||||
@@ -219,6 +233,17 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
backoff := s.backoff
|
||||
var lastErr error
|
||||
|
||||
s.coolMu.Lock()
|
||||
until := s.coolUntil
|
||||
s.coolMu.Unlock()
|
||||
if d := time.Until(until); d > 0 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return 0, nil, ctx.Err()
|
||||
case <-time.After(d):
|
||||
}
|
||||
}
|
||||
|
||||
for attempt := 0; attempt < driveMaxAttempts; attempt++ {
|
||||
if attempt > 0 {
|
||||
select {
|
||||
@@ -238,8 +263,12 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
continue
|
||||
}
|
||||
|
||||
pool := s.inflight
|
||||
if ctx.Value(laneKey{}) != nil {
|
||||
pool = s.reserved
|
||||
}
|
||||
select {
|
||||
case s.inflight <- struct{}{}:
|
||||
case pool <- struct{}{}:
|
||||
case <-ctx.Done():
|
||||
return 0, nil, ctx.Err()
|
||||
}
|
||||
@@ -248,8 +277,14 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
if body != nil {
|
||||
reader = bytes.NewReader(body)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, target, reader)
|
||||
attemptCtx, cancel := ctx, context.CancelFunc(func() {})
|
||||
if d, ok := ctx.Value(attemptKey{}).(time.Duration); ok {
|
||||
attemptCtx, cancel = context.WithTimeout(ctx, d)
|
||||
}
|
||||
req, err := http.NewRequestWithContext(attemptCtx, method, target, reader)
|
||||
if err != nil {
|
||||
cancel()
|
||||
<-pool
|
||||
return 0, nil, errors.New("failed to build a Drive request").Base(err)
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
@@ -257,9 +292,17 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
req.Header.Set("Content-Type", contentType)
|
||||
}
|
||||
|
||||
began := time.Now()
|
||||
resp, err := s.client.Do(req)
|
||||
<-s.inflight
|
||||
<-pool
|
||||
if err != nil {
|
||||
timedOut := attemptCtx.Err() == context.DeadlineExceeded && ctx.Err() == nil
|
||||
cancel()
|
||||
if timedOut {
|
||||
if t, ok := s.client.Transport.(*sessionTransport); ok {
|
||||
t.stalled(began)
|
||||
}
|
||||
}
|
||||
if ctx.Err() != nil {
|
||||
return 0, nil, ctx.Err()
|
||||
}
|
||||
@@ -270,6 +313,7 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
}
|
||||
payload, err := io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
cancel()
|
||||
if err != nil {
|
||||
if ctx.Err() != nil {
|
||||
return 0, nil, ctx.Err()
|
||||
@@ -286,13 +330,19 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
if resp.StatusCode == http.StatusForbidden && rateLimited(payload) {
|
||||
lastErr = errors.New("Drive is rate limiting: ", string(payload))
|
||||
errors.LogWarning(ctx, "rate limited by Drive, attempt ",
|
||||
attempt+1, " of ", driveMaxAttempts)
|
||||
attempt+1, " of ", driveMaxAttempts, ", ", method, " ", len(body),
|
||||
" bytes, body read ", strings.Contains(target, "alt=media"),
|
||||
", lane ", ctx.Value(laneKey{}) != nil, ": ", string(payload[:min(len(payload), 300)]))
|
||||
s.cool()
|
||||
continue
|
||||
}
|
||||
if retryableStatus(resp.StatusCode) {
|
||||
lastErr = errors.New("Drive answered ", resp.StatusCode, ": ", string(payload))
|
||||
errors.LogWarning(ctx, "retrying after Drive answered ", resp.StatusCode,
|
||||
", attempt ", attempt+1, " of ", driveMaxAttempts)
|
||||
if resp.StatusCode == http.StatusTooManyRequests {
|
||||
s.cool()
|
||||
}
|
||||
continue
|
||||
}
|
||||
return resp.StatusCode, payload, nil
|
||||
@@ -301,6 +351,12 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
|
||||
return 0, nil, lastErr
|
||||
}
|
||||
|
||||
func (s *driveStorage) cool() {
|
||||
s.coolMu.Lock()
|
||||
s.coolUntil = time.Now().Add(jitter(driveCooldown))
|
||||
s.coolMu.Unlock()
|
||||
}
|
||||
|
||||
func (s *driveStorage) invalidateToken() {
|
||||
s.tokenMu.Lock()
|
||||
s.token = ""
|
||||
@@ -542,12 +598,27 @@ func (s *driveStorage) Delete(ctx context.Context, name string) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
sem := make(chan struct{}, 8)
|
||||
var wg sync.WaitGroup
|
||||
var mu sync.Mutex
|
||||
var first error
|
||||
for childName, file := range children {
|
||||
if err := s.deleteID(ctx, childName, file.id); err != nil {
|
||||
return err
|
||||
wg.Add(1)
|
||||
sem <- struct{}{}
|
||||
go func(name, id string) {
|
||||
defer wg.Done()
|
||||
defer func() { <-sem }()
|
||||
if err := s.deleteID(ctx, name, id); err != nil {
|
||||
mu.Lock()
|
||||
if first == nil {
|
||||
first = err
|
||||
}
|
||||
mu.Unlock()
|
||||
}
|
||||
return nil
|
||||
}(childName, file.id)
|
||||
}
|
||||
wg.Wait()
|
||||
return first
|
||||
}
|
||||
|
||||
func (s *driveStorage) List(ctx context.Context, prefix string) ([]Entry, error) {
|
||||
@@ -604,6 +675,9 @@ func (s *driveStorage) makeSlot(ctx context.Context, name string) (string, error
|
||||
|
||||
// setSlot implements slotStore, a description and a body go in one request
|
||||
func (s *driveStorage) setSlot(ctx context.Context, id, description string, body []byte) error {
|
||||
if body == nil {
|
||||
ctx = context.WithValue(ctx, attemptKey{}, driveSlotWrite)
|
||||
}
|
||||
metadata, err := json.Marshal(map[string]string{"description": description})
|
||||
if err != nil {
|
||||
return errors.New("failed to build the slot metadata").Base(err)
|
||||
@@ -631,6 +705,7 @@ func (s *driveStorage) setSlot(ctx context.Context, id, description string, body
|
||||
|
||||
// readDescription implements slotStore
|
||||
func (s *driveStorage) readDescription(ctx context.Context, id string) (string, error) {
|
||||
ctx = context.WithValue(ctx, attemptKey{}, driveSlotRead)
|
||||
status, payload, err := s.do(ctx, http.MethodGet,
|
||||
s.filesURL+"/"+url.PathEscape(id)+"?fields=description&supportsAllDrives=true", "", nil)
|
||||
switch {
|
||||
|
||||
@@ -7,8 +7,13 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptrace"
|
||||
"net/url"
|
||||
"os"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -343,3 +348,130 @@ func TestLiveDriveListLag(t *testing.T) {
|
||||
}
|
||||
t.Logf("worst listing lag: %v", worst.Round(time.Millisecond))
|
||||
}
|
||||
|
||||
func TestLiveDriveLatencyFloor(t *testing.T) {
|
||||
config := liveDriveConfig(t)
|
||||
root, err := newDriveStorage(&internet.MemoryStreamConfig{}, config)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer root.Close()
|
||||
store := root.session().(*driveStorage)
|
||||
defer store.Close()
|
||||
|
||||
ctx := context.Background()
|
||||
const probe = "latency-floor/probe"
|
||||
id, err := store.makeSlot(ctx, probe)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer store.deleteID(ctx, flatten(probe), id)
|
||||
|
||||
small := newSlotFrame(1, slotInline, randomPayload(700)).encode()
|
||||
large := newSlotFrame(1, slotInline, randomPayload(slotInlineLimit)).encode()
|
||||
file := store.filesURL + "/" + url.PathEscape(id)
|
||||
|
||||
type sample struct {
|
||||
total, ttfb time.Duration
|
||||
reused bool
|
||||
}
|
||||
measure := func(label string, op func(context.Context) error) []sample {
|
||||
var out []sample
|
||||
for i := 0; i < 28; i++ {
|
||||
var wrote, first time.Time
|
||||
var reused bool
|
||||
trace := &httptrace.ClientTrace{
|
||||
GotConn: func(info httptrace.GotConnInfo) { reused = info.Reused },
|
||||
WroteRequest: func(httptrace.WroteRequestInfo) { wrote = time.Now() },
|
||||
GotFirstResponseByte: func() { first = time.Now() },
|
||||
}
|
||||
start := time.Now()
|
||||
err := op(httptrace.WithClientTrace(ctx, trace))
|
||||
total := time.Since(start)
|
||||
if err != nil {
|
||||
t.Logf("latency-floor: %s sample failed: %v", label, err)
|
||||
} else if i >= 3 {
|
||||
var ttfb time.Duration
|
||||
if !wrote.IsZero() && !first.IsZero() {
|
||||
ttfb = first.Sub(wrote)
|
||||
}
|
||||
out = append(out, sample{total, ttfb, reused})
|
||||
}
|
||||
time.Sleep(250 * time.Millisecond)
|
||||
}
|
||||
return out
|
||||
}
|
||||
report := func(label string, s []sample) time.Duration {
|
||||
if len(s) == 0 {
|
||||
t.Logf("latency-floor: %-30s no samples", label)
|
||||
return 0
|
||||
}
|
||||
totals := make([]time.Duration, len(s))
|
||||
ttfbs := make([]time.Duration, len(s))
|
||||
reused := 0
|
||||
for i, x := range s {
|
||||
totals[i], ttfbs[i] = x.total, x.ttfb
|
||||
if x.reused {
|
||||
reused++
|
||||
}
|
||||
}
|
||||
sort.Slice(totals, func(a, b int) bool { return totals[a] < totals[b] })
|
||||
sort.Slice(ttfbs, func(a, b int) bool { return ttfbs[a] < ttfbs[b] })
|
||||
q := func(v []time.Duration, p float64) time.Duration { return v[min(int(p*float64(len(v))), len(v)-1)] }
|
||||
t.Logf("latency-floor: %-30s n=%d min=%v p50=%v p95=%v ttfb_p50=%v reused=%d/%d",
|
||||
label, len(s), totals[0], q(totals, 0.5), q(totals, 0.95), q(ttfbs, 0.5), reused, len(s))
|
||||
return q(ttfbs, 0.5)
|
||||
}
|
||||
patch := func(target string, doc map[string]any) func(context.Context) error {
|
||||
body, _ := json.Marshal(doc)
|
||||
return func(ctx context.Context) error {
|
||||
status, reply, err := store.do(ctx, http.MethodPatch, target, "application/json; charset=UTF-8", body)
|
||||
if err == nil && status != http.StatusOK {
|
||||
err = fmt.Errorf("status %d: %s", status, reply)
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
get := func(target string) func(context.Context) error {
|
||||
return func(ctx context.Context) error {
|
||||
status, reply, err := store.do(ctx, http.MethodGet, target, "", nil)
|
||||
if err == nil && status != http.StatusOK {
|
||||
err = fmt.Errorf("status %d: %s", status, reply)
|
||||
}
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
proto := ""
|
||||
netFloor := report("net floor (401, no auth)", measure("net", func(ctx context.Context) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://www.googleapis.com/drive/v3/about?fields=user", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := store.client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
proto = resp.Proto
|
||||
io.Copy(io.Discard, resp.Body)
|
||||
return resp.Body.Close()
|
||||
}))
|
||||
t.Logf("latency-floor: negotiated protocol %s", proto)
|
||||
readDesc := report("GET description (prod)", measure("get", func(ctx context.Context) error {
|
||||
_, err := store.readDescription(ctx, id)
|
||||
return err
|
||||
}))
|
||||
report("GET fields=id", measure("getid", get(file+"?fields=id&supportsAllDrives=true")))
|
||||
report("GET description, no allDrives", measure("getnoall", get(file+"?fields=description")))
|
||||
write1k := report("PATCH inline 1 KB (prod)", measure("p1k", func(ctx context.Context) error {
|
||||
return store.setSlot(ctx, id, small, nil)
|
||||
}))
|
||||
write14k := report("PATCH inline 14 KB (prod)", measure("p14k", func(ctx context.Context) error {
|
||||
return store.setSlot(ctx, id, large, nil)
|
||||
}))
|
||||
report("PATCH 1 KB, no allDrives", measure("pnoall", patch(file+"?fields=id", map[string]any{"description": small})))
|
||||
report("PATCH name 200 B", measure("pname", patch(file+"?fields=id&supportsAllDrives=true", map[string]any{"name": strings.Repeat("n", 200)})))
|
||||
report("PATCH appProperties 100 B", measure("pprop", patch(file+"?fields=id&supportsAllDrives=true", map[string]any{"appProperties": map[string]string{"k": strings.Repeat("a", 100)}})))
|
||||
t.Logf("latency-floor: derived: write backend ~ %v, read backend ~ %v, 14 KB payload cost ~ %v",
|
||||
write1k-netFloor, readDesc-netFloor, write14k-write1k)
|
||||
}
|
||||
|
||||
@@ -13,7 +13,7 @@ const (
|
||||
defaultConcurrency = 8
|
||||
|
||||
maxSegmentBytes = 16 * 1024 * 1024
|
||||
maxConcurrency = 64
|
||||
maxConcurrency = 256
|
||||
)
|
||||
|
||||
type params struct {
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"io"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/xtls/xray-core/common/errors"
|
||||
@@ -27,7 +28,6 @@ const (
|
||||
// slotInlineLimit is how large a segment is at most to go in the description
|
||||
// of its slot, which takes one request less than a body
|
||||
slotInlineLimit = 14 << 10
|
||||
slotDepthLimit = 8
|
||||
// inlineRampSegments segments of one size in a row double it, from
|
||||
// slotInlineLimit up to segmentBytes
|
||||
inlineRampSegments = 8
|
||||
@@ -41,19 +41,26 @@ const (
|
||||
var (
|
||||
// the head is asked for between slotHuntFloor and slotHuntCeiling, a segment
|
||||
// behind it at about the pace that segments come
|
||||
slotHuntFloor = 50 * time.Millisecond
|
||||
slotHuntCeiling = 200 * time.Millisecond
|
||||
slotHuntFloor = 10 * time.Millisecond
|
||||
slotHuntCeiling = 25 * time.Millisecond
|
||||
slotTailMin = 150 * time.Millisecond
|
||||
slotTailMax = 800 * time.Millisecond
|
||||
slotFetchAhead = 6
|
||||
slotFetchAhead = 12
|
||||
|
||||
slotAckRetry = 100 * time.Millisecond
|
||||
slotWatermarkTTL = 250 * time.Millisecond
|
||||
slotAckRetry = 30 * time.Millisecond
|
||||
slotWatermarkTTL = 100 * time.Millisecond
|
||||
slotStallTimeout = 60 * time.Second
|
||||
slotCloseGrace = 10 * time.Second
|
||||
slotGoneTimeout = 60 * time.Second
|
||||
slotBodyHedge = 2 * time.Second
|
||||
|
||||
slotReadAhead int64 = 3
|
||||
slotReadAheadPoll = 40 * time.Millisecond
|
||||
slotReadAheadBurst = 500 * time.Millisecond
|
||||
)
|
||||
|
||||
var slotWriteInflight = 12
|
||||
|
||||
// slotStore is a storage of objects that are written and read again by id,
|
||||
// each with a description and a body
|
||||
type slotStore interface {
|
||||
@@ -143,7 +150,7 @@ func decodeRing(b []byte) (ring, bool) {
|
||||
|
||||
// makeRing makes the slots and watermarks of a session, all at once
|
||||
func makeRing(ctx context.Context, store slotStore, session string, p params) (ring, error) {
|
||||
depth := min(max(p.concurrency, 2), slotDepthLimit)
|
||||
depth := max(p.concurrency, 2)
|
||||
r := ring{
|
||||
Up: slotTable{Slots: make([]string, depth)},
|
||||
Down: slotTable{Slots: make([]string, depth)},
|
||||
@@ -177,7 +184,7 @@ func makeRing(ctx context.Context, store slotStore, session string, p params) (r
|
||||
}
|
||||
|
||||
func readWatermark(ctx context.Context, store slotStore, id string) (int64, error) {
|
||||
d, err := store.readDescription(ctx, id)
|
||||
d, err := store.readDescription(context.WithValue(ctx, laneKey{}, true), id)
|
||||
if err != nil || d == "" {
|
||||
return -1, err
|
||||
}
|
||||
@@ -216,7 +223,7 @@ func newSlotWriter(ctx context.Context, store slotStore, table slotTable, p para
|
||||
store: store,
|
||||
table: table,
|
||||
params: p,
|
||||
slots: make(chan struct{}, len(table.Slots)),
|
||||
slots: make(chan struct{}, min(len(table.Slots), slotWriteInflight)),
|
||||
size: min(slotInlineLimit, p.segmentBytes),
|
||||
acked: -1,
|
||||
progress: time.Now(),
|
||||
@@ -271,6 +278,10 @@ func (w *slotWriter) readyToFlush() bool {
|
||||
grew := len(w.buf) > w.lastSize
|
||||
w.lastSize = len(w.buf)
|
||||
|
||||
if len(w.slots) > 0 && w.held < maxCoalescedTicks && len(w.buf) < w.size/2 {
|
||||
w.held++
|
||||
return false
|
||||
}
|
||||
if grew && w.held < maxCoalescedTicks {
|
||||
w.held++
|
||||
return false
|
||||
@@ -296,11 +307,19 @@ func (w *slotWriter) flushLocked() error {
|
||||
return w.err
|
||||
}
|
||||
|
||||
if w.size < w.segmentBytes && len(w.buf) >= w.size && len(w.slots)*4 > cap(w.slots)*3 {
|
||||
w.size = w.segmentBytes
|
||||
w.full = 0
|
||||
}
|
||||
|
||||
n := min(len(w.buf), w.size)
|
||||
if len(w.buf) > 4*slotInlineLimit {
|
||||
// a burst goes at the full size at once
|
||||
n = min(len(w.buf), w.segmentBytes)
|
||||
}
|
||||
if n > slotInlineLimit && len(w.buf) <= 2*cap(w.slots)*slotInlineLimit {
|
||||
n = slotInlineLimit
|
||||
}
|
||||
if n == w.size && w.size < w.segmentBytes {
|
||||
if w.full++; w.full == inlineRampSegments {
|
||||
w.full = 0
|
||||
@@ -332,15 +351,21 @@ func (w *slotWriter) upload(seq int64, chunk []byte) {
|
||||
if len(chunk) <= slotInlineLimit {
|
||||
mode = slotInline
|
||||
}
|
||||
began := time.Now()
|
||||
if err := w.await(seq); err != nil {
|
||||
w.fail(errors.New("failed to store segment ", seq).Base(err))
|
||||
return
|
||||
}
|
||||
awaited := time.Since(began)
|
||||
putAt := time.Now()
|
||||
if err := w.write(seq, mode, chunk); err != nil {
|
||||
w.fail(errors.New("failed to store segment ", seq).Base(err))
|
||||
// the slot is free, the reader learns there what happened
|
||||
w.write(seq, slotErr, nil)
|
||||
return
|
||||
}
|
||||
errors.LogDebug(w.ctx, "xdrive write ", slotTag(w.table.Watermark), " seq=", seq, " bytes=", len(chunk),
|
||||
" mode=", mode, " await=", awaited, " put=", time.Since(putAt))
|
||||
}
|
||||
|
||||
func (w *slotWriter) write(seq int64, mode byte, data []byte) error {
|
||||
@@ -393,7 +418,7 @@ func (w *slotWriter) await(seq int64) error {
|
||||
w.ackMu.Lock()
|
||||
still := time.Since(w.progress)
|
||||
w.ackMu.Unlock()
|
||||
if time.Since(start) > slotStallTimeout || closed && still > slotCloseGrace {
|
||||
if min(time.Since(start), still) > slotStallTimeout || closed && still > slotCloseGrace {
|
||||
return errors.New("the reader took nothing from slot ", w.slotID(seq), " for ", still).Base(err)
|
||||
}
|
||||
|
||||
@@ -461,6 +486,7 @@ type slotReader struct {
|
||||
ch chan []byte
|
||||
wake chan struct{}
|
||||
acks chan struct{}
|
||||
grow chan struct{}
|
||||
|
||||
mu sync.Mutex
|
||||
head int64
|
||||
@@ -472,7 +498,11 @@ type slotReader struct {
|
||||
gap time.Duration
|
||||
activeAt time.Time
|
||||
active chan struct{}
|
||||
frontier int64
|
||||
err error
|
||||
reads atomic.Int64
|
||||
tailMin time.Duration
|
||||
tailMax time.Duration
|
||||
}
|
||||
|
||||
func newSlotReader(ctx context.Context, store slotStore, table slotTable, p params) *slotReader {
|
||||
@@ -484,9 +514,12 @@ func newSlotReader(ctx context.Context, store slotStore, table slotTable, p para
|
||||
ch: make(chan []byte, p.concurrency),
|
||||
wake: make(chan struct{}, 1),
|
||||
acks: make(chan struct{}, 1),
|
||||
grow: make(chan struct{}, 1),
|
||||
hitAt: time.Now(),
|
||||
activeAt: time.Now(),
|
||||
active: make(chan struct{}),
|
||||
tailMin: slotTailMin,
|
||||
tailMax: slotTailMax,
|
||||
}
|
||||
go r.run()
|
||||
go r.ackLoop()
|
||||
@@ -516,9 +549,20 @@ type slotFetch struct {
|
||||
ready chan struct{}
|
||||
// promoted is closed when the segment becomes the head
|
||||
promoted chan struct{}
|
||||
near chan struct{}
|
||||
body []byte
|
||||
end bool
|
||||
err error
|
||||
born time.Time
|
||||
readyAt time.Time
|
||||
}
|
||||
|
||||
func (f *slotFetch) enter() {
|
||||
select {
|
||||
case <-f.near:
|
||||
default:
|
||||
close(f.near)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *slotReader) run() {
|
||||
@@ -529,17 +573,20 @@ func (r *slotReader) run() {
|
||||
fetches := make(map[int64]*slotFetch)
|
||||
for {
|
||||
r.mu.Lock()
|
||||
head, started := r.head, r.started
|
||||
head, started, frontier := r.head, r.started, r.frontier
|
||||
r.mu.Unlock()
|
||||
|
||||
window := 1
|
||||
if started {
|
||||
window = min(slotFetchAhead, len(r.table.Slots))
|
||||
window = min(max(int(frontier-head)+slotFetchAhead, slotFetchAhead), len(r.table.Slots))
|
||||
}
|
||||
for i := range window {
|
||||
seq := head + int64(i)
|
||||
if fetches[seq] == nil {
|
||||
fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{})}
|
||||
fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{}), near: make(chan struct{}), born: time.Now()}
|
||||
if int64(i) <= slotReadAhead {
|
||||
fetches[seq].enter()
|
||||
}
|
||||
go r.fetch(ctx, seq, i, fetches[seq])
|
||||
}
|
||||
}
|
||||
@@ -547,6 +594,8 @@ func (r *slotReader) run() {
|
||||
f := fetches[head]
|
||||
select {
|
||||
case <-f.ready:
|
||||
case <-r.grow:
|
||||
continue
|
||||
case <-ctx.Done():
|
||||
return
|
||||
}
|
||||
@@ -564,8 +613,14 @@ func (r *slotReader) run() {
|
||||
r.head++
|
||||
r.started = true
|
||||
r.mu.Unlock()
|
||||
errors.LogDebug(r.ctx, "xdrive read ", slotTag(r.table.Watermark), " seq=", head, " fetch=",
|
||||
f.readyAt.Sub(f.born), " blocked=", time.Since(f.readyAt), " bytes=", len(f.body),
|
||||
" reads=", r.reads.Load())
|
||||
if next := fetches[head+1]; next != nil {
|
||||
close(next.promoted)
|
||||
if nf := fetches[head+1+slotReadAhead]; nf != nil {
|
||||
nf.enter()
|
||||
}
|
||||
}
|
||||
select {
|
||||
case r.acks <- struct{}{}:
|
||||
@@ -586,11 +641,35 @@ func (r *slotReader) isHead(seq int64) bool {
|
||||
return r.head == seq
|
||||
}
|
||||
|
||||
func (r *slotReader) bursting() bool {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
return time.Since(r.hitAt) <= slotReadAheadBurst
|
||||
}
|
||||
|
||||
func (r *slotReader) reach(seq int64) {
|
||||
r.mu.Lock()
|
||||
grown := seq > r.frontier
|
||||
if grown {
|
||||
r.frontier = seq
|
||||
}
|
||||
r.mu.Unlock()
|
||||
if grown {
|
||||
select {
|
||||
case r.grow <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// fetch asks for the slot of seq until it holds seq: the head often, a segment
|
||||
// behind it at the pace that segments come, as it would wait for the head anyway
|
||||
// behind it soon after it was asked for, then at the pace that segments come
|
||||
func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFetch) {
|
||||
defer close(f.ready)
|
||||
if offset > 0 && !r.hold(ctx, time.Duration(offset)*r.pace(), f.promoted, nil) {
|
||||
defer func() {
|
||||
f.readyAt = time.Now()
|
||||
close(f.ready)
|
||||
}()
|
||||
if offset > 0 && !r.hold(ctx, r.tailMin, f.promoted, f.near) {
|
||||
return
|
||||
}
|
||||
|
||||
@@ -598,7 +677,17 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
|
||||
backoff := slotHuntFloor
|
||||
var goneSince, warnedAt time.Time
|
||||
for {
|
||||
description, err := r.store.readDescription(ctx, id)
|
||||
var description string
|
||||
var err error
|
||||
r.mu.Lock()
|
||||
hot := time.Since(r.activeAt) <= r.eagerWindow
|
||||
r.mu.Unlock()
|
||||
if hot && r.isHead(seq) {
|
||||
description, err = r.readHead(ctx, id, seq)
|
||||
} else {
|
||||
r.reads.Add(1)
|
||||
description, err = r.store.readDescription(ctx, id)
|
||||
}
|
||||
frame, ok := decodeSlot(description)
|
||||
switch {
|
||||
case err == nil && ok && frame.seq == seq:
|
||||
@@ -606,6 +695,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
|
||||
r.mu.Lock()
|
||||
r.touch()
|
||||
r.mu.Unlock()
|
||||
r.reach(seq)
|
||||
switch frame.mode {
|
||||
case slotEnd:
|
||||
f.end = true
|
||||
@@ -618,7 +708,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
|
||||
r.hit()
|
||||
return
|
||||
case slotMedia:
|
||||
if body, err := r.store.readBody(ctx, id); err == nil && frame.holds(body) {
|
||||
if body, err := r.readBodyHedged(ctx, id); err == nil && frame.holds(body) {
|
||||
f.body = body
|
||||
r.hit()
|
||||
return
|
||||
@@ -654,7 +744,15 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
|
||||
// only the head is asked for until something comes
|
||||
wait, a, b, backoff = 0, f.promoted, active, slotHuntFloor
|
||||
default:
|
||||
wait, a, backoff = min(max(r.pace(), slotTailMin), slotTailMax), f.promoted, slotHuntFloor
|
||||
wait, a, b, backoff = min(max(r.pace(), r.tailMin), r.tailMax), f.promoted, f.near, slotHuntFloor
|
||||
select {
|
||||
case <-f.near:
|
||||
b = nil
|
||||
if r.bursting() {
|
||||
wait = slotReadAheadPoll
|
||||
}
|
||||
default:
|
||||
}
|
||||
}
|
||||
if !r.hold(ctx, wait, a, b) {
|
||||
return
|
||||
@@ -662,6 +760,106 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
|
||||
}
|
||||
}
|
||||
|
||||
func (r *slotReader) readHead(ctx context.Context, id string, seq int64) (string, error) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
ctx = context.WithValue(ctx, laneKey{}, true)
|
||||
|
||||
type result struct {
|
||||
description string
|
||||
err error
|
||||
}
|
||||
results := make(chan result, 4)
|
||||
outstanding := 0
|
||||
launch := func() {
|
||||
outstanding++
|
||||
go func() {
|
||||
r.reads.Add(1)
|
||||
description, err := r.store.readDescription(ctx, id)
|
||||
results <- result{description, err}
|
||||
}()
|
||||
}
|
||||
launch()
|
||||
ticker := time.NewTicker(slotHuntCeiling)
|
||||
defer ticker.Stop()
|
||||
var last result
|
||||
seen := false
|
||||
for {
|
||||
select {
|
||||
case res := <-results:
|
||||
outstanding--
|
||||
if res.err != nil {
|
||||
return res.description, res.err
|
||||
}
|
||||
if frame, ok := decodeSlot(res.description); ok && frame.seq == seq {
|
||||
return res.description, nil
|
||||
}
|
||||
last, seen = res, true
|
||||
case <-ticker.C:
|
||||
r.mu.Lock()
|
||||
idle := time.Since(r.activeAt) > r.eagerWindow
|
||||
r.mu.Unlock()
|
||||
switch {
|
||||
case idle && seen:
|
||||
return last.description, nil
|
||||
case !idle && outstanding < 3:
|
||||
launch()
|
||||
}
|
||||
case <-r.wake:
|
||||
if outstanding < 3 {
|
||||
launch()
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return "", ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (r *slotReader) readBodyHedged(ctx context.Context, id string) ([]byte, error) {
|
||||
if slotBodyHedge <= 0 {
|
||||
return r.store.readBody(ctx, id)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
type result struct {
|
||||
body []byte
|
||||
err error
|
||||
}
|
||||
results := make(chan result, 2)
|
||||
launch := func() {
|
||||
go func() {
|
||||
body, err := r.store.readBody(ctx, id)
|
||||
results <- result{body, err}
|
||||
}()
|
||||
}
|
||||
launch()
|
||||
outstanding := 1
|
||||
timer := time.NewTimer(slotBodyHedge)
|
||||
defer timer.Stop()
|
||||
var last result
|
||||
for {
|
||||
select {
|
||||
case res := <-results:
|
||||
outstanding--
|
||||
if res.err == nil {
|
||||
return res.body, nil
|
||||
}
|
||||
last = res
|
||||
if outstanding == 0 {
|
||||
return nil, last.err
|
||||
}
|
||||
case <-timer.C:
|
||||
if outstanding == 1 {
|
||||
launch()
|
||||
outstanding++
|
||||
}
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// pace is about how long segments take to come one after the other
|
||||
func (r *slotReader) pace() time.Duration {
|
||||
r.mu.Lock()
|
||||
@@ -726,7 +924,7 @@ func (r *slotReader) ackLoop() {
|
||||
if want <= acked {
|
||||
break
|
||||
}
|
||||
if err := r.store.setSlot(r.ctx, r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil {
|
||||
if err := r.store.setSlot(context.WithValue(r.ctx, laneKey{}, true), r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil {
|
||||
acked, wait = want, slotAckRetry
|
||||
continue
|
||||
}
|
||||
@@ -756,3 +954,10 @@ func (r *slotReader) Err() error {
|
||||
}
|
||||
return io.EOF
|
||||
}
|
||||
|
||||
func slotTag(id string) string {
|
||||
if len(id) > 6 {
|
||||
return id[len(id)-6:]
|
||||
}
|
||||
return id
|
||||
}
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
"os"
|
||||
"strconv"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
@@ -304,6 +305,373 @@ func TestSlotStaleBody(t *testing.T) {
|
||||
client, server := slotPair(t, &staleSlots{memSlots: newMemSlots(), served: make(map[string]bool)}, 2, 64<<10)
|
||||
payload := randomPayload(200 << 10)
|
||||
errs := make(chan error, 1)
|
||||
writeAll(client, payload, true, errs)
|
||||
go func() {
|
||||
_, err := client.Write(payload)
|
||||
if err == nil {
|
||||
err = client.Close()
|
||||
}
|
||||
errs <- err
|
||||
}()
|
||||
readAll(t, server, payload)
|
||||
}
|
||||
|
||||
func TestSlotBurstSplitsInline(t *testing.T) {
|
||||
store := newMemSlots()
|
||||
client, server := slotPair(t, store, 8, 512<<10)
|
||||
payload := randomPayload(90 << 10)
|
||||
errs := make(chan error, 1)
|
||||
writeAll(client, payload, true, errs)
|
||||
readAll(t, server, payload)
|
||||
store.mu.Lock()
|
||||
bodies := len(store.bodies)
|
||||
store.mu.Unlock()
|
||||
if bodies != 0 {
|
||||
t.Fatalf("a burst of 90 KiB went in %d bodies, want none", bodies)
|
||||
}
|
||||
}
|
||||
|
||||
type stuckBodySlots struct {
|
||||
*memSlots
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (s *stuckBodySlots) readBody(ctx context.Context, id string) ([]byte, error) {
|
||||
if s.calls.Add(1) == 1 {
|
||||
<-ctx.Done()
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
return s.memSlots.readBody(ctx, id)
|
||||
}
|
||||
|
||||
func TestSlotBodyHedge(t *testing.T) {
|
||||
old := slotBodyHedge
|
||||
slotBodyHedge = 50 * time.Millisecond
|
||||
defer func() { slotBodyHedge = old }()
|
||||
|
||||
store := &stuckBodySlots{memSlots: newMemSlots()}
|
||||
client, server := slotPair(t, store, 2, 64<<10)
|
||||
payload := randomPayload(200 << 10)
|
||||
errs := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := client.Write(payload)
|
||||
if err == nil {
|
||||
err = client.Close()
|
||||
}
|
||||
errs <- err
|
||||
}()
|
||||
readAll(t, server, payload)
|
||||
if store.calls.Load() < 2 {
|
||||
t.Fatalf("the stuck body read was not hedged, %d reads", store.calls.Load())
|
||||
}
|
||||
}
|
||||
|
||||
func TestTakeSpareSkipsStale(t *testing.T) {
|
||||
cfg := &Config{}
|
||||
p := params{sessionTTL: 5 * time.Minute}
|
||||
st1, err := newLocalStorage(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
st2, err := newLocalStorage(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
box := spareBoxFor(cfg)
|
||||
box.ch <- spareRing{storage: st1, session: "old", at: time.Now().Add(-time.Hour)}
|
||||
box.ch <- spareRing{storage: st2, session: "new", at: time.Now()}
|
||||
got, ok := takeSpare(cfg, p)
|
||||
if !ok || got.session != "new" {
|
||||
t.Fatalf("takeSpare gave %q, %v, want the fresh one", got.session, ok)
|
||||
}
|
||||
if _, ok := takeSpare(cfg, p); ok {
|
||||
t.Fatal("takeSpare returned a second spare")
|
||||
}
|
||||
}
|
||||
|
||||
func TestDropSpareDeletesSession(t *testing.T) {
|
||||
st, err := newLocalStorage(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
name := sessionPrefix("gone") + "/" + uplinkDir + "/s0"
|
||||
if err := st.Put(context.Background(), name, []byte("x")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
dropSpare(spareRing{storage: st, session: "gone"})
|
||||
if got, _ := st.List(context.Background(), sessionPrefix("gone")); len(got) != 0 {
|
||||
t.Fatalf("%d entries left of a dropped spare", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
type stuckHeadSlots struct {
|
||||
*memSlots
|
||||
calls atomic.Int32
|
||||
}
|
||||
|
||||
func (s *stuckHeadSlots) readDescription(ctx context.Context, id string) (string, error) {
|
||||
if s.calls.Add(1) == 1 {
|
||||
<-ctx.Done()
|
||||
return "", ctx.Err()
|
||||
}
|
||||
return s.memSlots.readDescription(ctx, id)
|
||||
}
|
||||
|
||||
func TestSlotHeadReadOverlaps(t *testing.T) {
|
||||
client, server := slotPair(t, &stuckHeadSlots{memSlots: newMemSlots()}, 4, 64<<10)
|
||||
for _, c := range [][2]*Conn{{client, server}, {server, client}} {
|
||||
c[0].Write([]byte("ping"))
|
||||
if _, err := io.ReadFull(c[1], make([]byte, 4)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlotBurstReadsTogether(t *testing.T) {
|
||||
client, server := slotPair(t, newMemSlots(), 16, 512<<10)
|
||||
payload := randomPayload(8 * slotInlineLimit)
|
||||
go client.Write(payload)
|
||||
start := time.Now()
|
||||
got := make([]byte, len(payload))
|
||||
if _, err := io.ReadFull(server, got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatal("the burst came back changed")
|
||||
}
|
||||
if took := time.Since(start); took > time.Second {
|
||||
t.Fatalf("a burst of 8 segments took %v to read", took)
|
||||
}
|
||||
}
|
||||
|
||||
type lagSlots struct{ *memSlots }
|
||||
|
||||
func (s lagSlots) readDescription(ctx context.Context, id string) (string, error) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return "", ctx.Err()
|
||||
case <-time.After(200 * time.Millisecond):
|
||||
}
|
||||
return s.memSlots.readDescription(ctx, id)
|
||||
}
|
||||
|
||||
func (s lagSlots) setSlot(ctx context.Context, id, description string, body []byte) error {
|
||||
if _, ok := decodeSlot(description); ok {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case <-time.After(600 * time.Millisecond):
|
||||
}
|
||||
}
|
||||
return s.memSlots.setSlot(ctx, id, description, body)
|
||||
}
|
||||
|
||||
func TestSlotReadAheadBurst(t *testing.T) {
|
||||
oldMin, oldMax := slotTailMin, slotTailMax
|
||||
slotTailMin, slotTailMax = 150*time.Millisecond, 800*time.Millisecond
|
||||
t.Cleanup(func() { slotTailMin, slotTailMax = oldMin, oldMax })
|
||||
|
||||
client, server := slotPair(t, lagSlots{newMemSlots()}, 16, 512<<10)
|
||||
payload := randomPayload(8 * slotInlineLimit)
|
||||
go client.Write(payload)
|
||||
start := time.Now()
|
||||
got := make([]byte, len(payload))
|
||||
if _, err := io.ReadFull(server, got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatal("the burst came back changed")
|
||||
}
|
||||
if took := time.Since(start); took > 1700*time.Millisecond {
|
||||
t.Fatalf("a burst of 8 segments took %v to read", took)
|
||||
}
|
||||
client.Close()
|
||||
server.Close()
|
||||
}
|
||||
|
||||
func TestSlotLargeBurstInline(t *testing.T) {
|
||||
store := newMemSlots()
|
||||
client, server := slotPair(t, store, 16, 512<<10)
|
||||
payload := randomPayload(10 * slotInlineLimit)
|
||||
go client.Write(payload)
|
||||
got := make([]byte, len(payload))
|
||||
if _, err := io.ReadFull(server, got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatal("the burst came back changed")
|
||||
}
|
||||
store.mu.Lock()
|
||||
bodies := len(store.bodies)
|
||||
store.mu.Unlock()
|
||||
if bodies != 0 {
|
||||
t.Fatalf("a burst of 140 KiB went in %d bodies, want none", bodies)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlotDepthFollowsConcurrency(t *testing.T) {
|
||||
for _, c := range []struct{ set, want int }{{100, 100}, {maxConcurrency, maxConcurrency}, {1 << 20, maxConcurrency}} {
|
||||
p := paramsFromConfig(&Config{Concurrency: uint32(c.set)})
|
||||
r, err := makeRing(context.Background(), newMemSlots(), "session", p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(r.Up.Slots) != c.want || len(r.Down.Slots) != c.want {
|
||||
t.Fatalf("concurrency %d made %d and %d slots, want %d", c.set, len(r.Up.Slots), len(r.Down.Slots), c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlotDeepRingRoundTrip(t *testing.T) {
|
||||
client, server := slotPair(t, newMemSlots(), 64, 64<<10)
|
||||
payload := randomPayload(2 << 20)
|
||||
errs := make(chan error, 1)
|
||||
writeAll(client, payload, true, errs)
|
||||
readAll(t, server, payload)
|
||||
if err := <-errs; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
type pollSlots struct {
|
||||
*memSlots
|
||||
pmu sync.Mutex
|
||||
polled map[string]bool
|
||||
}
|
||||
|
||||
func (s *pollSlots) readDescription(ctx context.Context, id string) (string, error) {
|
||||
s.pmu.Lock()
|
||||
s.polled[id] = true
|
||||
s.pmu.Unlock()
|
||||
return s.memSlots.readDescription(ctx, id)
|
||||
}
|
||||
|
||||
func TestSlotReaderReachesPastWindow(t *testing.T) {
|
||||
store := &pollSlots{memSlots: newMemSlots(), polled: make(map[string]bool)}
|
||||
p := paramsFromConfig(&Config{FlushIntervalMs: 1, PollIntervalMs: 1, MaxPollIntervalMs: 20, EagerWindowMs: 50, Concurrency: 64})
|
||||
ring, err := makeRing(context.Background(), store, "session", p)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
defer cancel()
|
||||
put := func(seq int64) {
|
||||
slot := ring.Down.Slots[seq%int64(len(ring.Down.Slots))]
|
||||
if err := store.setSlot(ctx, slot, newSlotFrame(seq, slotInline, []byte{byte(seq)}).encode(), nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
reader := newSlotReader(ctx, store, ring.Down, p)
|
||||
next := func() []byte {
|
||||
select {
|
||||
case chunk := <-reader.chunks():
|
||||
return chunk
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("no segment came")
|
||||
return nil
|
||||
}
|
||||
}
|
||||
put(0)
|
||||
if got := next(); !bytes.Equal(got, []byte{0}) {
|
||||
t.Fatalf("segment 0 came as %v", got)
|
||||
}
|
||||
for seq := int64(2); seq <= 40; seq++ {
|
||||
put(seq)
|
||||
}
|
||||
last := ring.Down.Slots[40]
|
||||
deadline := time.Now().Add(10 * time.Second)
|
||||
for {
|
||||
reader.Wake()
|
||||
store.pmu.Lock()
|
||||
polled := store.polled[last]
|
||||
store.pmu.Unlock()
|
||||
if polled {
|
||||
break
|
||||
}
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatal("segment 40 was never asked for while segment 1 was missing")
|
||||
}
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
put(1)
|
||||
for seq := int64(1); seq <= 40; seq++ {
|
||||
if got := next(); !bytes.Equal(got, []byte{byte(seq)}) {
|
||||
t.Fatalf("segment %d came as %v", seq, got)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
type countSlots struct {
|
||||
*memSlots
|
||||
cur atomic.Int32
|
||||
peak atomic.Int32
|
||||
}
|
||||
|
||||
func (s *countSlots) setSlot(ctx context.Context, id, description string, body []byte) error {
|
||||
if _, ok := decodeSlot(description); !ok {
|
||||
return s.memSlots.setSlot(ctx, id, description, body)
|
||||
}
|
||||
n := s.cur.Add(1)
|
||||
for {
|
||||
p := s.peak.Load()
|
||||
if n <= p || s.peak.CompareAndSwap(p, n) {
|
||||
break
|
||||
}
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
s.cur.Add(-1)
|
||||
return s.memSlots.setSlot(ctx, id, description, body)
|
||||
}
|
||||
|
||||
func TestSlotWriteInflightCap(t *testing.T) {
|
||||
store := &countSlots{memSlots: newMemSlots()}
|
||||
client, server := slotPair(t, store, 48, 512<<10)
|
||||
payload := randomPayload(60 * slotInlineLimit)
|
||||
errs := make(chan error, 1)
|
||||
go func() {
|
||||
for off := 0; off < len(payload); off += slotInlineLimit {
|
||||
if _, err := client.Write(payload[off : off+slotInlineLimit]); err != nil {
|
||||
errs <- err
|
||||
return
|
||||
}
|
||||
}
|
||||
errs <- nil
|
||||
}()
|
||||
got := make([]byte, len(payload))
|
||||
if _, err := io.ReadFull(server, got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := <-errs; err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatal("the payload came back changed")
|
||||
}
|
||||
if p := int(store.peak.Load()); p > slotWriteInflight {
|
||||
t.Fatalf("%d segments were written at once, want at most %d", p, slotWriteInflight)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSlotBurstInlineWaves(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
pieces int
|
||||
media bool
|
||||
}{{20, false}, {30, true}} {
|
||||
store := newMemSlots()
|
||||
client, server := slotPair(t, store, 16, 512<<10)
|
||||
payload := randomPayload(c.pieces * slotInlineLimit)
|
||||
go client.Write(payload)
|
||||
got := make([]byte, len(payload))
|
||||
if _, err := io.ReadFull(server, got); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(got, payload) {
|
||||
t.Fatal("the burst came back changed")
|
||||
}
|
||||
store.mu.Lock()
|
||||
bodies := len(store.bodies)
|
||||
store.mu.Unlock()
|
||||
if (bodies > 0) != c.media {
|
||||
t.Fatalf("a burst of %d pieces went in %d bodies, media wanted %v", c.pieces, bodies, c.media)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/xtls/xray-core/common"
|
||||
@@ -26,6 +27,11 @@ const (
|
||||
downlinkDir = "s2c"
|
||||
)
|
||||
|
||||
const (
|
||||
spareCount = 2
|
||||
spareIdleStop = 10 * time.Minute
|
||||
)
|
||||
|
||||
func init() {
|
||||
common.Must(internet.RegisterProtocolConfigCreator(protocolName, func() interface{} {
|
||||
return new(Config)
|
||||
@@ -78,30 +84,152 @@ func streamConfig(streamSettings *internet.MemoryStreamConfig) (*Config, error)
|
||||
return config, nil
|
||||
}
|
||||
|
||||
type spareRing struct {
|
||||
storage Storage
|
||||
session string
|
||||
ring ring
|
||||
at time.Time
|
||||
}
|
||||
|
||||
type spareBox struct {
|
||||
ch chan spareRing
|
||||
busy atomic.Bool
|
||||
last atomic.Int64
|
||||
stop atomic.Bool
|
||||
}
|
||||
|
||||
var spareBoxes sync.Map
|
||||
|
||||
func spareBoxFor(config *Config) *spareBox {
|
||||
if b, ok := spareBoxes.Load(config); ok {
|
||||
return b.(*spareBox)
|
||||
}
|
||||
b, _ := spareBoxes.LoadOrStore(config, &spareBox{ch: make(chan spareRing, spareCount)})
|
||||
return b.(*spareBox)
|
||||
}
|
||||
|
||||
func dropSpare(s spareRing) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
s.storage.Delete(ctx, sessionPrefix(s.session))
|
||||
s.storage.Close()
|
||||
}
|
||||
|
||||
func takeSpare(config *Config, p params) (spareRing, bool) {
|
||||
box := spareBoxFor(config)
|
||||
for {
|
||||
select {
|
||||
case s := <-box.ch:
|
||||
if time.Since(s.at) < p.sessionTTL*2/3 {
|
||||
return s, true
|
||||
}
|
||||
go dropSpare(s)
|
||||
default:
|
||||
return spareRing{}, false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func warmSpares(streamSettings *internet.MemoryStreamConfig, config *Config, p params) {
|
||||
spareBoxFor(config).last.Store(time.Now().UnixNano())
|
||||
fillSpares(streamSettings, config, p)
|
||||
}
|
||||
|
||||
func fillSpares(streamSettings *internet.MemoryStreamConfig, config *Config, p params) {
|
||||
box := spareBoxFor(config)
|
||||
if !box.busy.CompareAndSwap(false, true) {
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
defer box.busy.Store(false)
|
||||
for n := len(box.ch); n > 0; n-- {
|
||||
select {
|
||||
case s := <-box.ch:
|
||||
if time.Since(s.at) < p.sessionTTL/3 {
|
||||
box.ch <- s
|
||||
} else {
|
||||
dropSpare(s)
|
||||
}
|
||||
default:
|
||||
}
|
||||
}
|
||||
for len(box.ch) < cap(box.ch) {
|
||||
for range 40 {
|
||||
if box.stop.Load() {
|
||||
return
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
}
|
||||
if box.stop.Load() {
|
||||
return
|
||||
}
|
||||
storage, err := newStorage(streamSettings)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
storage = sessionStorage(storage)
|
||||
slots, ok := storage.(slotStore)
|
||||
if !ok {
|
||||
storage.Close()
|
||||
return
|
||||
}
|
||||
session, err := newSessionID()
|
||||
if err != nil {
|
||||
storage.Close()
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
r, err := makeRing(ctx, slots, session, p)
|
||||
cancel()
|
||||
if err != nil {
|
||||
dropSpare(spareRing{storage: storage, session: session})
|
||||
return
|
||||
}
|
||||
select {
|
||||
case box.ch <- spareRing{storage: storage, session: session, ring: r, at: time.Now()}:
|
||||
default:
|
||||
storage.Close()
|
||||
return
|
||||
}
|
||||
}
|
||||
time.AfterFunc(p.sessionTTL/4, func() {
|
||||
if !box.stop.Load() && time.Since(time.Unix(0, box.last.Load())) < spareIdleStop {
|
||||
fillSpares(streamSettings, config, p)
|
||||
}
|
||||
})
|
||||
}()
|
||||
}
|
||||
|
||||
func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.MemoryStreamConfig) (stat.Connection, error) {
|
||||
config, err := streamConfig(streamSettings)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
storage, err := newStorage(streamSettings)
|
||||
p := paramsFromConfig(config)
|
||||
|
||||
spare, warm := takeSpare(config, p)
|
||||
storage, session, r := spare.storage, spare.session, spare.ring
|
||||
if !warm {
|
||||
storage, err = newStorage(streamSettings)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
storage = sessionStorage(storage)
|
||||
|
||||
session, err := newSessionID()
|
||||
session, err = newSessionID()
|
||||
if err != nil {
|
||||
storage.Close()
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
p := paramsFromConfig(config)
|
||||
var announce []byte
|
||||
var r ring
|
||||
slots, ok := storage.(slotStore)
|
||||
if ok {
|
||||
if r, err = makeRing(ctx, slots, session, p); err == nil {
|
||||
if warm {
|
||||
announce, _ = json.Marshal(r)
|
||||
} else if r, err = makeRing(ctx, slots, session, p); err == nil {
|
||||
announce, _ = json.Marshal(r)
|
||||
} else {
|
||||
ok = false
|
||||
@@ -115,6 +243,7 @@ func Dial(ctx context.Context, dest net.Destination, streamSettings *internet.Me
|
||||
}
|
||||
|
||||
if ok {
|
||||
warmSpares(streamSettings, config, p)
|
||||
errors.LogInfo(ctx, "opened session ", session, " in ", len(r.Up.Slots), " slots")
|
||||
return newSlotConn(context.Background(), slots, r.Up, r.Down, p, func() {
|
||||
storage.Close()
|
||||
@@ -320,12 +449,13 @@ func (l *Listener) collect() error {
|
||||
}
|
||||
l.mu.Unlock()
|
||||
|
||||
var first error
|
||||
for _, session := range expired {
|
||||
if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil {
|
||||
return err
|
||||
if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil && first == nil {
|
||||
first = err
|
||||
}
|
||||
}
|
||||
return nil
|
||||
return first
|
||||
}
|
||||
|
||||
func (l *Listener) Addr() net.Addr {
|
||||
|
||||
@@ -69,6 +69,13 @@ func pairWith(t *testing.T, streamSettings *internet.MemoryStreamConfig) (client
|
||||
}
|
||||
|
||||
return client, server, func() {
|
||||
if config, err := streamConfig(streamSettings); err == nil {
|
||||
box := spareBoxFor(config)
|
||||
box.stop.Store(true)
|
||||
for box.busy.Load() {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
}
|
||||
client.Close()
|
||||
server.Close()
|
||||
listener.Close()
|
||||
|
||||
Reference in New Issue
Block a user