mirror of
https://github.com/tladesignz/dnstt.git
synced 2026-10-08 14:48:14 +03:00
Rename ClientMap to RemoteMap.
We're now using it (via QueuePacketConn) in DNSPacketConn, HTTPPacketConn, and TLSPacketConn, where there remote peer is actually a server, not a client.
This commit is contained in:
David Fifield
@@ -1,149 +0,0 @@
|
||||
package turbotunnel
|
||||
|
||||
import (
|
||||
"container/heap"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// clientRecord is a record of a recently seen client, with the time it was last
|
||||
// seen and a send queue.
|
||||
type clientRecord struct {
|
||||
Addr net.Addr
|
||||
LastSeen time.Time
|
||||
SendQueue chan []byte
|
||||
}
|
||||
|
||||
// ClientMap manages a mapping of live clients (keyed by address, which will be
|
||||
// a ClientID) to their respective send queues. ClientMap's functions are safe
|
||||
// to call from multiple goroutines.
|
||||
type ClientMap struct {
|
||||
// We use an inner structure to avoid exposing public heap.Interface
|
||||
// functions to users of clientMap.
|
||||
inner clientMapInner
|
||||
// Synchronizes access to inner.
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
// NewClientMap creates a ClientMap that expires clients after a timeout.
|
||||
//
|
||||
// If the timeout is 0, clients never expire.
|
||||
//
|
||||
// The timeout does not have to be kept in sync with smux's idle timeout. If a
|
||||
// client is removed from the client map while the smux session is still live,
|
||||
// the worst that can happen is a loss of whatever packets were in the send
|
||||
// queue at the time. If smux later decides to send more packets to the same
|
||||
// client, we'll instantiate a new send queue, and if the client ever connects
|
||||
// again with the proper client ID, we'll deliver them.
|
||||
func NewClientMap(timeout time.Duration) *ClientMap {
|
||||
m := &ClientMap{
|
||||
inner: clientMapInner{
|
||||
byAge: make([]*clientRecord, 0),
|
||||
byAddr: make(map[net.Addr]int),
|
||||
},
|
||||
}
|
||||
if timeout > 0 {
|
||||
go func() {
|
||||
for {
|
||||
time.Sleep(timeout / 2)
|
||||
now := time.Now()
|
||||
m.lock.Lock()
|
||||
m.inner.removeExpired(now, timeout)
|
||||
m.lock.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// SendQueue returns the send queue corresponding to addr, creating it if
|
||||
// necessary.
|
||||
func (m *ClientMap) SendQueue(addr net.Addr) chan []byte {
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
return m.inner.SendQueue(addr, time.Now())
|
||||
}
|
||||
|
||||
// clientMapInner is the inner type of ClientMap, implementing heap.Interface.
|
||||
// byAge is the backing store, a heap ordered by LastSeen time, to facilitate
|
||||
// expiring old client records. byAddr is a map from addresses (i.e., ClientIDs)
|
||||
// to heap indices, to allow looking up by address. Unlike ClientMap,
|
||||
// clientMapInner requires external synchonization.
|
||||
type clientMapInner struct {
|
||||
byAge []*clientRecord
|
||||
byAddr map[net.Addr]int
|
||||
}
|
||||
|
||||
// removeExpired removes all client records whose LastSeen timestamp is more
|
||||
// than timeout in the past.
|
||||
func (inner *clientMapInner) removeExpired(now time.Time, timeout time.Duration) {
|
||||
for len(inner.byAge) > 0 && now.Sub(inner.byAge[0].LastSeen) >= timeout {
|
||||
record := heap.Pop(inner).(*clientRecord)
|
||||
close(record.SendQueue)
|
||||
}
|
||||
}
|
||||
|
||||
// SendQueue finds the existing client record corresponding to addr, or creates
|
||||
// a new one if none exists yet. It updates the client record's LastSeen time
|
||||
// and returns its SendQueue.
|
||||
func (inner *clientMapInner) SendQueue(addr net.Addr, now time.Time) chan []byte {
|
||||
var record *clientRecord
|
||||
i, ok := inner.byAddr[addr]
|
||||
if ok {
|
||||
// Found one, update its LastSeen.
|
||||
record = inner.byAge[i]
|
||||
record.LastSeen = now
|
||||
heap.Fix(inner, i)
|
||||
} else {
|
||||
// Not found, create a new one.
|
||||
record = &clientRecord{
|
||||
Addr: addr,
|
||||
LastSeen: now,
|
||||
SendQueue: make(chan []byte, queueSize),
|
||||
}
|
||||
heap.Push(inner, record)
|
||||
}
|
||||
return record.SendQueue
|
||||
}
|
||||
|
||||
// heap.Interface for clientMapInner.
|
||||
|
||||
func (inner *clientMapInner) Len() int {
|
||||
if len(inner.byAge) != len(inner.byAddr) {
|
||||
panic("inconsistent clientMap")
|
||||
}
|
||||
return len(inner.byAge)
|
||||
}
|
||||
|
||||
func (inner *clientMapInner) Less(i, j int) bool {
|
||||
return inner.byAge[i].LastSeen.Before(inner.byAge[j].LastSeen)
|
||||
}
|
||||
|
||||
func (inner *clientMapInner) Swap(i, j int) {
|
||||
inner.byAge[i], inner.byAge[j] = inner.byAge[j], inner.byAge[i]
|
||||
inner.byAddr[inner.byAge[i].Addr] = i
|
||||
inner.byAddr[inner.byAge[j].Addr] = j
|
||||
}
|
||||
|
||||
func (inner *clientMapInner) Push(x interface{}) {
|
||||
record := x.(*clientRecord)
|
||||
if _, ok := inner.byAddr[record.Addr]; ok {
|
||||
panic("duplicate address in clientMap")
|
||||
}
|
||||
// Insert into byAddr map.
|
||||
inner.byAddr[record.Addr] = len(inner.byAge)
|
||||
// Insert into byAge slice.
|
||||
inner.byAge = append(inner.byAge, record)
|
||||
}
|
||||
|
||||
func (inner *clientMapInner) Pop() interface{} {
|
||||
n := len(inner.byAddr)
|
||||
// Remove from byAge slice.
|
||||
record := inner.byAge[n-1]
|
||||
inner.byAge[n-1] = nil
|
||||
inner.byAge = inner.byAge[:n-1]
|
||||
// Remove from byAddr map.
|
||||
delete(inner.byAddr, record.Addr)
|
||||
return record
|
||||
}
|
||||
@@ -16,13 +16,13 @@ type taggedPacket struct {
|
||||
|
||||
// QueuePacketConn implements net.PacketConn by storing queues of packets. There
|
||||
// is one incoming queue (where packets are additionally tagged by the source
|
||||
// address of the client that sent them). There are many outgoing queues, one
|
||||
// for each client address that has been recently seen. The QueueIncoming method
|
||||
// inserts a packet into the incoming queue, to eventually be returned by
|
||||
// address of the peer that sent them). There are many outgoing queues, one for
|
||||
// each remote peer address that has been recently seen. The QueueIncoming
|
||||
// method inserts a packet into the incoming queue, to eventually be returned by
|
||||
// ReadFrom. WriteTo inserts a packet into an address-specific outgoing queue,
|
||||
// which can later by accessed through the OutgoingQueue method.
|
||||
type QueuePacketConn struct {
|
||||
clients *ClientMap
|
||||
remotes *RemoteMap
|
||||
localAddr net.Addr
|
||||
recvQueue chan taggedPacket
|
||||
closeOnce sync.Once
|
||||
@@ -31,11 +31,11 @@ type QueuePacketConn struct {
|
||||
err atomic.Value
|
||||
}
|
||||
|
||||
// NewQueuePacketConn makes a new QueuePacketConn, set to track recent clients
|
||||
// NewQueuePacketConn makes a new QueuePacketConn, set to track recent peers
|
||||
// for at least a duration of timeout.
|
||||
func NewQueuePacketConn(localAddr net.Addr, timeout time.Duration) *QueuePacketConn {
|
||||
return &QueuePacketConn{
|
||||
clients: NewClientMap(timeout),
|
||||
remotes: NewRemoteMap(timeout),
|
||||
localAddr: localAddr,
|
||||
recvQueue: make(chan taggedPacket, queueSize),
|
||||
closed: make(chan struct{}),
|
||||
@@ -65,7 +65,7 @@ func (c *QueuePacketConn) QueueIncoming(p []byte, addr net.Addr) {
|
||||
// creating it if necessary. The contents of the queue will be packets that are
|
||||
// written to the address in question using WriteTo.
|
||||
func (c *QueuePacketConn) OutgoingQueue(addr net.Addr) <-chan []byte {
|
||||
return c.clients.SendQueue(addr)
|
||||
return c.remotes.SendQueue(addr)
|
||||
}
|
||||
|
||||
// ReadFrom returns a packet and address previously stored by QueueIncoming.
|
||||
@@ -95,7 +95,7 @@ func (c *QueuePacketConn) WriteTo(p []byte, addr net.Addr) (int, error) {
|
||||
buf := make([]byte, len(p))
|
||||
copy(buf, p)
|
||||
select {
|
||||
case c.clients.SendQueue(addr) <- buf:
|
||||
case c.remotes.SendQueue(addr) <- buf:
|
||||
return len(buf), nil
|
||||
default:
|
||||
// Drop the outgoing packet if the send queue is full.
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
package turbotunnel
|
||||
|
||||
import (
|
||||
"container/heap"
|
||||
"net"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// remoteRecord is a record of a recently seen remote peer, with the time it was
|
||||
// last seen and a send queue.
|
||||
type remoteRecord struct {
|
||||
Addr net.Addr
|
||||
LastSeen time.Time
|
||||
SendQueue chan []byte
|
||||
}
|
||||
|
||||
// RemoteMap manages a mapping of live remote peers, keyed by address, to their
|
||||
// respective send queues. RemoteMap's functions are safe to call from multiple
|
||||
// goroutines.
|
||||
type RemoteMap struct {
|
||||
// We use an inner structure to avoid exposing public heap.Interface
|
||||
// functions to users of remoteMap.
|
||||
inner remoteMapInner
|
||||
// Synchronizes access to inner.
|
||||
lock sync.Mutex
|
||||
}
|
||||
|
||||
// NewRemoteMap creates a RemoteMap that expires peers after a timeout.
|
||||
//
|
||||
// If the timeout is 0, peers never expire.
|
||||
//
|
||||
// The timeout does not have to be kept in sync with smux's idle timeout. If a
|
||||
// peer is removed from the map while the smux session is still live, the worst
|
||||
// that can happen is a loss of whatever packets were in the send queue at the
|
||||
// time. If smux later decides to send more packets to the same peer, we'll
|
||||
// instantiate a new send queue, and if the peer is ever seen again with a
|
||||
// matching address, we'll deliver them.
|
||||
func NewRemoteMap(timeout time.Duration) *RemoteMap {
|
||||
m := &RemoteMap{
|
||||
inner: remoteMapInner{
|
||||
byAge: make([]*remoteRecord, 0),
|
||||
byAddr: make(map[net.Addr]int),
|
||||
},
|
||||
}
|
||||
if timeout > 0 {
|
||||
go func() {
|
||||
for {
|
||||
time.Sleep(timeout / 2)
|
||||
now := time.Now()
|
||||
m.lock.Lock()
|
||||
m.inner.removeExpired(now, timeout)
|
||||
m.lock.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// SendQueue returns the send queue corresponding to addr, creating it if
|
||||
// necessary.
|
||||
func (m *RemoteMap) SendQueue(addr net.Addr) chan []byte {
|
||||
m.lock.Lock()
|
||||
defer m.lock.Unlock()
|
||||
return m.inner.SendQueue(addr, time.Now())
|
||||
}
|
||||
|
||||
// remoteMapInner is the inner type of RemoteMap, implementing heap.Interface.
|
||||
// byAge is the backing store, a heap ordered by LastSeen time, to facilitate
|
||||
// expiring old records. byAddr is a map from addresses to heap indices, to
|
||||
// allow looking up by address. Unlike RemoteMap, remoteMapInner requires
|
||||
// external synchonization.
|
||||
type remoteMapInner struct {
|
||||
byAge []*remoteRecord
|
||||
byAddr map[net.Addr]int
|
||||
}
|
||||
|
||||
// removeExpired removes all records whose LastSeen timestamp is more than
|
||||
// timeout in the past.
|
||||
func (inner *remoteMapInner) removeExpired(now time.Time, timeout time.Duration) {
|
||||
for len(inner.byAge) > 0 && now.Sub(inner.byAge[0].LastSeen) >= timeout {
|
||||
record := heap.Pop(inner).(*remoteRecord)
|
||||
close(record.SendQueue)
|
||||
}
|
||||
}
|
||||
|
||||
// SendQueue finds the existing record corresponding to addr, or creates a new
|
||||
// one if none exists yet. It updates the record's LastSeen time and returns its
|
||||
// SendQueue.
|
||||
func (inner *remoteMapInner) SendQueue(addr net.Addr, now time.Time) chan []byte {
|
||||
var record *remoteRecord
|
||||
i, ok := inner.byAddr[addr]
|
||||
if ok {
|
||||
// Found one, update its LastSeen.
|
||||
record = inner.byAge[i]
|
||||
record.LastSeen = now
|
||||
heap.Fix(inner, i)
|
||||
} else {
|
||||
// Not found, create a new one.
|
||||
record = &remoteRecord{
|
||||
Addr: addr,
|
||||
LastSeen: now,
|
||||
SendQueue: make(chan []byte, queueSize),
|
||||
}
|
||||
heap.Push(inner, record)
|
||||
}
|
||||
return record.SendQueue
|
||||
}
|
||||
|
||||
// heap.Interface for remoteMapInner.
|
||||
|
||||
func (inner *remoteMapInner) Len() int {
|
||||
if len(inner.byAge) != len(inner.byAddr) {
|
||||
panic("inconsistent remoteMap")
|
||||
}
|
||||
return len(inner.byAge)
|
||||
}
|
||||
|
||||
func (inner *remoteMapInner) Less(i, j int) bool {
|
||||
return inner.byAge[i].LastSeen.Before(inner.byAge[j].LastSeen)
|
||||
}
|
||||
|
||||
func (inner *remoteMapInner) Swap(i, j int) {
|
||||
inner.byAge[i], inner.byAge[j] = inner.byAge[j], inner.byAge[i]
|
||||
inner.byAddr[inner.byAge[i].Addr] = i
|
||||
inner.byAddr[inner.byAge[j].Addr] = j
|
||||
}
|
||||
|
||||
func (inner *remoteMapInner) Push(x interface{}) {
|
||||
record := x.(*remoteRecord)
|
||||
if _, ok := inner.byAddr[record.Addr]; ok {
|
||||
panic("duplicate address in remoteMap")
|
||||
}
|
||||
// Insert into byAddr map.
|
||||
inner.byAddr[record.Addr] = len(inner.byAge)
|
||||
// Insert into byAge slice.
|
||||
inner.byAge = append(inner.byAge, record)
|
||||
}
|
||||
|
||||
func (inner *remoteMapInner) Pop() interface{} {
|
||||
n := len(inner.byAddr)
|
||||
// Remove from byAge slice.
|
||||
record := inner.byAge[n-1]
|
||||
inner.byAge[n-1] = nil
|
||||
inner.byAge = inner.byAge[:n-1]
|
||||
// Remove from byAddr map.
|
||||
delete(inner.byAddr, record.Addr)
|
||||
return record
|
||||
}
|
||||
Reference in New Issue
Block a user