Compare commits

..
Author SHA1 Message Date
patternihaandClaude Opus 5.5 9638715047 TUN inbound's autoSystemWfpBlockLeak: A CIDR in front of any value instead of "ipv4:" and "ipv6:"; Ports alone for TCP and UDP
A CIDR in front of any value limits it to its addresses, an IPv6 one in
brackets, e.g. "1.1.1.0/24:53", "[2001:db8::/32]:tcp/853" or
"[::/0]:all": "0.0.0.0/0:" and "[::/0]:" take the place of "ipv4:" and
"ipv6:". Like autoSystemRoutingTable, it takes CIDRs only, a single
address as "8.8.8.8/32", and no GeoIP.

"<port>" or "<low>-<high>" alone stands for TCP and UDP, so "dns" is
"53", "853" and "windnsclient" now.

The filters follow:
- One filter per layer permits what goes through the TUN, instead of one
  for each value.
- "all" and "misconfigtun" outweigh the local network that "allexceptlan"
  leaves open, so with it, "192.168.1.5/32:all" still blocks that address.
- The DNS Client service's mDNS and LLMNR stay allowed on the local network
  only, and only while no other value blocks them.
- The warning about blocked DNS servers only names those a value blocks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 22:21:31 +03:30
patternihaandClaude Opus 5.5 f5cc086d6c TUN inbound: README.md says "ipv4:allexceptlan" includes "ipv4:dns" only
Add "dns" to block DNS over IPv6 as well.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 18:10:11 +03:30
patternihaandClaude Opus 5.5 6b45eadf45 TUN inbound: Recommend at least "dns" for autoSystemWfpBlockLeak in README.md, and shorten it
README.md now recommends at least "dns" to block DNS leaks, and lists the
values of autoSystemWfpBlockLeak more briefly, "dns" first. It also lists
FreeBSD among the systems, and mentions that on Linux, as on Android, the
TUN can be a file descriptor from xray.tun.fd (XRAY_TUN_FD).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 17:53:27 +03:30
patternihaandClaude Opus 5.5 c684201cd8 TUN inbound's autoSystemWfpBlockLeak: "ipv4:" or "ipv6:" in front of any value instead of "tcpv4" and the like
"ipv4:" or "ipv6:" in front of any value limits it to that IP version,
e.g. "ipv4:udp/53", "ipv6:all", "ipv4:dns" or "ipv6:misconfigtun".
"tcpv4", "udpv6", "icmpv4" and the like are gone. Values that only differ
in the IP version become one: "ipv4:udp/53" with "ipv6:udp/53" is
"udp/53".

"all" overrides "allexceptlan" per IP version, and "all" over one IP
version needs dns, as port 53 over one does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 17:52:23 +03:30
patternihaandClaude Opus 5.5 2218fec173 TUN inbound's autoSystemWfpBlockLeak: Add "tcpv4", "tcpv6", "udpv4", "udpv6", "icmpv4" and "icmpv6"
"v4" or "v6" after the protocol limits a value to one IP version, ports
written the same way, e.g. "tcpv4/853" or "udpv6/10-20"; "icmpv4" is ICMP
and "icmpv6" is ICMPv6. Their filters go to the ALE layers of that version
only. "tcp", "udp" and "icmp" keep covering both.

Anything covering TCP or UDP port 53 in either IP version, like
"udpv4/53", still needs dns.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 13:09:55 +03:30
patternihaandClaude Opus 5.5 9ffa0f0552 TUN inbound's autoSystemWfpBlockLeak: Drop "/0"; Ports have to be 1-65535
"tcp/0" and "udp/0" no longer stand for all ports, which "tcp" and "udp"
already say: no syntactic sugar. Port 0 is rejected like any port outside
1-65535. Ranges keep being checked strictly: both ends within 1-65535, the
left one not greater than the right one.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 12:49:56 +03:30
patternihaandClaude Opus 5.5 2933b0c8d5 TUN inbound: Don't warn that loopback DNS servers are blocked
With autoSystemWfpBlockLeak, loopback now always stays allowed, so a DNS
server in dns like 127.0.0.1, a local resolver, stays reachable although
it lies in neither gateway nor autoSystemRoutingTable. Leave loopback
servers out of the warning that queries to them cannot go through the TUN
and are blocked, and say so in the README.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 12:27:26 +03:30
patternihaandClaude Opus 5.5 bb4cd4b6d3 TUN inbound's autoSystemWfpBlockLeak: Add "allexceptlan"
"allexceptlan" is "all" but for the local network: traffic to and from
private (10.0.0.0/8, 172.16.0.0/12, 192.168.0.0/16, fc00::/7), link-local
(169.254.0.0/16, fe80::/10), multicast (224.0.0.0/4, ff00::/8) and broadcast
(255.255.255.255) addresses stays allowed, so that printers, the router's
settings page, or casting to a TV keep working while nothing else leaves
outside the TUN.

