Update cancel context usage

This commit is contained in:
世界 2023-04-10 13:00:57 +08:00
parent f7f9a7ae20
commit bb63429079
No known key found for this signature in database
GPG key ID: CD109927C34A63C4
5 changed files with 23 additions and 12 deletions

View file

@ -2,6 +2,7 @@ package ntp
import ( import (
"context" "context"
"os"
"time" "time"
"github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/adapter"
@ -9,6 +10,7 @@ import (
"github.com/sagernet/sing-box/common/settings" "github.com/sagernet/sing-box/common/settings"
C "github.com/sagernet/sing-box/constant" C "github.com/sagernet/sing-box/constant"
"github.com/sagernet/sing-box/option" "github.com/sagernet/sing-box/option"
"github.com/sagernet/sing/common"
E "github.com/sagernet/sing/common/exceptions" E "github.com/sagernet/sing/common/exceptions"
"github.com/sagernet/sing/common/logger" "github.com/sagernet/sing/common/logger"
M "github.com/sagernet/sing/common/metadata" M "github.com/sagernet/sing/common/metadata"
@ -20,7 +22,7 @@ var _ adapter.TimeService = (*Service)(nil)
type Service struct { type Service struct {
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel common.ContextCancelCauseFunc
server M.Socksaddr server M.Socksaddr
writeToSystem bool writeToSystem bool
dialer N.Dialer dialer N.Dialer
@ -30,7 +32,7 @@ type Service struct {
} }
func NewService(ctx context.Context, router adapter.Router, logger logger.Logger, options option.NTPOptions) *Service { func NewService(ctx context.Context, router adapter.Router, logger logger.Logger, options option.NTPOptions) *Service {
ctx, cancel := context.WithCancel(ctx) ctx, cancel := common.ContextWithCancelCause(ctx)
server := options.ServerOptions.Build() server := options.ServerOptions.Build()
if server.Port == 0 { if server.Port == 0 {
server.Port = 123 server.Port = 123
@ -64,7 +66,7 @@ func (s *Service) Start() error {
func (s *Service) Close() error { func (s *Service) Close() error {
s.ticker.Stop() s.ticker.Stop()
s.cancel() s.cancel(os.ErrClosed)
return nil return nil
} }

View file

@ -102,11 +102,10 @@ func (d *DNS) handleConnection(ctx context.Context, conn net.Conn, metadata adap
func (d *DNS) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error { func (d *DNS) NewPacketConnection(ctx context.Context, conn N.PacketConn, metadata adapter.InboundContext) error {
ctx = adapter.WithContext(ctx, &metadata) ctx = adapter.WithContext(ctx, &metadata)
fastClose, cancel := context.WithCancel(ctx) fastClose, cancel := common.ContextWithCancelCause(ctx)
timeout := canceler.New(fastClose, cancel, C.DNSTimeout) timeout := canceler.New(fastClose, cancel, C.DNSTimeout)
var group task.Group var group task.Group
group.Append0(func(ctx context.Context) error { group.Append0(func(ctx context.Context) error {
defer cancel()
_buffer := buf.StackNewSize(dns.FixedPacketSize) _buffer := buf.StackNewSize(dns.FixedPacketSize)
defer common.KeepAlive(_buffer) defer common.KeepAlive(_buffer)
buffer := common.Dup(_buffer) buffer := common.Dup(_buffer)
@ -115,11 +114,13 @@ func (d *DNS) NewPacketConnection(ctx context.Context, conn N.PacketConn, metada
buffer.FullReset() buffer.FullReset()
destination, err := conn.ReadPacket(buffer) destination, err := conn.ReadPacket(buffer)
if err != nil { if err != nil {
cancel(err)
return err return err
} }
var message mDNS.Msg var message mDNS.Msg
err = message.Unpack(buffer.Bytes()) err = message.Unpack(buffer.Bytes())
if err != nil { if err != nil {
cancel(err)
return err return err
} }
timeout.Update() timeout.Update()
@ -127,17 +128,22 @@ func (d *DNS) NewPacketConnection(ctx context.Context, conn N.PacketConn, metada
go func() error { go func() error {
response, err := d.router.Exchange(adapter.WithContext(ctx, &metadataInQuery), &message) response, err := d.router.Exchange(adapter.WithContext(ctx, &metadataInQuery), &message)
if err != nil { if err != nil {
cancel(err)
return err return err
} }
timeout.Update() timeout.Update()
responseBuffer := buf.NewPacket() responseBuffer := buf.NewPacket()
n, err := response.PackBuffer(responseBuffer.FreeBytes()) n, err := response.PackBuffer(responseBuffer.FreeBytes())
if err != nil { if err != nil {
cancel(err)
responseBuffer.Release() responseBuffer.Release()
return err return err
} }
responseBuffer.Truncate(len(n)) responseBuffer.Truncate(len(n))
err = conn.WritePacket(responseBuffer, destination) err = conn.WritePacket(responseBuffer, destination)
if err != nil {
cancel(err)
}
return err return err
}() }()
} }

View file

@ -101,10 +101,10 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) {
return nil, err return nil, err
} }
client := NewGunServiceClient(clientConn).(GunServiceCustomNameClient) client := NewGunServiceClient(clientConn).(GunServiceCustomNameClient)
ctx, cancel := context.WithCancel(ctx) ctx, cancel := common.ContextWithCancelCause(ctx)
stream, err := client.TunCustomName(ctx, c.serviceName) stream, err := client.TunCustomName(ctx, c.serviceName)
if err != nil { if err != nil {
cancel() cancel(err)
return nil, err return nil, err
} }
return NewGRPCConn(stream, cancel), nil return NewGRPCConn(stream, cancel), nil

View file

@ -1,12 +1,12 @@
package v2raygrpc package v2raygrpc
import ( import (
"context"
"net" "net"
"os" "os"
"time" "time"
"github.com/sagernet/sing-box/common/baderror" "github.com/sagernet/sing-box/common/baderror"
"github.com/sagernet/sing/common"
"github.com/sagernet/sing/common/rw" "github.com/sagernet/sing/common/rw"
) )
@ -14,11 +14,11 @@ var _ net.Conn = (*GRPCConn)(nil)
type GRPCConn struct { type GRPCConn struct {
GunService GunService
cancel context.CancelFunc cancel common.ContextCancelCauseFunc
cache []byte cache []byte
} }
func NewGRPCConn(service GunService, cancel context.CancelFunc) *GRPCConn { func NewGRPCConn(service GunService, cancel common.ContextCancelCauseFunc) *GRPCConn {
if client, isClient := service.(GunService_TunClient); isClient { if client, isClient := service.(GunService_TunClient); isClient {
service = &clientConnWrapper{client} service = &clientConnWrapper{client}
} }
@ -37,6 +37,7 @@ func (c *GRPCConn) Read(b []byte) (n int, err error) {
hunk, err := c.Recv() hunk, err := c.Recv()
err = baderror.WrapGRPC(err) err = baderror.WrapGRPC(err)
if err != nil { if err != nil {
c.cancel(err)
return return
} }
n = copy(b, hunk.Data) n = copy(b, hunk.Data)
@ -49,13 +50,14 @@ func (c *GRPCConn) Read(b []byte) (n int, err error) {
func (c *GRPCConn) Write(b []byte) (n int, err error) { func (c *GRPCConn) Write(b []byte) (n int, err error) {
err = baderror.WrapGRPC(c.Send(&Hunk{Data: b})) err = baderror.WrapGRPC(c.Send(&Hunk{Data: b}))
if err != nil { if err != nil {
c.cancel(err)
return return
} }
return len(b), nil return len(b), nil
} }
func (c *GRPCConn) Close() error { func (c *GRPCConn) Close() error {
c.cancel() c.cancel(net.ErrClosed)
return nil return nil
} }

View file

@ -10,6 +10,7 @@ import (
"github.com/sagernet/sing-box/adapter" "github.com/sagernet/sing-box/adapter"
"github.com/sagernet/sing-box/common/tls" "github.com/sagernet/sing-box/common/tls"
"github.com/sagernet/sing-box/option" "github.com/sagernet/sing-box/option"
"github.com/sagernet/sing/common"
M "github.com/sagernet/sing/common/metadata" M "github.com/sagernet/sing/common/metadata"
N "github.com/sagernet/sing/common/network" N "github.com/sagernet/sing/common/network"
@ -45,7 +46,7 @@ func NewServer(ctx context.Context, options option.V2RayGRPCOptions, tlsConfig t
} }
func (s *Server) Tun(server GunService_TunServer) error { func (s *Server) Tun(server GunService_TunServer) error {
ctx, cancel := context.WithCancel(s.ctx) ctx, cancel := common.ContextWithCancelCause(s.ctx)
conn := NewGRPCConn(server, cancel) conn := NewGRPCConn(server, cancel)
var metadata M.Metadata var metadata M.Metadata
if remotePeer, loaded := peer.FromContext(server.Context()); loaded { if remotePeer, loaded := peer.FromContext(server.Context()); loaded {