mirror of
https://github.com/XTLS/Xray-core.git
synced 2026-01-13 14:17:09 +08:00
Compare commits
1 Commits
timer-redu
...
leak-fix
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e22ecdcb35 |
@@ -162,7 +162,6 @@ type udpConn struct {
|
||||
uplink stats.Counter
|
||||
downlink stats.Counter
|
||||
inactive bool
|
||||
cancel context.CancelFunc
|
||||
}
|
||||
|
||||
func (c *udpConn) setInactive() {
|
||||
@@ -205,9 +204,6 @@ func (c *udpConn) Write(buf []byte) (int, error) {
|
||||
}
|
||||
|
||||
func (c *udpConn) Close() error {
|
||||
if c.cancel != nil {
|
||||
c.cancel()
|
||||
}
|
||||
common.Must(c.done.Close())
|
||||
common.Must(common.Close(c.writer))
|
||||
return nil
|
||||
@@ -264,7 +260,6 @@ func (w *udpWorker) getConnection(id connID) (*udpConn, bool) {
|
||||
defer w.Unlock()
|
||||
|
||||
if conn, found := w.activeConn[id]; found && !conn.done.Done() {
|
||||
conn.updateActivity()
|
||||
return conn, true
|
||||
}
|
||||
|
||||
@@ -312,8 +307,7 @@ func (w *udpWorker) callback(b *buf.Buffer, source net.Destination, originalDest
|
||||
common.Must(w.checker.Start())
|
||||
|
||||
go func() {
|
||||
ctx, cancel := context.WithCancel(w.ctx)
|
||||
conn.cancel = cancel
|
||||
ctx := w.ctx
|
||||
sid := session.NewID()
|
||||
ctx = c.ContextWithID(ctx, sid)
|
||||
|
||||
|
||||
@@ -15,10 +15,9 @@ type ActivityUpdater interface {
|
||||
|
||||
type ActivityTimer struct {
|
||||
sync.RWMutex
|
||||
updated chan struct{}
|
||||
checkTask *task.Periodic
|
||||
onTimeout func()
|
||||
overridden bool
|
||||
updated chan struct{}
|
||||
checkTask *task.Periodic
|
||||
onTimeout func()
|
||||
}
|
||||
|
||||
func (t *ActivityTimer) Update() {
|
||||
@@ -32,16 +31,14 @@ func (t *ActivityTimer) check() error {
|
||||
select {
|
||||
case <-t.updated:
|
||||
default:
|
||||
t.finish(false)
|
||||
t.finish()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *ActivityTimer) finish(locked bool) {
|
||||
if !locked {
|
||||
t.Lock()
|
||||
defer t.Unlock()
|
||||
}
|
||||
func (t *ActivityTimer) finish() {
|
||||
t.Lock()
|
||||
defer t.Unlock()
|
||||
|
||||
if t.onTimeout != nil {
|
||||
t.onTimeout()
|
||||
@@ -53,12 +50,9 @@ func (t *ActivityTimer) finish(locked bool) {
|
||||
}
|
||||
}
|
||||
|
||||
func (t *ActivityTimer) setTimeout(timeout time.Duration) {
|
||||
if t.onTimeout == nil {
|
||||
return
|
||||
}
|
||||
func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
|
||||
if timeout == 0 {
|
||||
t.finish(true)
|
||||
t.finish()
|
||||
return
|
||||
}
|
||||
|
||||
@@ -67,26 +61,14 @@ func (t *ActivityTimer) setTimeout(timeout time.Duration) {
|
||||
Execute: t.check,
|
||||
}
|
||||
|
||||
t.Lock()
|
||||
|
||||
if t.checkTask != nil {
|
||||
t.checkTask.Close()
|
||||
t.overridden = true
|
||||
}
|
||||
t.checkTask = checkTask
|
||||
t.Update()
|
||||
common.Must(checkTask.Start())
|
||||
}
|
||||
|
||||
func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
|
||||
t.Lock()
|
||||
t.setTimeout(timeout)
|
||||
t.Unlock()
|
||||
}
|
||||
|
||||
func (t *ActivityTimer) SetTimeoutIfNotOverridden(timeout time.Duration) {
|
||||
t.Lock()
|
||||
if !t.overridden {
|
||||
t.setTimeout(timeout)
|
||||
}
|
||||
t.Unlock()
|
||||
}
|
||||
|
||||
|
||||
@@ -595,10 +595,10 @@ func CopyRawConnIfExist(ctx context.Context, readerConn net.Conn, writerConn net
|
||||
errors.LogInfo(ctx, "CopyRawConn splice")
|
||||
statWriter, _ := writer.(*dispatcher.SizeStatWriter)
|
||||
//runtime.Gosched() // necessary
|
||||
time.Sleep(time.Millisecond) // without this, there will be a rare ssl error for freedom splice
|
||||
timer.SetTimeoutIfNotOverridden(24 * time.Hour) // prevent leak, just in case
|
||||
time.Sleep(time.Millisecond) // without this, there will be a rare ssl error for freedom splice
|
||||
timer.SetTimeout(8 * time.Hour) // prevent leak, just in case
|
||||
if inTimer != nil {
|
||||
inTimer.SetTimeoutIfNotOverridden(24 * time.Hour)
|
||||
inTimer.SetTimeout(8 * time.Hour)
|
||||
}
|
||||
w, err := tc.ReadFrom(readerConn)
|
||||
if readCounter != nil {
|
||||
|
||||
@@ -128,7 +128,6 @@ func (s *Server) handleUDPPayload(ctx context.Context, conn stat.Connection, dis
|
||||
|
||||
conn.Write(data.Bytes())
|
||||
})
|
||||
defer udpServer.RemoveRay()
|
||||
|
||||
inbound := session.InboundFromContext(ctx)
|
||||
var dest *net.Destination
|
||||
|
||||
@@ -245,15 +245,13 @@ func (s *Server) handleUDPPayload(ctx context.Context, conn stat.Connection, dis
|
||||
udpMessage, err := EncodeUDPPacket(request, payload.Bytes())
|
||||
payload.Release()
|
||||
|
||||
defer udpMessage.Release()
|
||||
if err != nil {
|
||||
errors.LogWarningInner(ctx, err, "failed to write UDP response")
|
||||
return
|
||||
}
|
||||
defer udpMessage.Release()
|
||||
|
||||
conn.Write(udpMessage.Bytes())
|
||||
})
|
||||
defer udpServer.RemoveRay()
|
||||
|
||||
inbound := session.InboundFromContext(ctx)
|
||||
if inbound != nil && inbound.Source.IsValid() {
|
||||
|
||||
@@ -259,7 +259,6 @@ func (s *Server) handleUDPPayload(ctx context.Context, clientReader *PacketReade
|
||||
errors.LogWarningInner(ctx, err, "failed to write response")
|
||||
}
|
||||
})
|
||||
defer udpServer.RemoveRay()
|
||||
|
||||
inbound := session.InboundFromContext(ctx)
|
||||
user := inbound.User
|
||||
|
||||
@@ -206,10 +206,7 @@ func (h *Handler) Process(ctx context.Context, network net.Network, connection s
|
||||
|
||||
first := buf.FromBytes(make([]byte, buf.Size))
|
||||
first.Clear()
|
||||
firstLen, errR := first.ReadFrom(connection)
|
||||
if errR != nil {
|
||||
return errR
|
||||
}
|
||||
firstLen, _ := first.ReadFrom(connection)
|
||||
errors.LogInfo(ctx, "firstLen = ", firstLen)
|
||||
|
||||
reader := &buf.BufferedReader{
|
||||
|
||||
Reference in New Issue
Block a user