It includes "dns": a resolver on the local network, like the router's,
would forward the queries outside the TUN, so DNS stays blocked there. The
LAN permit sits below the other values, like the permit through the TUN, so
they still apply on the local network, e.g. "tcp/445" blocks Windows file
sharing with it. The DNS Client service keeps mDNS and LLMNR, to the local
network's addresses only. "all" overrides it.

The addresses are matched with FWP_V4_ADDR_MASK and FWP_V6_ADDR_MASK
conditions on the remote address, which match when any of them does.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 11:50:18 +03:30
patternihaandClaude Opus 5.5 468f745fc2 TUN inbound's autoSystemWfpBlockLeak: Add "all", "tcp", "udp", "icmp" and port ranges; Always allow loopback, DHCP and IPv6 neighbor discovery
autoSystemWfpBlockLeak now also takes:

- "all": everything outside the TUN, IPv4 and IPv6, in both directions, so
  nothing but Xray leaves except through the TUN.
- "tcp", "udp": all ports of that protocol, also written "tcp/0", "udp/0".
- "tcp/<low>-<high>", "udp/<low>-<high>": a range of ports, matched with
  FWP_MATCH_RANGE, as two conditions on the same field would match when
  either does.
- "icmp": ICMP and ICMPv6, e.g. ping. It takes no port: in ICMP filters the
  local port holds the type, and "icmp/0" would read as the echo reply.

Values are normalized: "tcp/0" and "tcp/1-65535" become "tcp", and
"tcp/53-53" becomes "tcp/53".

Whatever is blocked, loopback, which cannot leave the system, and what
Windows itself needs on the local link now always stay allowed: DHCP, DHCPv6,
and IPv6 neighbor and multicast listener discovery. These used to be allowed
only along with "misconfigtun", so "udp" or "icmp" would otherwise have cut
DHCP or IPv6 neighbor discovery, and "tcp" any local proxy, Xray's own
inbounds included. Loopback being allowed means that with "dns", a local
resolver on 127.0.0.1:53 now works for other programs (its own queries to
port 53 or 853 outside the TUN stay blocked), and the DNS Client service
may use it.

"all" reuses the filters of "misconfigtun", which now also let everything
through the TUN, for both IP versions whatever the routes.

