diff --git a/turbotunnel/clientmap.go b/turbotunnel/clientmap.go deleted file mode 100644 index 5f62cb6..0000000 --- a/turbotunnel/clientmap.go +++ /dev/null @@ -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 -} diff --git a/turbotunnel/queuepacketconn.go b/turbotunnel/queuepacketconn.go index 14a9833..98693b1 100644 --- a/turbotunnel/queuepacketconn.go +++ b/turbotunnel/queuepacketconn.go @@ -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. diff --git a/turbotunnel/remotemap.go b/turbotunnel/remotemap.go new file mode 100644 index 0000000..7e6dd79 --- /dev/null +++ b/turbotunnel/remotemap.go @@ -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 +}