-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #34 from PretendoNetwork/stability
Stability - Better concurrent map API and deferred packet handling
- Loading branch information
Showing
13 changed files
with
608 additions
and
167 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,76 @@ | ||
| package nex | ||
|
|
||
| import "sync" | ||
|
|
||
| // MutexMap implements a map type with go routine safe accessors through mutex locks. Embeds sync.RWMutex | ||
| type MutexMap[K comparable, V any] struct { | ||
| *sync.RWMutex | ||
| real map[K]V | ||
| } | ||
|
|
||
| // Set sets a key to a given value | ||
| func (m *MutexMap[K, V]) Set(key K, value V) { | ||
| m.Lock() | ||
| defer m.Unlock() | ||
|
|
||
| m.real[key] = value | ||
| } | ||
|
|
||
| // Get returns the given key value and a bool if found | ||
| func (m *MutexMap[K, V]) Get(key K) (V, bool) { | ||
| m.RLock() | ||
| defer m.RUnlock() | ||
|
|
||
| value, ok := m.real[key] | ||
|
|
||
| return value, ok | ||
| } | ||
|
|
||
| // Delete removes a key from the internal map | ||
| func (m *MutexMap[K, V]) Delete(key K) { | ||
| m.Lock() | ||
| defer m.Unlock() | ||
|
|
||
| delete(m.real, key) | ||
| } | ||
|
|
||
| // Size returns the length of the internal map | ||
| func (m *MutexMap[K, V]) Size() int { | ||
| m.RLock() | ||
| defer m.RUnlock() | ||
|
|
||
| return len(m.real) | ||
| } | ||
|
|
||
| // Each runs a callback function for every item in the map | ||
| // The map should not be modified inside the callback function | ||
| func (m *MutexMap[K, V]) Each(callback func(key K, value V)) { | ||
| m.RLock() | ||
| defer m.RUnlock() | ||
|
|
||
| for key, value := range m.real { | ||
| callback(key, value) | ||
| } | ||
| } | ||
|
|
||
| // Clear removes all items from the `real` map | ||
| // Accepts an optional callback function ran for every item before it is deleted | ||
| func (m *MutexMap[K, V]) Clear(callback func(key K, value V)) { | ||
| m.Lock() | ||
| defer m.Unlock() | ||
|
|
||
| for key, value := range m.real { | ||
| if callback != nil { | ||
| callback(key, value) | ||
| } | ||
| delete(m.real, key) | ||
| } | ||
| } | ||
|
|
||
| // NewMutexMap returns a new instance of MutexMap with the provided key/value types | ||
| func NewMutexMap[K comparable, V any]() *MutexMap[K, V] { | ||
| return &MutexMap[K, V]{ | ||
| RWMutex: &sync.RWMutex{}, | ||
| real: make(map[K]V), | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| package nex | ||
|
|
||
| // PacketManager implements an API for pushing/popping packets in the correct order | ||
| type PacketManager struct { | ||
| currentSequenceID *Counter | ||
| packets []PacketInterface | ||
| } | ||
|
|
||
| // Next gets the next packet in the sequence. Returns nil if the next packet has not been sent yet | ||
| func (p *PacketManager) Next() PacketInterface { | ||
| var packet PacketInterface | ||
|
|
||
| for i := 0; i < len(p.packets); i++ { | ||
| if p.currentSequenceID.Value() == uint32(p.packets[i].SequenceID()) { | ||
| packet = p.packets[i] | ||
| p.RemoveByIndex(i) | ||
| p.currentSequenceID.Increment() | ||
| break | ||
| } | ||
| } | ||
|
|
||
| return packet | ||
| } | ||
|
|
||
| // Push adds a packet to the pool to choose from in Next | ||
| func (p *PacketManager) Push(packet PacketInterface) { | ||
| p.packets = append(p.packets, packet) | ||
| } | ||
|
|
||
| // RemoveByIndex removes a packet from the pool using it's index in the slice | ||
| func (p *PacketManager) RemoveByIndex(i int) { | ||
| // * https://stackoverflow.com/a/37335777 | ||
| p.packets[i] = p.packets[len(p.packets)-1] | ||
| p.packets = p.packets[:len(p.packets)-1] | ||
| } | ||
|
|
||
| // NewPacketManager returns a new PacketManager | ||
| func NewPacketManager() *PacketManager { | ||
| return &PacketManager{ | ||
| currentSequenceID: NewCounter(0), | ||
| packets: make([]PacketInterface, 0), | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,114 @@ | ||
| package nex | ||
|
|
||
| import ( | ||
| "time" | ||
| ) | ||
|
|
||
| // PendingPacket represents a packet which the server has sent but not received an ACK for | ||
| // it handles it's own retransmission on a per-packet timer | ||
| type PendingPacket struct { | ||
| ticking bool | ||
| ticker *time.Ticker | ||
| quit chan struct{} | ||
| packet PacketInterface | ||
| iterations *Counter | ||
| timeout time.Duration | ||
| timeoutInc time.Duration | ||
| maxIterations int | ||
| } | ||
|
|
||
| // BeginTimeoutTimer starts the pending packets timeout timer until it is either stopped or maxIterations is hit | ||
| func (p *PendingPacket) BeginTimeoutTimer() { | ||
| go func() { | ||
| for { | ||
| select { | ||
| case <-p.quit: | ||
| return | ||
| case <-p.ticker.C: | ||
| client := p.packet.Sender() | ||
| server := client.Server() | ||
|
|
||
| if int(p.iterations.Increment()) > p.maxIterations { | ||
| // * Max iterations hit. Assume client is dead | ||
| server.TimeoutKick(client) | ||
| p.StopTimeoutTimer() | ||
| return | ||
| } else { | ||
| if p.timeoutInc != 0 { | ||
| p.timeout += p.timeoutInc | ||
| p.ticker.Reset(p.timeout) | ||
| } | ||
|
|
||
| // * Resend the packet | ||
| server.SendRaw(client.Address(), p.packet.Bytes()) | ||
| } | ||
| } | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| // StopTimeoutTimer stops the packet retransmission timer | ||
| func (p *PendingPacket) StopTimeoutTimer() { | ||
| if p.ticking { | ||
| close(p.quit) | ||
| p.ticker.Stop() | ||
| p.ticking = false | ||
| } | ||
| } | ||
|
|
||
| // NewPendingPacket returns a new PendingPacket | ||
| func NewPendingPacket(packet PacketInterface, timeoutTime time.Duration, timeoutIncrement time.Duration, maxIterations int) *PendingPacket { | ||
| p := &PendingPacket{ | ||
| ticking: true, | ||
| ticker: time.NewTicker(timeoutTime), | ||
| quit: make(chan struct{}), | ||
| packet: packet, | ||
| iterations: NewCounter(0), | ||
| timeout: timeoutTime, | ||
| timeoutInc: timeoutIncrement, | ||
| maxIterations: maxIterations, | ||
| } | ||
|
|
||
| return p | ||
| } | ||
|
|
||
| // PacketResendManager manages all the pending packets sent the client waiting to be ACKed | ||
| type PacketResendManager struct { | ||
| pending *MutexMap[uint16, *PendingPacket] | ||
| timeoutTime time.Duration | ||
| timeoutInc time.Duration | ||
| maxIterations int | ||
| } | ||
|
|
||
| // Add creates a PendingPacket, adds it to the pool, and begins it's timeout timer | ||
| func (p *PacketResendManager) Add(packet PacketInterface) { | ||
| cached := NewPendingPacket(packet, p.timeoutTime, p.timeoutInc, p.maxIterations) | ||
| p.pending.Set(packet.SequenceID(), cached) | ||
|
|
||
| cached.BeginTimeoutTimer() | ||
| } | ||
|
|
||
| // Remove removes a packet from pool and stops it's timer | ||
| func (p *PacketResendManager) Remove(sequenceID uint16) { | ||
| if cached, ok := p.pending.Get(sequenceID); ok { | ||
| cached.StopTimeoutTimer() | ||
| p.pending.Delete(sequenceID) | ||
| } | ||
| } | ||
|
|
||
| // Clear removes all packets from pool and stops their timers | ||
| func (p *PacketResendManager) Clear() { | ||
| p.pending.Clear(func(key uint16, value *PendingPacket) { | ||
| value.StopTimeoutTimer() | ||
| }) | ||
| } | ||
|
|
||
| // NewPacketResendManager returns a new PacketResendManager | ||
| func NewPacketResendManager(timeoutTime time.Duration, timeoutIncrement time.Duration, maxIterations int) *PacketResendManager { | ||
| return &PacketResendManager{ | ||
| pending: NewMutexMap[uint16, *PendingPacket](), | ||
| timeoutTime: timeoutTime, | ||
| timeoutInc: timeoutIncrement, | ||
| maxIterations: maxIterations, | ||
| } | ||
| } |
Oops, something went wrong.