Skip to content

Commit 2e45653

Browse files
mjuragamjuraga
committed
BUG/MAJOR: runtime: fix race condition when calling runtime socket
Switched from using a channel and a goroutine to simple mutex, to avoid race condition when the goroutine is finished and a channel is being written to, making it block forever. Now a simple mutex is controlling the read/write to the runtime unix socket
1 parent 0c6a33e commit 2e45653

6 files changed

+44
-134
lines changed

‎runtime/acls_test.go

+6-20
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,8 @@
11
package runtime
22

33
import (
4-
"context"
54
"reflect"
65
"testing"
7-
"time"
86

97
"github.com/haproxytech/client-native/v5/models"
108
)
@@ -53,9 +51,7 @@ func TestSingleRuntime_ShowACLS(t *testing.T) {
5351
t.Run(tt.name, func(t *testing.T) {
5452
haProxy.SetResponses(&tt.socketResponse)
5553
s := &SingleRuntime{}
56-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
57-
defer cancel()
58-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
54+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
5955
if err != nil {
6056
t.Errorf("SingleRuntime.Init() error = %v", err)
6157
return
@@ -161,9 +157,7 @@ func TestSingleRuntime_GetACL(t *testing.T) {
161157
t.Run(tt.name, func(t *testing.T) {
162158
haProxy.SetResponses(&tt.socketResponse)
163159
s := &SingleRuntime{}
164-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
165-
defer cancel()
166-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
160+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
167161
if err != nil {
168162
t.Errorf("SingleRuntime.Init() error = %v", err)
169163
return
@@ -256,9 +250,7 @@ func TestSingleRuntime_ShowACLFileEntries(t *testing.T) {
256250
t.Run(tt.name, func(t *testing.T) {
257251
haProxy.SetResponses(&tt.socketResponse)
258252
s := &SingleRuntime{}
259-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
260-
defer cancel()
261-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
253+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
262254
if err != nil {
263255
t.Errorf("SingleRuntime.Init() error = %v", err)
264256
return
@@ -333,9 +325,7 @@ func TestSingleRuntime_GetACLFileEntry(t *testing.T) {
333325
t.Run(tt.name, func(t *testing.T) {
334326
haProxy.SetResponses(&tt.socketResponse)
335327
s := &SingleRuntime{}
336-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
337-
defer cancel()
338-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
328+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
339329
if err != nil {
340330
t.Errorf("SingleRuntime.Init() error = %v", err)
341331
return
@@ -441,9 +431,7 @@ func TestSingleRuntime_AddACLFileEntry(t *testing.T) {
441431
t.Run(tt.name, func(t *testing.T) {
442432
haProxy.SetResponses(&tt.socketResponse)
443433
s := &SingleRuntime{}
444-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
445-
defer cancel()
446-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
434+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
447435
if err != nil {
448436
t.Errorf("SingleRuntime.Init() error = %v", err)
449437
return
@@ -527,9 +515,7 @@ func TestSingleRuntime_DeleteACLFileEntry(t *testing.T) {
527515
t.Run(tt.name, func(t *testing.T) {
528516
haProxy.SetResponses(&tt.socketResponse)
529517
s := &SingleRuntime{}
530-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
531-
defer cancel()
532-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
518+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
533519
if err != nil {
534520
t.Errorf("SingleRuntime.Init() error = %v", err)
535521
return

‎runtime/certs_test.go

+8-30
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,6 @@
11
package runtime
22

33
import (
4-
"context"
54
"reflect"
65
"testing"
76
"time"
@@ -57,9 +56,7 @@ func TestSingleRuntime_ShowCerts(t *testing.T) {
5756
t.Run(tt.name, func(t *testing.T) {
5857
haProxy.SetResponses(&tt.socketResponse)
5958
s := &SingleRuntime{}
60-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
61-
defer cancel()
62-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
59+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
6360
if err != nil {
6461
t.Errorf("SingleRuntime.Init() error = %v", err)
6562
return
@@ -148,9 +145,7 @@ func TestSingleRuntime_GetCert(t *testing.T) {
148145
t.Run(tt.name, func(t *testing.T) {
149146
haProxy.SetResponses(&tt.socketResponse)
150147
s := &SingleRuntime{}
151-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
152-
defer cancel()
153-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
148+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
154149
if err != nil {
155150
t.Errorf("SingleRuntime.Init() error = %v", err)
156151
return
@@ -234,9 +229,7 @@ func TestSingleRuntime_ShowCertEntry(t *testing.T) {
234229
t.Run(tt.name, func(t *testing.T) {
235230
haProxy.SetResponses(&tt.socketResponse)
236231
s := &SingleRuntime{}
237-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
238-
defer cancel()
239-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
232+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
240233
if err != nil {
241234
t.Errorf("SingleRuntime.Init() error = %v", err)
242235
return
@@ -259,7 +252,6 @@ func TestSingleRuntime_NewCertEntry(t *testing.T) {
259252
defer haProxy.Stop()
260253

261254
type fields struct {
262-
jobs chan Task
263255
socketPath string
264256
worker int
265257
process int
@@ -312,9 +304,7 @@ func TestSingleRuntime_NewCertEntry(t *testing.T) {
312304
t.Run(tt.name, func(t *testing.T) {
313305
haProxy.SetResponses(&tt.socketResponse)
314306
s := &SingleRuntime{}
315-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
316-
defer cancel()
317-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
307+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
318308
if err != nil {
319309
t.Errorf("SingleRuntime.Init() error = %v", err)
320310
return
@@ -332,7 +322,6 @@ func TestSingleRuntime_SetCertEntry(t *testing.T) {
332322
defer haProxy.Stop()
333323

334324
type fields struct {
335-
jobs chan Task
336325
socketPath string
337326
worker int
338327
process int
@@ -401,9 +390,7 @@ func TestSingleRuntime_SetCertEntry(t *testing.T) {
401390
t.Run(tt.name, func(t *testing.T) {
402391
haProxy.SetResponses(&tt.socketResponse)
403392
s := &SingleRuntime{}
404-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
405-
defer cancel()
406-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
393+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
407394
if err != nil {
408395
t.Errorf("SingleRuntime.Init() error = %v", err)
409396
return
@@ -421,7 +408,6 @@ func TestSingleRuntime_CommitCertEntry(t *testing.T) {
421408
defer haProxy.Stop()
422409

423410
type fields struct {
424-
jobs chan Task
425411
socketPath string
426412
worker int
427413
process int
@@ -476,9 +462,7 @@ func TestSingleRuntime_CommitCertEntry(t *testing.T) {
476462
t.Run(tt.name, func(t *testing.T) {
477463
haProxy.SetResponses(&tt.socketResponse)
478464
s := &SingleRuntime{}
479-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
480-
defer cancel()
481-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
465+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
482466
if err != nil {
483467
t.Errorf("SingleRuntime.Init() error = %v", err)
484468
return
@@ -496,7 +480,6 @@ func TestSingleRuntime_AbortCertEntry(t *testing.T) {
496480
defer haProxy.Stop()
497481

498482
type fields struct {
499-
jobs chan Task
500483
socketPath string
501484
worker int
502485
process int
@@ -549,9 +532,7 @@ func TestSingleRuntime_AbortCertEntry(t *testing.T) {
549532
t.Run(tt.name, func(t *testing.T) {
550533
haProxy.SetResponses(&tt.socketResponse)
551534
s := &SingleRuntime{}
552-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
553-
defer cancel()
554-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
535+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
555536
if err != nil {
556537
t.Errorf("SingleRuntime.Init() error = %v", err)
557538
return
@@ -569,7 +550,6 @@ func TestSingleRuntime_DeleteCertEntry(t *testing.T) {
569550
defer haProxy.Stop()
570551

571552
type fields struct {
572-
jobs chan Task
573553
socketPath string
574554
worker int
575555
process int
@@ -622,9 +602,7 @@ func TestSingleRuntime_DeleteCertEntry(t *testing.T) {
622602
t.Run(tt.name, func(t *testing.T) {
623603
haProxy.SetResponses(&tt.socketResponse)
624604
s := &SingleRuntime{}
625-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
626-
defer cancel()
627-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
605+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
628606
if err != nil {
629607
t.Errorf("SingleRuntime.Init() error = %v", err)
630608
return

‎runtime/crt-lists_test.go

+5-20
Original file line numberDiff line numberDiff line change
@@ -1,10 +1,8 @@
11
package runtime
22

33
import (
4-
"context"
54
"reflect"
65
"testing"
7-
"time"
86
)
97

108
func TestSingleRuntime_ShowCrtLists(t *testing.T) {
@@ -51,9 +49,7 @@ func TestSingleRuntime_ShowCrtLists(t *testing.T) {
5149
t.Run(tt.name, func(t *testing.T) {
5250
haProxy.SetResponses(&tt.socketResponse)
5351
s := &SingleRuntime{}
54-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
55-
defer cancel()
56-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
52+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
5753
if err != nil {
5854
t.Errorf("SingleRuntime.Init() error = %v", err)
5955
return
@@ -138,9 +134,7 @@ func TestSingleRuntime_GetCrtList(t *testing.T) {
138134
t.Run(tt.name, func(t *testing.T) {
139135
haProxy.SetResponses(&tt.socketResponse)
140136
s := &SingleRuntime{}
141-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
142-
defer cancel()
143-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
137+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
144138
if err != nil {
145139
t.Errorf("SingleRuntime.Init() error = %v", err)
146140
return
@@ -163,7 +157,6 @@ func TestSingleRuntime_ShowCrtListEntries(t *testing.T) {
163157
defer haProxy.Stop()
164158

165159
type fields struct {
166-
jobs chan Task
167160
socketPath string
168161
worker int
169162
process int
@@ -239,9 +232,7 @@ func TestSingleRuntime_ShowCrtListEntries(t *testing.T) {
239232
t.Run(tt.name, func(t *testing.T) {
240233
haProxy.SetResponses(&tt.socketResponse)
241234
s := &SingleRuntime{}
242-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
243-
defer cancel()
244-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
235+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
245236
if err != nil {
246237
t.Errorf("SingleRuntime.Init() error = %v", err)
247238
return
@@ -266,7 +257,6 @@ func TestSingleRuntime_AddCrtListEntry(t *testing.T) {
266257
defer haProxy.Stop()
267258

268259
type fields struct {
269-
jobs chan Task
270260
socketPath string
271261
worker int
272262
process int
@@ -342,9 +332,7 @@ func TestSingleRuntime_AddCrtListEntry(t *testing.T) {
342332
t.Run(tt.name, func(t *testing.T) {
343333
haProxy.SetResponses(&tt.socketResponse)
344334
s := &SingleRuntime{}
345-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
346-
defer cancel()
347-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
335+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
348336
if err != nil {
349337
t.Errorf("SingleRuntime.Init() error = %v", err)
350338
return
@@ -362,7 +350,6 @@ func TestSingleRuntime_DeleteCrtListEntry(t *testing.T) {
362350
defer haProxy.Stop()
363351

364352
type fields struct {
365-
jobs chan Task
366353
socketPath string
367354
worker int
368355
process int
@@ -412,9 +399,7 @@ func TestSingleRuntime_DeleteCrtListEntry(t *testing.T) {
412399
t.Run(tt.name, func(t *testing.T) {
413400
haProxy.SetResponses(&tt.socketResponse)
414401
s := &SingleRuntime{}
415-
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(time.Second))
416-
defer cancel()
417-
err := s.Init(ctx, tt.fields.socketPath, tt.fields.process, tt.fields.worker)
402+
err := s.Init(tt.fields.socketPath, tt.fields.process, tt.fields.worker)
418403
if err != nil {
419404
t.Errorf("SingleRuntime.Init() error = %v", err)
420405
return

‎runtime/interface.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -147,7 +147,7 @@ type Runtime interface {
147147
Raw
148148
}
149149

150-
func New(ctx context.Context, opt ...options.RuntimeOption) (Runtime, error) {
150+
func New(_ context.Context, opt ...options.RuntimeOption) (Runtime, error) {
151151
c := &client{
152152
options: options.RuntimeOptions{},
153153
}
@@ -161,9 +161,9 @@ func New(ctx context.Context, opt ...options.RuntimeOption) (Runtime, error) {
161161
}
162162

163163
if c.options.MasterSocketData != nil {
164-
err = c.initWithMasterSocket(ctx, c.options)
164+
err = c.initWithMasterSocket(c.options)
165165
} else {
166-
err = c.initWithSockets(ctx, c.options)
166+
err = c.initWithSockets(c.options)
167167
}
168168
if err != nil {
169169
return nil, err

‎runtime/runtime_client.go

+9-10
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,6 @@
1616
package runtime
1717

1818
import (
19-
"context"
2019
"fmt"
2120
"io"
2221
"mime/multipart"
@@ -38,7 +37,7 @@ import (
3837
type client struct {
3938
haproxyVersion *HAProxyVersion
4039
options options.RuntimeOptions
41-
runtimes []SingleRuntime
40+
runtimes []*SingleRuntime
4241
}
4342

4443
const (
@@ -49,12 +48,12 @@ const (
4948
maxBufSize = 8192
5049
)
5150

52-
func (c *client) initWithSockets(ctx context.Context, opt options.RuntimeOptions) error {
51+
func (c *client) initWithSockets(opt options.RuntimeOptions) error {
5352
socketPath := opt.Sockets
54-
c.runtimes = make([]SingleRuntime, 0)
53+
c.runtimes = make([]*SingleRuntime, 0)
5554
for process, path := range socketPath {
56-
runtime := SingleRuntime{}
57-
err := runtime.Init(ctx, path, 0, process, opt)
55+
runtime := &SingleRuntime{}
56+
err := runtime.Init(path, 0, process, opt)
5857
if err != nil {
5958
return err
6059
}
@@ -64,7 +63,7 @@ func (c *client) initWithSockets(ctx context.Context, opt options.RuntimeOptions
6463
return nil
6564
}
6665

67-
func (c *client) initWithMasterSocket(ctx context.Context, opt options.RuntimeOptions) error {
66+
func (c *client) initWithMasterSocket(opt options.RuntimeOptions) error {
6867
masterSocketPath := opt.MasterSocketData.MasterSocketPath
6968
nbproc := c.options.MasterSocketData.Nbproc
7069
if nbproc == 0 {
@@ -73,10 +72,10 @@ func (c *client) initWithMasterSocket(ctx context.Context, opt options.RuntimeOp
7372
if masterSocketPath == "" {
7473
return fmt.Errorf("master socket not configured")
7574
}
76-
c.runtimes = make([]SingleRuntime, nbproc)
75+
c.runtimes = make([]*SingleRuntime, nbproc)
7776
for i := 1; i <= nbproc; i++ {
78-
runtime := SingleRuntime{}
79-
err := runtime.Init(ctx, masterSocketPath, i, i, opt)
77+
runtime := &SingleRuntime{}
78+
err := runtime.Init(masterSocketPath, i, i, opt)
8079
if err != nil {
8180
return err
8281
}

0 commit comments

Comments
 (0)