Skip to content

Commit f4250b8

Browse files
tech-enginetech-engine
committed
refactor: use atomic pointer instead of mutex for hot path
1 parent e992098 commit f4250b8

1 file changed

Lines changed: 158 additions & 92 deletions

File tree

‎pkg/signal/signal.go‎

Lines changed: 158 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,7 @@ package signal
33
import (
44
"context"
55
"sync"
6+
"sync/atomic"
67

78
"github.com/tech-engine/goscrapy/pkg/core"
89
)
@@ -24,30 +25,30 @@ type ItemBus[OUT any] interface {
2425

2526
// typed signal bus
2627
type Bus[OUT any] struct {
27-
mu sync.RWMutex
28+
mu sync.Mutex
2829

2930
// spider signals
30-
spiderOpened []func(context.Context)
31-
spiderClosed []func(context.Context)
32-
spiderError []func(context.Context, error)
33-
spiderIdle []func(context.Context)
31+
spiderOpened atomic.Pointer[[]func(context.Context)]
32+
spiderClosed atomic.Pointer[[]func(context.Context)]
33+
spiderError atomic.Pointer[[]func(context.Context, error)]
34+
spiderIdle atomic.Pointer[[]func(context.Context)]
3435

3536
// item signals
36-
itemScraped []func(context.Context, OUT)
37-
itemDropped []func(context.Context, OUT, error)
38-
itemError []func(context.Context, OUT, error)
37+
itemScraped atomic.Pointer[[]func(context.Context, OUT)]
38+
itemDropped atomic.Pointer[[]func(context.Context, OUT, error)]
39+
itemError atomic.Pointer[[]func(context.Context, OUT, error)]
3940

4041
// request signals
41-
requestScheduled []func(context.Context, *core.Request)
42-
requestDropped []func(context.Context, *core.Request, error)
43-
requestError []func(context.Context, *core.Request, error)
42+
requestScheduled atomic.Pointer[[]func(context.Context, *core.Request)]
43+
requestDropped atomic.Pointer[[]func(context.Context, *core.Request, error)]
44+
requestError atomic.Pointer[[]func(context.Context, *core.Request, error)]
4445

4546
// response signals
46-
responseReceived []func(context.Context, core.IResponseReader)
47+
responseReceived atomic.Pointer[[]func(context.Context, core.IResponseReader)]
4748

4849
// engine signals
49-
engineStarted []func(context.Context)
50-
engineStopped []func(context.Context)
50+
engineStarted atomic.Pointer[[]func(context.Context)]
51+
engineStopped atomic.Pointer[[]func(context.Context)]
5152
}
5253

5354
// create new signal bus
@@ -58,194 +59,259 @@ func New[OUT any]() *Bus[OUT] {
5859
func (b *Bus[OUT]) OnSpiderOpened(h func(context.Context)) {
5960
b.mu.Lock()
6061
defer b.mu.Unlock()
61-
b.spiderOpened = append(b.spiderOpened, h)
62+
old := b.spiderOpened.Load()
63+
var newHandlers []func(context.Context)
64+
if old != nil {
65+
newHandlers = append([]func(context.Context){}, *old...)
66+
}
67+
newHandlers = append(newHandlers, h)
68+
b.spiderOpened.Store(&newHandlers)
6269
}
6370

6471
func (b *Bus[OUT]) OnSpiderClosed(h func(context.Context)) {
6572
b.mu.Lock()
6673
defer b.mu.Unlock()
67-
b.spiderClosed = append(b.spiderClosed, h)
74+
old := b.spiderClosed.Load()
75+
var newHandlers []func(context.Context)
76+
if old != nil {
77+
newHandlers = append([]func(context.Context){}, *old...)
78+
}
79+
newHandlers = append(newHandlers, h)
80+
b.spiderClosed.Store(&newHandlers)
6881
}
6982

7083
func (b *Bus[OUT]) OnSpiderError(h func(context.Context, error)) {
7184
b.mu.Lock()
7285
defer b.mu.Unlock()
73-
b.spiderError = append(b.spiderError, h)
86+
old := b.spiderError.Load()
87+
var newHandlers []func(context.Context, error)
88+
if old != nil {
89+
newHandlers = append([]func(context.Context, error){}, *old...)
90+
}
91+
newHandlers = append(newHandlers, h)
92+
b.spiderError.Store(&newHandlers)
7493
}
7594

7695
func (b *Bus[OUT]) OnSpiderIdle(h func(context.Context)) {
7796
b.mu.Lock()
7897
defer b.mu.Unlock()
79-
b.spiderIdle = append(b.spiderIdle, h)
98+
old := b.spiderIdle.Load()
99+
var newHandlers []func(context.Context)
100+
if old != nil {
101+
newHandlers = append([]func(context.Context){}, *old...)
102+
}
103+
newHandlers = append(newHandlers, h)
104+
b.spiderIdle.Store(&newHandlers)
80105
}
81106

82107
func (b *Bus[OUT]) OnItemScraped(h func(context.Context, OUT)) {
83108
b.mu.Lock()
84109
defer b.mu.Unlock()
85-
b.itemScraped = append(b.itemScraped, h)
110+
old := b.itemScraped.Load()
111+
var newHandlers []func(context.Context, OUT)
112+
if old != nil {
113+
newHandlers = append([]func(context.Context, OUT){}, *old...)
114+
}
115+
newHandlers = append(newHandlers, h)
116+
b.itemScraped.Store(&newHandlers)
86117
}
87118

88119
func (b *Bus[OUT]) OnItemDropped(h func(context.Context, OUT, error)) {
89120
b.mu.Lock()
90121
defer b.mu.Unlock()
91-
b.itemDropped = append(b.itemDropped, h)
122+
old := b.itemDropped.Load()
123+
var newHandlers []func(context.Context, OUT, error)
124+
if old != nil {
125+
newHandlers = append([]func(context.Context, OUT, error){}, *old...)
126+
}
127+
newHandlers = append(newHandlers, h)
128+
b.itemDropped.Store(&newHandlers)
92129
}
93130

94131
func (b *Bus[OUT]) OnItemError(h func(context.Context, OUT, error)) {
95132
b.mu.Lock()
96133
defer b.mu.Unlock()
97-
b.itemError = append(b.itemError, h)
134+
old := b.itemError.Load()
135+
var newHandlers []func(context.Context, OUT, error)
136+
if old != nil {
137+
newHandlers = append([]func(context.Context, OUT, error){}, *old...)
138+
}
139+
newHandlers = append(newHandlers, h)
140+
b.itemError.Store(&newHandlers)
98141
}
99142

100143
func (b *Bus[OUT]) OnRequestScheduled(h func(context.Context, *core.Request)) {
101144
b.mu.Lock()
102145
defer b.mu.Unlock()
103-
b.requestScheduled = append(b.requestScheduled, h)
146+
old := b.requestScheduled.Load()
147+
var newHandlers []func(context.Context, *core.Request)
148+
if old != nil {
149+
newHandlers = append([]func(context.Context, *core.Request){}, *old...)
150+
}
151+
newHandlers = append(newHandlers, h)
152+
b.requestScheduled.Store(&newHandlers)
104153
}
105154

106155
func (b *Bus[OUT]) OnRequestDropped(h func(context.Context, *core.Request, error)) {
107156
b.mu.Lock()
108157
defer b.mu.Unlock()
109-
b.requestDropped = append(b.requestDropped, h)
158+
old := b.requestDropped.Load()
159+
var newHandlers []func(context.Context, *core.Request, error)
160+
if old != nil {
161+
newHandlers = append([]func(context.Context, *core.Request, error){}, *old...)
162+
}
163+
newHandlers = append(newHandlers, h)
164+
b.requestDropped.Store(&newHandlers)
110165
}
111166

112167
func (b *Bus[OUT]) OnRequestError(h func(context.Context, *core.Request, error)) {
113168
b.mu.Lock()
114169
defer b.mu.Unlock()
115-
b.requestError = append(b.requestError, h)
170+
old := b.requestError.Load()
171+
var newHandlers []func(context.Context, *core.Request, error)
172+
if old != nil {
173+
newHandlers = append([]func(context.Context, *core.Request, error){}, *old...)
174+
}
175+
newHandlers = append(newHandlers, h)
176+
b.requestError.Store(&newHandlers)
116177
}
117178

118179
func (b *Bus[OUT]) OnResponseReceived(h func(context.Context, core.IResponseReader)) {
119180
b.mu.Lock()
120181
defer b.mu.Unlock()
121-
b.responseReceived = append(b.responseReceived, h)
182+
old := b.responseReceived.Load()
183+
var newHandlers []func(context.Context, core.IResponseReader)
184+
if old != nil {
185+
newHandlers = append([]func(context.Context, core.IResponseReader){}, *old...)
186+
}
187+
newHandlers = append(newHandlers, h)
188+
b.responseReceived.Store(&newHandlers)
122189
}
123190

124191
func (b *Bus[OUT]) OnEngineStarted(h func(context.Context)) {
125192
b.mu.Lock()
126193
defer b.mu.Unlock()
127-
b.engineStarted = append(b.engineStarted, h)
194+
old := b.engineStarted.Load()
195+
var newHandlers []func(context.Context)
196+
if old != nil {
197+
newHandlers = append([]func(context.Context){}, *old...)
198+
}
199+
newHandlers = append(newHandlers, h)
200+
b.engineStarted.Store(&newHandlers)
128201
}
129202

130203
func (b *Bus[OUT]) OnEngineStopped(h func(context.Context)) {
131204
b.mu.Lock()
132205
defer b.mu.Unlock()
133-
b.engineStopped = append(b.engineStopped, h)
206+
old := b.engineStopped.Load()
207+
var newHandlers []func(context.Context)
208+
if old != nil {
209+
newHandlers = append([]func(context.Context){}, *old...)
210+
}
211+
newHandlers = append(newHandlers, h)
212+
b.engineStopped.Store(&newHandlers)
134213
}
135214

136215
func (b *Bus[OUT]) EmitSpiderOpened(ctx context.Context) {
137-
b.mu.RLock()
138-
handlers := b.spiderOpened
139-
b.mu.RUnlock()
140-
for _, h := range handlers {
141-
h(ctx)
216+
if handlers := b.spiderOpened.Load(); handlers != nil {
217+
for _, h := range *handlers {
218+
h(ctx)
219+
}
142220
}
143221
}
144222

145223
func (b *Bus[OUT]) EmitSpiderClosed(ctx context.Context) {
146-
b.mu.RLock()
147-
handlers := b.spiderClosed
148-
b.mu.RUnlock()
149-
for _, h := range handlers {
150-
h(ctx)
224+
if handlers := b.spiderClosed.Load(); handlers != nil {
225+
for _, h := range *handlers {
226+
h(ctx)
227+
}
151228
}
152229
}
153230

154231
func (b *Bus[OUT]) EmitSpiderError(ctx context.Context, err error) {
155-
b.mu.RLock()
156-
handlers := b.spiderError
157-
b.mu.RUnlock()
158-
for _, h := range handlers {
159-
h(ctx, err)
232+
if handlers := b.spiderError.Load(); handlers != nil {
233+
for _, h := range *handlers {
234+
h(ctx, err)
235+
}
160236
}
161237
}
162238

163239
func (b *Bus[OUT]) EmitSpiderIdle(ctx context.Context) {
164-
b.mu.RLock()
165-
handlers := b.spiderIdle
166-
b.mu.RUnlock()
167-
for _, h := range handlers {
168-
h(ctx)
240+
if handlers := b.spiderIdle.Load(); handlers != nil {
241+
for _, h := range *handlers {
242+
h(ctx)
243+
}
169244
}
170245
}
171246

172247
func (b *Bus[OUT]) EmitItemScraped(ctx context.Context, item OUT) {
173-
b.mu.RLock()
174-
handlers := b.itemScraped
175-
b.mu.RUnlock()
176-
for _, h := range handlers {
177-
h(ctx, item)
248+
if handlers := b.itemScraped.Load(); handlers != nil {
249+
for _, h := range *handlers {
250+
h(ctx, item)
251+
}
178252
}
179253
}
180254

181255
func (b *Bus[OUT]) EmitItemDropped(ctx context.Context, item OUT, err error) {
182-
b.mu.RLock()
183-
handlers := b.itemDropped
184-
b.mu.RUnlock()
185-
for _, h := range handlers {
186-
h(ctx, item, err)
256+
if handlers := b.itemDropped.Load(); handlers != nil {
257+
for _, h := range *handlers {
258+
h(ctx, item, err)
259+
}
187260
}
188261
}
189262

190263
func (b *Bus[OUT]) EmitItemError(ctx context.Context, item OUT, err error) {
191-
b.mu.RLock()
192-
handlers := b.itemError
193-
b.mu.RUnlock()
194-
for _, h := range handlers {
195-
h(ctx, item, err)
264+
if handlers := b.itemError.Load(); handlers != nil {
265+
for _, h := range *handlers {
266+
h(ctx, item, err)
267+
}
196268
}
197269
}
198270

199271
func (b *Bus[OUT]) EmitRequestScheduled(ctx context.Context, req *core.Request) {
200-
b.mu.RLock()
201-
handlers := b.requestScheduled
202-
b.mu.RUnlock()
203-
for _, h := range handlers {
204-
h(ctx, req)
272+
if handlers := b.requestScheduled.Load(); handlers != nil {
273+
for _, h := range *handlers {
274+
h(ctx, req)
275+
}
205276
}
206277
}
207278

208279
func (b *Bus[OUT]) EmitRequestDropped(ctx context.Context, req *core.Request, err error) {
209-
b.mu.RLock()
210-
handlers := b.requestDropped
211-
b.mu.RUnlock()
212-
for _, h := range handlers {
213-
h(ctx, req, err)
280+
if handlers := b.requestDropped.Load(); handlers != nil {
281+
for _, h := range *handlers {
282+
h(ctx, req, err)
283+
}
214284
}
215285
}
216286

217287
func (b *Bus[OUT]) EmitRequestError(ctx context.Context, req *core.Request, err error) {
218-
b.mu.RLock()
219-
handlers := b.requestError
220-
b.mu.RUnlock()
221-
for _, h := range handlers {
222-
h(ctx, req, err)
288+
if handlers := b.requestError.Load(); handlers != nil {
289+
for _, h := range *handlers {
290+
h(ctx, req, err)
291+
}
223292
}
224293
}
225294

226295
func (b *Bus[OUT]) EmitResponseReceived(ctx context.Context, resp core.IResponseReader) {
227-
b.mu.RLock()
228-
handlers := b.responseReceived
229-
b.mu.RUnlock()
230-
for _, h := range handlers {
231-
h(ctx, resp)
296+
if handlers := b.responseReceived.Load(); handlers != nil {
297+
for _, h := range *handlers {
298+
h(ctx, resp)
299+
}
232300
}
233301
}
234302

235303
func (b *Bus[OUT]) EmitEngineStarted(ctx context.Context) {
236-
b.mu.RLock()
237-
handlers := b.engineStarted
238-
b.mu.RUnlock()
239-
for _, h := range handlers {
240-
h(ctx)
304+
if handlers := b.engineStarted.Load(); handlers != nil {
305+
for _, h := range *handlers {
306+
h(ctx)
307+
}
241308
}
242309
}
243310

244311
func (b *Bus[OUT]) EmitEngineStopped(ctx context.Context) {
245-
b.mu.RLock()
246-
handlers := b.engineStopped
247-
b.mu.RUnlock()
248-
for _, h := range handlers {
249-
h(ctx)
312+
if handlers := b.engineStopped.Load(); handlers != nil {
313+
for _, h := range *handlers {
314+
h(ctx)
315+
}
250316
}
251317
}

0 commit comments

Comments
 (0)