Skip to content

Commit 4cb78b1

Browse files
committed
Add MutexMap Atomic/RAtomic and fix crash
Fix formatting
1 parent 5f38bf3 commit 4cb78b1

File tree

2 files changed

+128
-49
lines changed

2 files changed

+128
-49
lines changed

mutex_map.go

Lines changed: 116 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -2,53 +2,42 @@ package nex
22

33
import "sync"
44

5+
type MapInterface[K comparable, V any] struct {
6+
real map[K]V
7+
}
8+
59
// MutexMap implements a map type with go routine safe accessors through mutex locks. Embeds sync.RWMutex
610
type MutexMap[K comparable, V any] struct {
7-
*sync.RWMutex
8-
real map[K]V
11+
mutex *sync.RWMutex
12+
mapInterface MapInterface[K,V]
913
}
1014

1115
// Set sets a key to a given value
12-
func (m *MutexMap[K, V]) Set(key K, value V) {
13-
m.Lock()
14-
defer m.Unlock()
15-
16+
func (m *MapInterface[K, V]) Set(key K, value V) {
1617
m.real[key] = value
1718
}
1819

1920
// Get returns the given key value and a bool if found
20-
func (m *MutexMap[K, V]) Get(key K) (V, bool) {
21-
m.RLock()
22-
defer m.RUnlock()
23-
21+
func (m *MapInterface[K, V]) Get(key K) (V, bool) {
2422
value, ok := m.real[key]
2523

2624
return value, ok
2725
}
2826

2927
// Has checks if a key exists in the map
30-
func (m *MutexMap[K, V]) Has(key K) bool {
31-
m.RLock()
32-
defer m.RUnlock()
33-
28+
func (m *MapInterface[K, V]) Has(key K) bool {
3429
_, ok := m.real[key]
3530
return ok
3631
}
3732

3833
// Delete removes a key from the internal map
39-
func (m *MutexMap[K, V]) Delete(key K) {
40-
m.Lock()
41-
defer m.Unlock()
42-
34+
func (m *MapInterface[K, V]) Delete(key K) {
4335
delete(m.real, key)
4436
}
4537

4638
// DeleteIf deletes every element if the predicate returns true.
4739
// Returns the amount of elements deleted.
48-
func (m *MutexMap[K, V]) DeleteIf(predicate func(key K, value V) bool) int {
49-
m.Lock()
50-
defer m.Unlock()
51-
40+
func (m *MapInterface[K, V]) DeleteIf(predicate func(key K, value V) bool) int {
5241
amount := 0
5342
for key, value := range m.real {
5443
if predicate(key, value) {
@@ -61,31 +50,22 @@ func (m *MutexMap[K, V]) DeleteIf(predicate func(key K, value V) bool) int {
6150
}
6251

6352
// RunAndDelete runs a callback and removes the key afterwards
64-
func (m *MutexMap[K, V]) RunAndDelete(key K, callback func(key K, value V)) {
65-
m.Lock()
66-
defer m.Unlock()
67-
53+
func (m *MapInterface[K, V]) RunAndDelete(key K, callback func(key K, value V)) {
6854
if value, ok := m.real[key]; ok {
6955
callback(key, value)
7056
delete(m.real, key)
7157
}
7258
}
7359

7460
// Size returns the length of the internal map
75-
func (m *MutexMap[K, V]) Size() int {
76-
m.RLock()
77-
defer m.RUnlock()
78-
61+
func (m *MapInterface[K, V]) Size() int {
7962
return len(m.real)
8063
}
8164

8265
// Each runs a callback function for every item in the map
8366
// The map should not be modified inside the callback function
8467
// Returns true if the loop was terminated early
85-
func (m *MutexMap[K, V]) Each(callback func(key K, value V) bool) bool {
86-
m.RLock()
87-
defer m.RUnlock()
88-
68+
func (m *MapInterface[K, V]) Each(callback func(key K, value V) bool) bool {
8969
for key, value := range m.real {
9070
if callback(key, value) {
9171
return true
@@ -97,10 +77,7 @@ func (m *MutexMap[K, V]) Each(callback func(key K, value V) bool) bool {
9777

9878
// Clear removes all items from the `real` map
9979
// Accepts an optional callback function ran for every item before it is deleted
100-
func (m *MutexMap[K, V]) Clear(callback func(key K, value V)) {
101-
m.Lock()
102-
defer m.Unlock()
103-
80+
func (m *MapInterface[K, V]) Clear(callback func(key K, value V)) {
10481
for key, value := range m.real {
10582
if callback != nil {
10683
callback(key, value)
@@ -109,10 +86,109 @@ func (m *MutexMap[K, V]) Clear(callback func(key K, value V)) {
10986
}
11087
}
11188

89+
// Set sets a key to a given value
90+
func (m *MutexMap[K, V]) Set(key K, value V) {
91+
m.mutex.Lock()
92+
defer m.mutex.Unlock()
93+
94+
m.mapInterface.Set(key, value)
95+
}
96+
97+
// Get returns the given key value and a bool if found
98+
func (m *MutexMap[K, V]) Get(key K) (V, bool) {
99+
m.mutex.RLock()
100+
defer m.mutex.RUnlock()
101+
102+
value, ok := m.mapInterface.Get(key)
103+
104+
return value, ok
105+
}
106+
107+
// Has checks if a key exists in the map
108+
func (m *MutexMap[K, V]) Has(key K) bool {
109+
m.mutex.RLock()
110+
defer m.mutex.RUnlock()
111+
112+
return m.mapInterface.Has(key)
113+
}
114+
115+
// Delete removes a key from the internal map
116+
func (m *MutexMap[K, V]) Delete(key K) {
117+
m.mutex.Lock()
118+
defer m.mutex.Unlock()
119+
120+
m.mapInterface.Delete(key)
121+
}
122+
123+
// DeleteIf deletes every element if the predicate returns true.
124+
// Returns the amount of elements deleted.
125+
func (m *MutexMap[K, V]) DeleteIf(predicate func(key K, value V) bool) int {
126+
m.mutex.Lock()
127+
defer m.mutex.Unlock()
128+
129+
return m.mapInterface.DeleteIf(predicate)
130+
}
131+
132+
// RunAndDelete runs a callback and removes the key afterwards
133+
func (m *MutexMap[K, V]) RunAndDelete(key K, callback func(key K, value V)) {
134+
m.mutex.Lock()
135+
defer m.mutex.Unlock()
136+
137+
m.mapInterface.RunAndDelete(key, callback)
138+
}
139+
140+
// Size returns the length of the internal map
141+
func (m *MutexMap[K, V]) Size() int {
142+
m.mutex.RLock()
143+
defer m.mutex.RUnlock()
144+
145+
return m.mapInterface.Size()
146+
}
147+
148+
// Each runs a callback function for every item in the map
149+
// The map should not be modified inside the callback function
150+
// Returns true if the loop was terminated early
151+
func (m *MutexMap[K, V]) Each(callback func(key K, value V) bool) bool {
152+
m.mutex.RLock()
153+
defer m.mutex.RUnlock()
154+
155+
return m.mapInterface.Each(callback)
156+
}
157+
158+
// Clear removes all items from the `real` map
159+
// Accepts an optional callback function ran for every item before it is deleted
160+
func (m *MutexMap[K, V]) Clear(callback func(key K, value V)) {
161+
m.mutex.Lock()
162+
defer m.mutex.Unlock()
163+
164+
m.mapInterface.Clear(callback)
165+
}
166+
167+
// RAtomic read-locks the map and runs the provided callback. All the operations to
168+
// the map interface in the callback will be done atomically. Do not perform write
169+
// operations to the map interface.
170+
func (m *MutexMap[K, V]) RAtomic(callback func(mapInterface *MapInterface[K,V])) {
171+
m.mutex.RLock()
172+
defer m.mutex.RUnlock()
173+
174+
callback(&m.mapInterface)
175+
}
176+
177+
// Atomic write-locks the map and runs the provided callback. All the operations to
178+
// the map interface in the callback will be done atomically.
179+
func (m *MutexMap[K, V]) Atomic(callback func(mapInterface *MapInterface[K,V])) {
180+
m.mutex.Lock()
181+
defer m.mutex.Unlock()
182+
183+
callback(&m.mapInterface)
184+
}
185+
112186
// NewMutexMap returns a new instance of MutexMap with the provided key/value types
113187
func NewMutexMap[K comparable, V any]() *MutexMap[K, V] {
114188
return &MutexMap[K, V]{
115-
RWMutex: &sync.RWMutex{},
116-
real: make(map[K]V),
189+
mutex: &sync.RWMutex{},
190+
mapInterface: MapInterface[K,V]{
191+
real: make(map[K]V),
192+
},
117193
}
118194
}

sliding_window.go

Lines changed: 12 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -17,16 +17,19 @@ type SlidingWindow struct {
1717
func (sw *SlidingWindow) Update(packet PRUDPPacketInterface) []PRUDPPacketInterface {
1818
packets := make([]PRUDPPacketInterface, 0)
1919

20-
if packet.SequenceID() >= sw.incomingSequenceIDCounter.Value && !sw.pendingPackets.Has(packet.SequenceID()) {
21-
sw.pendingPackets.Set(packet.SequenceID(), packet)
22-
23-
for sw.pendingPackets.Has(sw.incomingSequenceIDCounter.Value) {
24-
storedPacket, _ := sw.pendingPackets.Get(sw.incomingSequenceIDCounter.Value)
25-
packets = append(packets, storedPacket)
26-
sw.pendingPackets.Delete(sw.incomingSequenceIDCounter.Value)
27-
sw.incomingSequenceIDCounter.Next()
20+
sw.pendingPackets.Atomic(func(mapInterface *MapInterface[uint16, PRUDPPacketInterface]) {
21+
if packet.SequenceID() >= sw.incomingSequenceIDCounter.Value && !mapInterface.Has(packet.SequenceID()) {
22+
mapInterface.Set(packet.SequenceID(), packet)
23+
24+
for mapInterface.Has(sw.incomingSequenceIDCounter.Value) {
25+
if storedPacket, ok := mapInterface.Get(sw.incomingSequenceIDCounter.Value); ok {
26+
packets = append(packets, storedPacket)
27+
}
28+
mapInterface.Delete(sw.incomingSequenceIDCounter.Value)
29+
sw.incomingSequenceIDCounter.Next()
30+
}
2831
}
29-
}
32+
})
3033

3134
return packets
3235
}

0 commit comments

Comments
 (0)