xmengnet/mihomo
1
1
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

chore: packet deadline support CreateReadWaiter interface

Browse source
This commit is contained in:
gVisor bot 2023-05-20 11:44:11 +08:00
parent 9ff246b29d
commit 5af4f4995f
3 changed files with 171 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

27
common/net/deadline/packet.go
View file

@ -3,6 +3,7 @@ package deadline
import (
"net"
"os"
"runtime"
"time"
"github.com/Dreamacro/clash/common/atomic"
@ -13,8 +14,6 @@ type readResult struct {
data []byte
addr net.Addr
err error
enhanceReadResult
singReadResult
}
type NetPacketConn struct {
@ -23,14 +22,14 @@ type NetPacketConn struct {
pipeDeadline pipeDeadline
disablePipe atomic.Bool
inRead atomic.Bool
resultCh chan *readResult
resultCh chan any
}
func NewNetPacketConn(pc net.PacketConn) net.PacketConn {
npc := &NetPacketConn{
PacketConn: pc,
pipeDeadline: makePipeDeadline(),
resultCh: make(chan *readResult, 1),
resultCh: make(chan any, 1),
}
npc.resultCh <- nil
if enhancePC, isEnhance := pc.(packet.EnhancePacketConn); isEnhance {
@ -65,21 +64,29 @@ func NewNetPacketConn(pc net.PacketConn) net.PacketConn {
}
func (c *NetPacketConn) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
FOR:
for {
select {
case result := <-c.resultCh:
if result != nil {
if result, ok := result.(*readResult); ok {
n = copy(p, result.data)
addr = result.addr
err = result.err
c.resultCh <- nil // finish cache read
return
}
c.resultCh <- result // another type of read
runtime.Gosched() // allowing other goroutines to run
continue FOR
} else {
c.resultCh <- nil
break
break FOR
}
case <-c.pipeDeadline.wait():
return 0, nil, os.ErrDeadlineExceeded
}
}
if c.disablePipe.Load() {
return c.PacketConn.ReadFrom(p)
@ -100,11 +107,11 @@ func (c *NetPacketConn) pipeReadFrom(size int) {
buffer := make([]byte, size)
n, addr, err := c.PacketConn.ReadFrom(buffer)
buffer = buffer[:n]
c.resultCh <- &readResult{
data: buffer,
addr: addr,
err: err,
}
result := &readResult{}
result.data = buffer
result.addr = addr
result.err = err
c.resultCh <- result
}
func (c *NetPacketConn) SetReadDeadline(t time.Time) error {

28
common/net/deadline/packet_enhance.go
View file

@ -3,6 +3,7 @@ package deadline
import (
"net"
"os"
"runtime"
"github.com/Dreamacro/clash/common/net/packet"
)
@ -19,7 +20,10 @@ func NewEnhancePacketConn(pc packet.EnhancePacketConn) packet.EnhancePacketConn
}
type enhanceReadResult struct {
data []byte
put func()
addr net.Addr
err error
}
type enhancePacketConn struct {
@ -28,22 +32,30 @@ type enhancePacketConn struct {
}
func (c *enhancePacketConn) WaitReadFrom() (data []byte, put func(), addr net.Addr, err error) {
FOR:
for {
select {
case result := <-c.netPacketConn.resultCh:
if result != nil {
if result, ok := result.(*enhanceReadResult); ok {
data = result.data
put = result.put
addr = result.addr
err = result.err
c.netPacketConn.resultCh <- nil // finish cache read
return
}
c.netPacketConn.resultCh <- result // another type of read
runtime.Gosched() // allowing other goroutines to run
continue FOR
} else {
c.netPacketConn.resultCh <- nil
break
break FOR
}
case <-c.netPacketConn.pipeDeadline.wait():
return nil, nil, nil, os.ErrDeadlineExceeded
}
}
if c.netPacketConn.disablePipe.Load() {
return c.enhancePacketConn.WaitReadFrom()
@ -62,12 +74,10 @@ func (c *enhancePacketConn) WaitReadFrom() (data []byte, put func(), addr net.Ad
func (c *enhancePacketConn) pipeWaitReadFrom() {
data, put, addr, err := c.enhancePacketConn.WaitReadFrom()
c.netPacketConn.resultCh <- &readResult{
data: data,
enhanceReadResult: enhanceReadResult{
put: put,
},
addr: addr,
err: err,
}
result := &enhanceReadResult{}
result.data = data
result.put = put
result.addr = addr
result.err = err
c.netPacketConn.resultCh <- result
}

100
common/net/deadline/packet_sing.go
View file

@ -2,10 +2,13 @@ package deadline
import (
"os"
"runtime"
"github.com/Dreamacro/clash/common/net/packet"
"github.com/sagernet/sing/common/buf"
"github.com/sagernet/sing/common/bufio"
M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network"
)
type SingPacketConn struct {
@ -33,6 +36,7 @@ var _ packet.EnhanceSingPacketConn = (*EnhanceSingPacketConn)(nil)
type singReadResult struct {
buffer *buf.Buffer
destination M.Socksaddr
err error
}
type singPacketConn struct {
@ -41,9 +45,12 @@ type singPacketConn struct {
}
func (c *singPacketConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr, err error) {
FOR:
for {
select {
case result := <-c.netPacketConn.resultCh:
if result != nil {
if result, ok := result.(*singReadResult); ok {
destination = result.destination
err = result.err
buffer.Resize(result.buffer.Start(), 0)
@ -55,13 +62,18 @@ func (c *singPacketConn) ReadPacket(buffer *buf.Buffer) (destination M.Socksaddr
}
c.netPacketConn.resultCh <- nil // finish cache read
return
}
c.netPacketConn.resultCh <- result // another type of read
runtime.Gosched() // allowing other goroutines to run
continue FOR
} else {
c.netPacketConn.resultCh <- nil
break
break FOR
}
case <-c.netPacketConn.pipeDeadline.wait():
return M.Socksaddr{}, os.ErrDeadlineExceeded
}
}
if c.netPacketConn.disablePipe.Load() {
return c.singPacketConn.ReadPacket(buffer)
@ -82,15 +94,87 @@ func (c *singPacketConn) pipeReadPacket(bufLen int, bufStart int) {
buffer := buf.NewSize(bufLen)
buffer.Advance(bufStart)
destination, err := c.singPacketConn.ReadPacket(buffer)
c.netPacketConn.resultCh <- &readResult{
singReadResult: singReadResult{
buffer: buffer,
destination: destination,
},
err: err,
}
result := &singReadResult{}
result.destination = destination
result.err = err
c.netPacketConn.resultCh <- result
}
func (c *singPacketConn) WritePacket(buffer *buf.Buffer, destination M.Socksaddr) error {
return c.singPacketConn.WritePacket(buffer, destination)
}
func (c *singPacketConn) CreateReadWaiter() (N.PacketReadWaiter, bool) {
prw, isReadWaiter := bufio.CreatePacketReadWaiter(c.singPacketConn)
if isReadWaiter {
return &singPacketReadWaiter{
netPacketConn: c.netPacketConn,
packetReadWaiter: prw,
}, true
}
return nil, false
}
var _ N.PacketReadWaiter = (*singPacketReadWaiter)(nil)
type singPacketReadWaiter struct {
netPacketConn *NetPacketConn
packetReadWaiter N.PacketReadWaiter
}
type singWaitReadResult singReadResult
func (c *singPacketReadWaiter) InitializeReadWaiter(newBuffer func() *buf.Buffer) {
c.packetReadWaiter.InitializeReadWaiter(newBuffer)
}
func (c *singPacketReadWaiter) WaitReadPacket() (destination M.Socksaddr, err error) {
FOR:
for {
select {
case result := <-c.netPacketConn.resultCh:
if result != nil {
if result, ok := result.(*singWaitReadResult); ok {
destination = result.destination
err = result.err
c.netPacketConn.resultCh <- nil // finish cache read
return
}
c.netPacketConn.resultCh <- result // another type of read
runtime.Gosched() // allowing other goroutines to run
continue FOR
} else {
c.netPacketConn.resultCh <- nil
break FOR
}
case <-c.netPacketConn.pipeDeadline.wait():
return M.Socksaddr{}, os.ErrDeadlineExceeded
}
}
if c.netPacketConn.disablePipe.Load() {
return c.packetReadWaiter.WaitReadPacket()
} else if c.netPacketConn.deadline.Load().IsZero() {
c.netPacketConn.inRead.Store(true)
defer c.netPacketConn.inRead.Store(false)
destination, err = c.packetReadWaiter.WaitReadPacket()
return
}
<-c.netPacketConn.resultCh
go c.pipeWaitReadPacket()
return c.WaitReadPacket()
}
func (c *singPacketReadWaiter) pipeWaitReadPacket() {
destination, err := c.packetReadWaiter.WaitReadPacket()
result := &singWaitReadResult{}
result.destination = destination
result.err = err
c.netPacketConn.resultCh <- result
}
func (c *singPacketReadWaiter) Upstream() any {
return c.packetReadWaiter
}