On other TUN options: keeping the system's DNS inside the TUN, with "all",
"dns", "windnsclient", or anything covering TCP or UDP port 53 ("udp",
"tcp/50-60", ...), needs dns, as Windows can no longer reach the DNS
servers of the other interfaces; the config is rejected otherwise. Like the
other values, the new ones need autoSystemRoutingTable. The filters do not
need autoOutboundsInterface, as Xray is allowed by its app ID; it still
keeps Xray's own connections out of the TUN's routes.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-02 11:13:10 +03:30
34 changed files with 135 additions and 1381 deletions
+3 -3
View File
@@ -9,8 +9,8 @@ ARG TARGETARCH
RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -o xray -trimpath -buildvcs=false -gcflags="all=-l=4" -ldflags "-s -w -buildid=" ./main
# Download geodat into a staging directory
ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geoip-full.dat /tmp/geodat/geoip.dat
ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geosite-expand.dat /tmp/geodat/geosite.dat
ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geoip/geoip-full.dat /tmp/geodat/geoip.dat
ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geosite/geosite-expand.dat /tmp/geodat/geosite.dat
RUN mkdir -p /tmp/empty
@@ -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
}
+3 -3
View File
@@ -9,8 +9,8 @@ ARG TARGETARCH
RUN GOOS=$TARGETOS GOARCH=$TARGETARCH CGO_ENABLED=0 go build -o xray -trimpath -buildvcs=false -gcflags="all=-l=4" -ldflags "-s -w -buildid=" ./main
# Download geodat into a staging directory
ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geoip-full.dat /tmp/geodat/geoip.dat
ADD https://github.com/XTLS/Xray-rules-dat/releases/latest/download/geosite-expand.dat /tmp/geodat/geosite.dat
ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geoip/geoip-full.dat /tmp/geodat/geoip.dat
ADD https://raw.githubusercontent.com/XTLS/Xray-rules-dat/release/geosite/geosite-expand.dat /tmp/geodat/geosite.dat
RUN mkdir -p /tmp/empty
@@ -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
}
+1 -1
View File
@@ -73,7 +73,7 @@ func (r *TimeoutWrapperReader) ReadMultiBufferTimeout(duration time.Duration) (M
}
return r.mb, r.err
case <-timeout:
return nil, ErrReadTimeout
return nil, nil
}
}
+1 -1
View File
@@ -134,7 +134,7 @@ func TestMarshalConfigJson(t *testing.T) {
func getConfig() string {
return `{
"log": {
"logLevel": "debug"
"loglevel": "debug"
},
"stats": {},
"policy": {
+1 -1
View File
@@ -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.2
github.com/klauspost/cpuid/v2 v2.4.1
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
+2 -2
View File
@@ -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.2 h1:xRw7IZEsAgVCp6trS/DgWbTAqpixhmywgVgBjYGg1bQ=
github.com/klauspost/cpuid/v2 v2.4.2/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU=
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/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=
+3 -33
View File
@@ -9,7 +9,6 @@ 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"
@@ -49,8 +48,6 @@ 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"`
}
@@ -283,37 +280,10 @@ func (c *FreedomFinalRuleConfig) Build() (*freedom.FinalRuleConfig, error) {
}
if c.BlockDelay != nil {
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")
rule.BlockDelay = &freedom.Range{
Min: uint64(c.BlockDelay.From),
Max: uint64(c.BlockDelay.To),
}
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
+5 -5
View File
@@ -4,7 +4,6 @@ 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"
@@ -47,8 +46,7 @@ func TestFreedomConfig(t *testing.T) {
"network": "tcp,udp",
"port": "53,443",
"ip": ["10.0.0.0/8", "2001:db8::/32"],
"delay": "30-60",
"logLevel": "debug"
"blockDelay": "30-60"
}, {
"action": "allow",
"network": ["udp"]
@@ -88,8 +86,10 @@ func TestFreedomConfig(t *testing.T) {
},
},
},
Delay: &freedom.Range{Min: 30, Max: 60},
LogLevel: log.Severity_Debug.Enum(),
BlockDelay: &freedom.Range{
Min: 30,
Max: 60,
},
},
{
Action: freedom.RuleAction_Allow,
+1 -11
View File
@@ -2,7 +2,6 @@ package conf
import (
"net/url"
"strings"
"github.com/robfig/cron/v3"
"github.com/xtls/xray-core/app/geodata"
@@ -20,20 +19,11 @@ func (c *GeodataAssetConfig) Build() (*geodata.Asset, error) {
return nil, errors.New("invalid geodata asset url: ", c.URL).Base(err)
}
return &geodata.Asset{
Url: migrateGeodataURL(c.URL),
Url: c.URL,
File: c.File,
}, nil
}
func migrateGeodataURL(assetURL string) string {
if strings.HasSuffix(strings.ToLower(assetURL), ".dat") && (strings.HasPrefix(strings.ToLower(assetURL), "https://github.com/xtls/xray-rules-dat/") ||
strings.HasPrefix(strings.ToLower(assetURL), "https://raw.githubusercontent.com/xtls/xray-rules-dat/")) {
errors.PrintDeprecatedFeatureWarning(assetURL, assetURL+".gz")
return assetURL + ".gz"
}
return assetURL
}
func validateHTTPS(s string) error {
u, err := url.ParseRequestURI(s)
if err != nil {
+1 -1
View File
@@ -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"`
}
+2 -2
View File
@@ -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",
+1 -1
View File
@@ -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": [{
+20 -34
View File
@@ -8,7 +8,6 @@ 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"
@@ -358,8 +357,7 @@ 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"`
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"`
BlockDelay *Range `protobuf:"bytes,5,opt,name=block_delay,json=blockDelay,proto3" json:"block_delay,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -422,20 +420,13 @@ func (x *FinalRuleConfig) GetIp() []*geodata.IPRule {
return nil
}
func (x *FinalRuleConfig) GetDelay() *Range {
func (x *FinalRuleConfig) GetBlockDelay() *Range {
if x != nil {
return x.Delay
return x.BlockDelay
}
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"`
@@ -532,7 +523,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\x1a\x14common/log/log.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\"S\n" +
"\x13DestinationOverride\x12<\n" +
"\x06server\x18\x01 \x01(\v2$.xray.common.protocol.ServerEndpointR\x06server\"\x98\x02\n" +
"\bFragment\x12!\n" +
@@ -558,16 +549,14 @@ 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\"\xe0\x02\n" +
"\x03max\x18\x02 \x01(\x04R\x03max\"\xa0\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" +
"\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" +
"\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" +
"\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" +
@@ -610,8 +599,7 @@ 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
(log.Severity)(0), // 11: xray.common.log.Severity
(internet.DomainStrategy)(0), // 12: xray.transport.internet.DomainStrategy
(internet.DomainStrategy)(0), // 11: 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
@@ -619,18 +607,17 @@ 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.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
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
}
func init() { file_proxy_freedom_config_proto_init() }
@@ -638,7 +625,6 @@ 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{
+1 -3
View File
@@ -11,7 +11,6 @@ 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;
@@ -52,8 +51,7 @@ message FinalRuleConfig {
repeated xray.common.net.Network networks = 2;
xray.common.net.PortList port_list = 3;
repeated xray.common.geodata.IPRule ip = 4;
Range delay = 5;
optional xray.common.log.Severity log_level = 6;
Range block_delay = 5;
}
message Config {
+20 -42
View File
@@ -15,7 +15,6 @@ 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"
@@ -73,26 +72,23 @@ func init() {
}
defaultBlockPrivateRule = &FinalRule{
action: RuleAction_Block,
network: allNetworks,
ip: geodata.GetPrivateIPMatcher(),
logLevel: log.Severity_Warning.Enum(),
action: RuleAction_Block,
network: allNetworks,
ip: geodata.GetPrivateIPMatcher(),
}
defaultBlockAllRule = &FinalRule{
action: RuleAction_Block,
network: allNetworks,
logLevel: log.Severity_Warning.Enum(),
action: RuleAction_Block,
network: allNetworks,
}
}
type FinalRule struct {
action RuleAction
network [8]bool
port net.MemoryPortList
ip geodata.IPMatcher
delay *Range
logLevel *log.Severity
action RuleAction
network [8]bool
port net.MemoryPortList
ip geodata.IPMatcher
blockDelay *Range
}
// Handler handles Freedom connections.
@@ -106,9 +102,8 @@ type Handler struct {
func buildFinalRule(config *FinalRuleConfig) (*FinalRule, error) {
rule := &FinalRule{
action: config.GetAction(),
delay: config.GetDelay(),
logLevel: config.LogLevel,
action: config.GetAction(),
blockDelay: config.GetBlockDelay(),
}
if len(config.Networks) == 0 {
@@ -169,7 +164,7 @@ func getDefaultFinalRule(inbound *session.Inbound) *FinalRule {
switch inbound.Name {
case "vless-reverse":
return defaultBlockAllRule
case "vless", "vmess", "trojan", "hysteria", "wireguard", "masque":
case "vless", "vmess", "trojan", "hysteria", "wireguard":
return defaultBlockPrivateRule
default:
if strings.HasPrefix(inbound.Name, "shadowsocks") {
@@ -217,12 +212,12 @@ func (h *Handler) policy() policy.Session {
return p
}
func (r *FinalRule) blockDuration() time.Duration {
func (h *Handler) blockDelay(rule *FinalRule) time.Duration {
min := uint64(30)
max := uint64(90)
if r.delay != nil {
min = r.delay.Min
max = r.delay.Max
if rule.blockDelay != nil {
min = rule.blockDelay.Min
max = rule.blockDelay.Max
}
span := max - min
if max < min {
@@ -231,26 +226,9 @@ func (r *FinalRule) blockDuration() 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 := rule.blockDuration()
rule.log(ctx, "target blocked by a finalrule: ", *dest, ", blackholing connection for ", delay)
delay := h.blockDelay(rule)
errors.LogInfo(ctx, "blocked target: ", *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
@@ -262,7 +240,7 @@ func (h *Handler) blackhole(ctx context.Context, input buf.Reader, output buf.Wr
case <-time.After(delay):
common.Interrupt(input)
common.Interrupt(output)
rule.log(ctx, "closed blackholed connection to blocked target: ", *dest)
errors.LogInfo(ctx, "closed blackholed connection to blocked target: ", *dest)
}
return nil
}
+1 -1
View File
@@ -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`, `dns` 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` 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.
+1 -8
View File
@@ -115,7 +115,6 @@ 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{})
@@ -176,7 +175,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 && !dns6 {
if err == windows.ERROR_NOT_FOUND && family == windows.AF_INET6 && !address6 && !route6 {
continue
}
return err
@@ -202,12 +201,6 @@ 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
+2 -6
View File
@@ -17,7 +17,7 @@ import (
type bind struct {
ctx context.Context
resolveFunc func(ctx context.Context, host string) (net.IP, error)
resolveFunc func(host string) (net.IP, error)
listenFunc func(ctx context.Context) (net.PacketConn, error)
downFunc func() error
reserved []byte
@@ -143,11 +143,7 @@ func (b *bind) ParseEndpoint(s string) (conn.Endpoint, error) {
if port < 0 || port > 65535 {
return nil, errors.New("invalid port " + sport)
}
// 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)
ip, err := b.resolveFunc(host)
if err != nil {
return nil, err
}
+3 -24
View File
@@ -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(ctx context.Context, host string) (net.IP, error) {
return func(ctx context.Context, host string) (net.IP, error) {
func MakeResolveLocal(d dns.Client, streamSettings *internet.MemoryStreamConfig) func(host string) (net.IP, error) {
return func(host string) (net.IP, error) {
if ip := net.ParseIP(host); ip != nil {
return ip, nil
}
ips, err := lookupIPContext(ctx, d, host)
ips, _, err := d.LookupIP(host, dns.IPOption{IPv4Enable: true, IPv6Enable: true})
if err != nil {
return nil, err
}
@@ -368,27 +368,6 @@ 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
+8 -8
View File
@@ -661,7 +661,7 @@ func TestVlessRealityFingerprints(t *testing.T) {
wg.Wait()
}
// A blocked connection must close after finalRules.delay even while the client keeps it open and sends nothing.
// A blocked connection must be closed after blockDelay 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("delay=", delay), func(t *testing.T) {
t.Run(fmt.Sprint("blockDelay=", delay), func(t *testing.T) {
userID := protocol.NewID(uuid.New())
serverPort := tcp.PickPort()
serverConfig := &core.Config{
@@ -697,8 +697,8 @@ func TestVlessBlockDelay(t *testing.T) {
ProxySettings: serial.ToTypedMessage(&freedom.Config{
FinalRules: []*freedom.FinalRuleConfig{
{
Action: freedom.RuleAction_Block,
Delay: &freedom.Range{Min: delay, Max: delay},
Action: freedom.RuleAction_Block,
BlockDelay: &freedom.Range{Min: delay, Max: delay},
},
},
}),
@@ -726,15 +726,15 @@ func TestVlessBlockDelay(t *testing.T) {
Port: dest.Port,
}, &encoding.Addons{}))
blockDuration := time.Duration(delay) * time.Second
common.Must(conn.SetReadDeadline(time.Now().Add(blockDuration + 5*time.Second)))
blockDelay := time.Duration(delay) * time.Second
common.Must(conn.SetReadDeadline(time.Now().Add(blockDelay + 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 < blockDuration {
t.Fatalf("connection is closed after %v, before finalRules.delay", elapsed)
if elapsed < blockDelay {
t.Fatalf("connection is closed after %v, before blockDelay", elapsed)
}
})
}
+1 -2
View File
@@ -6,7 +6,6 @@ import (
goerrors "errors"
"io"
mrand "math/rand"
stdnet "net"
"net/netip"
"sync"
"time"
@@ -165,7 +164,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() || goerrors.Is(err, stdnet.ErrClosed) {
if c.closed() {
return
}
var netErr net.Error
+4 -23
View File
@@ -438,11 +438,9 @@ func (b *doneBody) Close() error {
type sessionTransport struct {
pool *connPool
mu sync.Mutex
client *splithttp.XmuxClient
closed bool
slowAt, retiredAt time.Time
movedAt time.Time
mu sync.Mutex
client *splithttp.XmuxClient
closed bool
}
func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
@@ -452,7 +450,7 @@ func (t *sessionTransport) RoundTrip(r *http.Request) (*http.Response, error) {
return nil, errSessionClosed
}
client := t.client
if client == nil || client.XmuxConn.IsClosed() || client.LeftRequests.Add(-1) < 0 || !client.UnreusableAt.IsZero() && time.Now().After(client.UnreusableAt) {
if client == nil || 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()
@@ -460,7 +458,6 @@ 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
}
@@ -474,22 +471,6 @@ 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()
-100
View File
@@ -987,103 +987,3 @@ 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)
}
}
+13 -88
View File
@@ -21,12 +21,11 @@ import (
const (
flatSeparator = "~"
driveBoundary = "xdrive-boundary"
drivePageSize = 1000
driveMaxAttempts = 8
driveMaxInflight = 64
driveReservedInflight = 8
driveInlineLimit = 12000
driveBoundary = "xdrive-boundary"
drivePageSize = 1000
driveMaxAttempts = 8
driveMaxInflight = 32
driveInlineLimit = 12000
// driveTimeout ends a request that makes no progress for that long
driveTimeout = 60 * time.Second
)
@@ -37,21 +36,13 @@ 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()
}
@@ -74,9 +65,6 @@ 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) {
@@ -112,7 +100,6 @@ func newDriveStorage(streamSettings *internet.MemoryStreamConfig, config *Config
},
client: client,
inflight: make(chan struct{}, driveMaxInflight),
reserved: make(chan struct{}, driveReservedInflight),
}, nil
}
@@ -124,7 +111,6 @@ 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,
}
}
@@ -233,17 +219,6 @@ 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 {
@@ -263,12 +238,8 @@ 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 pool <- struct{}{}:
case s.inflight <- struct{}{}:
case <-ctx.Done():
return 0, nil, ctx.Err()
}
@@ -277,14 +248,8 @@ func (s *driveStorage) do(ctx context.Context, method, target, contentType strin
if body != nil {
reader = bytes.NewReader(body)
}
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)
req, err := http.NewRequestWithContext(ctx, 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)
@@ -292,17 +257,9 @@ 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)
<-pool
<-s.inflight
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()
}
@@ -313,7 +270,6 @@ 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()
@@ -330,19 +286,13 @@ 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, ", ", method, " ", len(body),
" bytes, body read ", strings.Contains(target, "alt=media"),
", lane ", ctx.Value(laneKey{}) != nil, ": ", string(payload[:min(len(payload), 300)]))
s.cool()
attempt+1, " of ", driveMaxAttempts)
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
@@ -351,12 +301,6 @@ 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 = ""
@@ -598,27 +542,12 @@ 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 {
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()
}
}(childName, file.id)
if err := s.deleteID(ctx, childName, file.id); err != nil {
return err
}
}
wg.Wait()
return first
return nil
}
func (s *driveStorage) List(ctx context.Context, prefix string) ([]Entry, error) {
@@ -675,9 +604,6 @@ 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)
@@ -705,7 +631,6 @@ 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,13 +7,8 @@ import (
"encoding/json"
"fmt"
"io"
"net/http"
"net/http/httptrace"
"net/url"
"os"
"sort"
"strconv"
"strings"
"sync"
"testing"
"time"
@@ -348,130 +343,3 @@ 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)
}
+1 -1
View File
@@ -72,7 +72,7 @@ func (s *localStorage) Get(ctx context.Context, name string) ([]byte, error) {
if err != nil {
return nil, err
}
data, err := readFile(ctx, full)
data, err := os.ReadFile(full)
if err != nil {
if os.IsNotExist(err) {
return nil, errNotFound
-12
View File
@@ -1,12 +0,0 @@
//go:build !windows
package xdrive
import (
"context"
"os"
)
func readFile(_ context.Context, name string) ([]byte, error) {
return os.ReadFile(name)
}
@@ -1,30 +0,0 @@
package xdrive
import (
"context"
"errors"
"os"
"time"
"golang.org/x/sys/windows"
)
// readFile retries os.ReadFile for up to 2 seconds on sharing violations:
// os.Rename (MoveFileEx) in Put holds the file with DELETE access, which
// os.ReadFile does not share, until a moment after its new name appears in List
func readFile(ctx context.Context, name string) ([]byte, error) {
deadline := time.Now().Add(2 * time.Second)
delay := time.Millisecond
for {
data, err := os.ReadFile(name)
if !errors.Is(err, windows.ERROR_SHARING_VIOLATION) || time.Now().Add(delay).After(deadline) {
return data, err
}
select {
case <-ctx.Done():
return nil, err
case <-time.After(delay):
}
delay = min(2*delay, 100*time.Millisecond)
}
}
@@ -1,57 +0,0 @@
package xdrive
import (
"context"
"errors"
"sync"
"testing"
"time"
"golang.org/x/sys/windows"
)
func TestLocalSharingViolation(t *testing.T) {
storage, err := newLocalStorage(t.TempDir())
if err != nil {
t.Fatalf("newLocalStorage: %v", err)
}
const name = "streams/session/c2s/000000000.seg"
if err := storage.Put(context.Background(), name, []byte("segment")); err != nil {
t.Fatalf("Put: %v", err)
}
full, err := storage.resolve(name)
if err != nil {
t.Fatalf("resolve: %v", err)
}
path, err := windows.UTF16PtrFromString(full)
if err != nil {
t.Fatalf("UTF16PtrFromString: %v", err)
}
// os.Rename holds the file like this until a moment after its new name
// appears
handle, err := windows.CreateFile(path, windows.DELETE,
windows.FILE_SHARE_READ|windows.FILE_SHARE_WRITE|windows.FILE_SHARE_DELETE,
nil, windows.OPEN_EXISTING, 0, 0)
if err != nil {
t.Fatalf("CreateFile: %v", err)
}
var once sync.Once
release := func() { once.Do(func() { windows.CloseHandle(handle) }) }
defer release()
ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()
if _, err := storage.Get(ctx, name); !errors.Is(err, windows.ERROR_SHARING_VIOLATION) {
t.Fatalf("Get returned %v while the file is held, want a sharing violation", err)
}
time.AfterFunc(100*time.Millisecond, release)
data, err := storage.Get(context.Background(), name)
if err != nil {
t.Fatalf("Get: %v", err)
}
if string(data) != "segment" {
t.Fatalf("Get returned %q, want %q", data, "segment")
}
}
+1 -1
View File
@@ -13,7 +13,7 @@ const (
defaultConcurrency = 8
maxSegmentBytes = 16 * 1024 * 1024
maxConcurrency = 256
maxConcurrency = 64
)
type params struct {
+20 -225
View File
@@ -10,7 +10,6 @@ import (
"io"
"strconv"
"sync"
"sync/atomic"
"time"
"github.com/xtls/xray-core/common/errors"
@@ -28,6 +27,7 @@ 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,26 +41,19 @@ const (
var (
// the head is asked for between slotHuntFloor and slotHuntCeiling, a segment
// behind it at about the pace that segments come
slotHuntFloor = 10 * time.Millisecond
slotHuntCeiling = 25 * time.Millisecond
slotHuntFloor = 50 * time.Millisecond
slotHuntCeiling = 200 * time.Millisecond
slotTailMin = 150 * time.Millisecond
slotTailMax = 800 * time.Millisecond
slotFetchAhead = 12
slotFetchAhead = 6
slotAckRetry = 30 * time.Millisecond
slotWatermarkTTL = 100 * time.Millisecond
slotAckRetry = 100 * time.Millisecond
slotWatermarkTTL = 250 * 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 {
@@ -150,7 +143,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 := max(p.concurrency, 2)
depth := min(max(p.concurrency, 2), slotDepthLimit)
r := ring{
Up: slotTable{Slots: make([]string, depth)},
Down: slotTable{Slots: make([]string, depth)},
@@ -184,7 +177,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(context.WithValue(ctx, laneKey{}, true), id)
d, err := store.readDescription(ctx, id)
if err != nil || d == "" {
return -1, err
}
@@ -223,7 +216,7 @@ func newSlotWriter(ctx context.Context, store slotStore, table slotTable, p para
store: store,
table: table,
params: p,
slots: make(chan struct{}, min(len(table.Slots), slotWriteInflight)),
slots: make(chan struct{}, len(table.Slots)),
size: min(slotInlineLimit, p.segmentBytes),
acked: -1,
progress: time.Now(),
@@ -278,10 +271,6 @@ 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
@@ -307,19 +296,11 @@ 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
@@ -351,21 +332,15 @@ 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 {
@@ -418,7 +393,7 @@ func (w *slotWriter) await(seq int64) error {
w.ackMu.Lock()
still := time.Since(w.progress)
w.ackMu.Unlock()
if min(time.Since(start), still) > slotStallTimeout || closed && still > slotCloseGrace {
if time.Since(start) > slotStallTimeout || closed && still > slotCloseGrace {
return errors.New("the reader took nothing from slot ", w.slotID(seq), " for ", still).Base(err)
}
@@ -486,7 +461,6 @@ type slotReader struct {
ch chan []byte
wake chan struct{}
acks chan struct{}
grow chan struct{}
mu sync.Mutex
head int64
@@ -498,11 +472,7 @@ 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 {
@@ -514,12 +484,9 @@ 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()
@@ -549,20 +516,9 @@ 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() {
@@ -573,20 +529,17 @@ func (r *slotReader) run() {
fetches := make(map[int64]*slotFetch)
for {
r.mu.Lock()
head, started, frontier := r.head, r.started, r.frontier
head, started := r.head, r.started
r.mu.Unlock()
window := 1
if started {
window = min(max(int(frontier-head)+slotFetchAhead, slotFetchAhead), len(r.table.Slots))
window = min(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{}), near: make(chan struct{}), born: time.Now()}
if int64(i) <= slotReadAhead {
fetches[seq].enter()
}
fetches[seq] = &slotFetch{ready: make(chan struct{}), promoted: make(chan struct{})}
go r.fetch(ctx, seq, i, fetches[seq])
}
}
@@ -594,8 +547,6 @@ func (r *slotReader) run() {
f := fetches[head]
select {
case <-f.ready:
case <-r.grow:
continue
case <-ctx.Done():
return
}
@@ -613,14 +564,8 @@ 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{}{}:
@@ -641,35 +586,11 @@ 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 soon after it was asked for, then at the pace that segments come
// behind it at the pace that segments come, as it would wait for the head anyway
func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFetch) {
defer func() {
f.readyAt = time.Now()
close(f.ready)
}()
if offset > 0 && !r.hold(ctx, r.tailMin, f.promoted, f.near) {
defer close(f.ready)
if offset > 0 && !r.hold(ctx, time.Duration(offset)*r.pace(), f.promoted, nil) {
return
}
@@ -677,17 +598,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
backoff := slotHuntFloor
var goneSince, warnedAt time.Time
for {
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)
}
description, err := r.store.readDescription(ctx, id)
frame, ok := decodeSlot(description)
switch {
case err == nil && ok && frame.seq == seq:
@@ -695,7 +606,6 @@ 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
@@ -708,7 +618,7 @@ func (r *slotReader) fetch(ctx context.Context, seq int64, offset int, f *slotFe
r.hit()
return
case slotMedia:
if body, err := r.readBodyHedged(ctx, id); err == nil && frame.holds(body) {
if body, err := r.store.readBody(ctx, id); err == nil && frame.holds(body) {
f.body = body
r.hit()
return
@@ -744,15 +654,7 @@ 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, 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:
}
wait, a, backoff = min(max(r.pace(), slotTailMin), slotTailMax), f.promoted, slotHuntFloor
}
if !r.hold(ctx, wait, a, b) {
return
@@ -760,106 +662,6 @@ 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()
@@ -924,7 +726,7 @@ func (r *slotReader) ackLoop() {
if want <= acked {
break
}
if err := r.store.setSlot(context.WithValue(r.ctx, laneKey{}, true), r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil {
if err := r.store.setSlot(r.ctx, r.table.Watermark, strconv.FormatInt(want, 10), nil); err == nil {
acked, wait = want, slotAckRetry
continue
}
@@ -954,10 +756,3 @@ func (r *slotReader) Err() error {
}
return io.EOF
}
func slotTag(id string) string {
if len(id) > 6 {
return id[len(id)-6:]
}
return id
}
-368
View File
@@ -10,7 +10,6 @@ import (
"os"
"strconv"
"sync"
"sync/atomic"
"testing"
"time"
)
@@ -305,373 +304,6 @@ 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)
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)
}
}
}
+15 -145
View File
@@ -9,7 +9,6 @@ import (
"strconv"
"strings"
"sync"
"sync/atomic"
"time"
"github.com/xtls/xray-core/common"
@@ -27,11 +26,6 @@ const (
downlinkDir = "s2c"
)
const (
spareCount = 2
spareIdleStop = 10 * time.Minute
)
func init() {
common.Must(internet.RegisterProtocolConfigCreator(protocolName, func() interface{} {
return new(Config)
@@ -84,152 +78,30 @@ 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
}
p := paramsFromConfig(config)
storage, err := newStorage(streamSettings)
if err != nil {
return nil, err
}
storage = sessionStorage(storage)
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()
if err != nil {
storage.Close()
return nil, err
}
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 warm {
announce, _ = json.Marshal(r)
} else if r, err = makeRing(ctx, slots, session, p); err == nil {
if r, err = makeRing(ctx, slots, session, p); err == nil {
announce, _ = json.Marshal(r)
} else {
ok = false
@@ -243,7 +115,6 @@ 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()
@@ -449,13 +320,12 @@ 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 && first == nil {
first = err
if err := l.storage.Delete(l.ctx, sessionPrefix(session)); err != nil {
return err
}
}
return first
return nil
}
func (l *Listener) Addr() net.Addr {
-7
View File
@@ -69,13 +69,6 @@ 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()