mirror of https://github.com/XTLS/Xray-core.git
Rename quic session to connection
Co-authored-by: 秋のかえで <autmaple@protonmail.com>
This commit is contained in:
yuhan6665
parent
c6550aecfc
commit
393d211d1e
|
|
@ -37,7 +37,7 @@ type QUICNameServer struct {
|
|||
reqID uint32
|
||||
name string
|
||||
destination *net.Destination
|
||||
session quic.Session
|
||||
connection quic.Connection
|
||||
}
|
||||
|
||||
// NewQUICNameServer creates DNS-over-QUIC client object for local resolving
|
||||
|
|
@ -194,7 +194,7 @@ func (s *QUICNameServer) sendQuery(ctx context.Context, domain string, clientIP
|
|||
|
||||
conn, err := s.openStream(dnsCtx)
|
||||
if err != nil {
|
||||
newError("failed to open quic session").Base(err).AtError().WriteToLog()
|
||||
newError("failed to open quic connection").Base(err).AtError().WriteToLog()
|
||||
return
|
||||
}
|
||||
|
||||
|
|
@ -322,7 +322,7 @@ func (s *QUICNameServer) QueryIP(ctx context.Context, domain string, clientIP ne
|
|||
}
|
||||
}
|
||||
|
||||
func isActive(s quic.Session) bool {
|
||||
func isActive(s quic.Connection) bool {
|
||||
select {
|
||||
case <-s.Context().Done():
|
||||
return false
|
||||
|
|
@ -331,17 +331,17 @@ func isActive(s quic.Session) bool {
|
|||
}
|
||||
}
|
||||
|
||||
func (s *QUICNameServer) getSession() (quic.Session, error) {
|
||||
var session quic.Session
|
||||
func (s *QUICNameServer) getConnection() (quic.Connection, error) {
|
||||
var conn quic.Connection
|
||||
s.RLock()
|
||||
session = s.session
|
||||
if session != nil && isActive(session) {
|
||||
conn = s.connection
|
||||
if conn != nil && isActive(conn) {
|
||||
s.RUnlock()
|
||||
return session, nil
|
||||
return conn, nil
|
||||
}
|
||||
if session != nil {
|
||||
// we're recreating the session, let's create a new one
|
||||
_ = session.CloseWithError(0, "")
|
||||
if conn != nil {
|
||||
// we're recreating the connection, let's create a new one
|
||||
_ = conn.CloseWithError(0, "")
|
||||
}
|
||||
s.RUnlock()
|
||||
|
||||
|
|
@ -349,42 +349,42 @@ func (s *QUICNameServer) getSession() (quic.Session, error) {
|
|||
defer s.Unlock()
|
||||
|
||||
var err error
|
||||
session, err = s.openSession()
|
||||
conn, err = s.openConnection()
|
||||
if err != nil {
|
||||
// This does not look too nice, but QUIC (or maybe quic-go)
|
||||
// doesn't seem stable enough.
|
||||
// Maybe retransmissions aren't fully implemented in quic-go?
|
||||
// Anyways, the simple solution is to make a second try when
|
||||
// it fails to open the QUIC session.
|
||||
session, err = s.openSession()
|
||||
// it fails to open the QUIC connection.
|
||||
conn, err = s.openConnection()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
s.session = session
|
||||
return session, nil
|
||||
s.connection = conn
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (s *QUICNameServer) openSession() (quic.Session, error) {
|
||||
func (s *QUICNameServer) openConnection() (quic.Connection, error) {
|
||||
tlsConfig := tls.Config{}
|
||||
quicConfig := &quic.Config{
|
||||
HandshakeIdleTimeout: handshakeTimeout,
|
||||
}
|
||||
|
||||
session, err := quic.DialAddrContext(context.Background(), s.destination.NetAddr(), tlsConfig.GetTLSConfig(tls.WithNextProto("http/1.1", http2.NextProtoTLS, NextProtoDQ)), quicConfig)
|
||||
conn, err := quic.DialAddrContext(context.Background(), s.destination.NetAddr(), tlsConfig.GetTLSConfig(tls.WithNextProto("http/1.1", http2.NextProtoTLS, NextProtoDQ)), quicConfig)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return session, nil
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (s *QUICNameServer) openStream(ctx context.Context) (quic.Stream, error) {
|
||||
session, err := s.getSession()
|
||||
conn, err := s.getConnection()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// open a new stream
|
||||
return session.OpenStreamSync(ctx)
|
||||
return conn.OpenStreamSync(ctx)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -15,39 +15,39 @@ import (
|
|||
"github.com/xtls/xray-core/transport/internet/tls"
|
||||
)
|
||||
|
||||
type sessionContext struct {
|
||||
type connectionContext struct {
|
||||
rawConn *sysConn
|
||||
session quic.Session
|
||||
conn quic.Connection
|
||||
}
|
||||
|
||||
var errSessionClosed = newError("session closed")
|
||||
var errConnectionClosed = newError("connection closed")
|
||||
|
||||
func (c *sessionContext) openStream(destAddr net.Addr) (*interConn, error) {
|
||||
if !isActive(c.session) {
|
||||
return nil, errSessionClosed
|
||||
func (c *connectionContext) openStream(destAddr net.Addr) (*interConn, error) {
|
||||
if !isActive(c.conn) {
|
||||
return nil, errConnectionClosed
|
||||
}
|
||||
|
||||
stream, err := c.session.OpenStream()
|
||||
stream, err := c.conn.OpenStream()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
conn := &interConn{
|
||||
stream: stream,
|
||||
local: c.session.LocalAddr(),
|
||||
local: c.conn.LocalAddr(),
|
||||
remote: destAddr,
|
||||
}
|
||||
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
type clientSessions struct {
|
||||
access sync.Mutex
|
||||
sessions map[net.Destination][]*sessionContext
|
||||
cleanup *task.Periodic
|
||||
type clientConnections struct {
|
||||
access sync.Mutex
|
||||
conns map[net.Destination][]*connectionContext
|
||||
cleanup *task.Periodic
|
||||
}
|
||||
|
||||
func isActive(s quic.Session) bool {
|
||||
func isActive(s quic.Connection) bool {
|
||||
select {
|
||||
case <-s.Context().Done():
|
||||
return false
|
||||
|
|
@ -56,81 +56,81 @@ func isActive(s quic.Session) bool {
|
|||
}
|
||||
}
|
||||
|
||||
func removeInactiveSessions(sessions []*sessionContext) []*sessionContext {
|
||||
activeSessions := make([]*sessionContext, 0, len(sessions))
|
||||
for i, s := range sessions {
|
||||
if isActive(s.session) {
|
||||
activeSessions = append(activeSessions, s)
|
||||
func removeInactiveConnections(conns []*connectionContext) []*connectionContext {
|
||||
activeConnections := make([]*connectionContext, 0, len(conns))
|
||||
for i, s := range conns {
|
||||
if isActive(s.conn) {
|
||||
activeConnections = append(activeConnections, s)
|
||||
continue
|
||||
}
|
||||
|
||||
newError("closing quic session at index: ", i).WriteToLog()
|
||||
if err := s.session.CloseWithError(0, ""); err != nil {
|
||||
newError("failed to close session").Base(err).WriteToLog()
|
||||
newError("closing quic connection at index: ", i).WriteToLog()
|
||||
if err := s.conn.CloseWithError(0, ""); err != nil {
|
||||
newError("failed to close connection").Base(err).WriteToLog()
|
||||
}
|
||||
if err := s.rawConn.Close(); err != nil {
|
||||
newError("failed to close raw connection").Base(err).WriteToLog()
|
||||
}
|
||||
}
|
||||
|
||||
if len(activeSessions) < len(sessions) {
|
||||
newError("active quic session reduced from ", len(sessions), " to ", len(activeSessions)).WriteToLog()
|
||||
return activeSessions
|
||||
if len(activeConnections) < len(conns) {
|
||||
newError("active quic connection reduced from ", len(conns), " to ", len(activeConnections)).WriteToLog()
|
||||
return activeConnections
|
||||
}
|
||||
|
||||
return sessions
|
||||
return conns
|
||||
}
|
||||
|
||||
func (s *clientSessions) cleanSessions() error {
|
||||
func (s *clientConnections) cleanConnections() error {
|
||||
s.access.Lock()
|
||||
defer s.access.Unlock()
|
||||
|
||||
if len(s.sessions) == 0 {
|
||||
if len(s.conns) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
newSessionMap := make(map[net.Destination][]*sessionContext)
|
||||
newConnMap := make(map[net.Destination][]*connectionContext)
|
||||
|
||||
for dest, sessions := range s.sessions {
|
||||
sessions = removeInactiveSessions(sessions)
|
||||
if len(sessions) > 0 {
|
||||
newSessionMap[dest] = sessions
|
||||
for dest, conns := range s.conns {
|
||||
conns = removeInactiveConnections(conns)
|
||||
if len(conns) > 0 {
|
||||
newConnMap[dest] = conns
|
||||
}
|
||||
}
|
||||
|
||||
s.sessions = newSessionMap
|
||||
s.conns = newConnMap
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *clientSessions) openConnection(ctx context.Context, destAddr net.Addr, config *Config, tlsConfig *tls.Config, sockopt *internet.SocketConfig) (stat.Connection, error) {
|
||||
func (s *clientConnections) openConnection(ctx context.Context, destAddr net.Addr, config *Config, tlsConfig *tls.Config, sockopt *internet.SocketConfig) (stat.Connection, error) {
|
||||
s.access.Lock()
|
||||
defer s.access.Unlock()
|
||||
|
||||
if s.sessions == nil {
|
||||
s.sessions = make(map[net.Destination][]*sessionContext)
|
||||
if s.conns == nil {
|
||||
s.conns = make(map[net.Destination][]*connectionContext)
|
||||
}
|
||||
|
||||
dest := net.DestinationFromAddr(destAddr)
|
||||
|
||||
var sessions []*sessionContext
|
||||
if s, found := s.sessions[dest]; found {
|
||||
sessions = s
|
||||
var conns []*connectionContext
|
||||
if s, found := s.conns[dest]; found {
|
||||
conns = s
|
||||
}
|
||||
|
||||
if len(sessions) > 0 {
|
||||
s := sessions[len(sessions)-1]
|
||||
if isActive(s.session) {
|
||||
if len(conns) > 0 {
|
||||
s := conns[len(conns)-1]
|
||||
if isActive(s.conn) {
|
||||
conn, err := s.openStream(destAddr)
|
||||
if err == nil {
|
||||
return conn, nil
|
||||
}
|
||||
newError("failed to openStream: ").Base(err).WriteToLog()
|
||||
} else {
|
||||
newError("current quic session is not active!").WriteToLog()
|
||||
newError("current quic connection is not active!").WriteToLog()
|
||||
}
|
||||
}
|
||||
|
||||
sessions = removeInactiveSessions(sessions)
|
||||
conns = removeInactiveConnections(conns)
|
||||
newError("dialing quic to ", dest).WriteToLog()
|
||||
rawConn, err := internet.DialSystem(ctx, dest, sockopt)
|
||||
if err != nil {
|
||||
|
|
@ -146,33 +146,33 @@ func (s *clientSessions) openConnection(ctx context.Context, destAddr net.Addr,
|
|||
if udpConn == nil {
|
||||
udpConn = rawConn.(*internet.PacketConnWrapper).Conn.(*net.UDPConn)
|
||||
}
|
||||
conn, err := wrapSysConn(udpConn, config)
|
||||
sysConn, err := wrapSysConn(udpConn, config)
|
||||
if err != nil {
|
||||
rawConn.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
session, err := quic.DialContext(context.Background(), conn, destAddr, "", tlsConfig.GetTLSConfig(tls.WithDestination(dest)), quicConfig)
|
||||
conn, err := quic.DialContext(context.Background(), sysConn, destAddr, "", tlsConfig.GetTLSConfig(tls.WithDestination(dest)), quicConfig)
|
||||
if err != nil {
|
||||
conn.Close()
|
||||
sysConn.Close()
|
||||
return nil, err
|
||||
}
|
||||
|
||||
context := &sessionContext{
|
||||
session: session,
|
||||
rawConn: conn,
|
||||
context := &connectionContext{
|
||||
conn: conn,
|
||||
rawConn: sysConn,
|
||||
}
|
||||
s.sessions[dest] = append(sessions, context)
|
||||
s.conns[dest] = append(conns, context)
|
||||
return context.openStream(destAddr)
|
||||
}
|
||||
|
||||
var client clientSessions
|
||||
var client clientConnections
|
||||
|
||||
func init() {
|
||||
client.sessions = make(map[net.Destination][]*sessionContext)
|
||||
client.conns = make(map[net.Destination][]*connectionContext)
|
||||
client.cleanup = &task.Periodic{
|
||||
Interval: time.Minute,
|
||||
Execute: client.cleanSessions,
|
||||
Execute: client.cleanConnections,
|
||||
}
|
||||
common.Must(client.cleanup.Start())
|
||||
}
|
||||
|
|
|
|||
|
|
@ -22,17 +22,17 @@ type Listener struct {
|
|||
addConn internet.ConnHandler
|
||||
}
|
||||
|
||||
func (l *Listener) acceptStreams(session quic.Session) {
|
||||
func (l *Listener) acceptStreams(conn quic.Connection) {
|
||||
for {
|
||||
stream, err := session.AcceptStream(context.Background())
|
||||
stream, err := conn.AcceptStream(context.Background())
|
||||
if err != nil {
|
||||
newError("failed to accept stream").Base(err).WriteToLog()
|
||||
select {
|
||||
case <-session.Context().Done():
|
||||
case <-conn.Context().Done():
|
||||
return
|
||||
case <-l.done.Wait():
|
||||
if err := session.CloseWithError(0, ""); err != nil {
|
||||
newError("failed to close session").Base(err).WriteToLog()
|
||||
if err := conn.CloseWithError(0, ""); err != nil {
|
||||
newError("failed to close connection").Base(err).WriteToLog()
|
||||
}
|
||||
return
|
||||
default:
|
||||
|
|
@ -43,8 +43,8 @@ func (l *Listener) acceptStreams(session quic.Session) {
|
|||
|
||||
conn := &interConn{
|
||||
stream: stream,
|
||||
local: session.LocalAddr(),
|
||||
remote: session.RemoteAddr(),
|
||||
local: conn.LocalAddr(),
|
||||
remote: conn.RemoteAddr(),
|
||||
}
|
||||
|
||||
l.addConn(conn)
|
||||
|
|
@ -55,7 +55,7 @@ func (l *Listener) keepAccepting() {
|
|||
for {
|
||||
conn, err := l.listener.Accept(context.Background())
|
||||
if err != nil {
|
||||
newError("failed to accept QUIC sessions").Base(err).WriteToLog()
|
||||
newError("failed to accept QUIC connection").Base(err).WriteToLog()
|
||||
if l.done.Done() {
|
||||
break
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue