Skip to content

Commit ca922d4

Browse files
committed
fix(network): fix goroutine leak in waitTimeout
waitTimeout spawned a goroutine blocked on wg.Wait() and used an unbuffered channel. When the timeout fired the function returned but the goroutine stayed alive forever, one leak per timeout event. Fix: use a buffered channel (capacity 1) so the goroutine can always send and exit once wg reaches zero, even after the caller has returned. Accept a context.Context so cancellation is also propagated correctly. Fixes #2124 Signed-off-by: AkramBitar <akram@il.ibm.com>
1 parent 312f321 commit ca922d4

2 files changed

Lines changed: 144 additions & 7 deletions

File tree

token/services/network/fabric/network.go

Lines changed: 17 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -418,7 +418,7 @@ func (n *Network) LookupTransferMetadataKey(namespace string, key string, timeou
418418
logger.Debugf("failed to remove lookup listener [%s]: %v", transferMetadataKey, err)
419419
}
420420
}()
421-
if err := waitTimeout(wg, timeout); err != nil {
421+
if err := waitTimeout(context.Background(), wg, timeout); err != nil {
422422
return nil, err
423423
}
424424
logger.Debugf("lookup transfer metadata key [%s] from [%s] in namespace [%s], done, result [%s][%s]", key, transferMetadataKey, namespace, l.value, l.err)
@@ -579,17 +579,27 @@ func (l *lookupListener) OnError(ctx context.Context, key string, err error) {
579579
}
580580
}
581581

582-
func waitTimeout(wg *sync.WaitGroup, timeout time.Duration) error {
583-
c := make(chan struct{})
582+
// waitTimeout waits for wg to reach zero or for the timeout to expire.
583+
//
584+
// The channel is buffered (capacity 1) to fix a goroutine leak (#2124):
585+
// with an unbuffered channel the goroutine blocks on send forever once
586+
// waitTimeout returns via the timeout path. The buffer absorbs the send
587+
// so the goroutine always exits once wg reaches zero.
588+
func waitTimeout(ctx context.Context, wg *sync.WaitGroup, timeout time.Duration) error {
589+
ctx, cancel := context.WithTimeout(ctx, timeout)
590+
defer cancel()
591+
592+
done := make(chan struct{}, 1)
584593
go func() {
585-
defer close(c)
586594
wg.Wait()
595+
done <- struct{}{}
587596
}()
597+
588598
select {
589-
case <-c:
599+
case <-done:
590600
return nil
591-
case <-time.After(timeout):
592-
return errors.Errorf("context done")
601+
case <-ctx.Done():
602+
return ctx.Err()
593603
}
594604
}
595605

Lines changed: 127 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,127 @@
1+
/*
2+
Copyright IBM Corp. All Rights Reserved.
3+
4+
SPDX-License-Identifier: Apache-2.0
5+
*/
6+
7+
package fabric
8+
9+
import (
10+
"context"
11+
"runtime"
12+
"sync"
13+
"testing"
14+
"time"
15+
16+
"github.com/stretchr/testify/assert"
17+
"github.com/stretchr/testify/require"
18+
)
19+
20+
// TestWaitTimeout_CompletesBeforeTimeout verifies the happy path: wg reaches zero
21+
// before the deadline and waitTimeout returns nil with no goroutine left behind.
22+
func TestWaitTimeout_CompletesBeforeTimeout(t *testing.T) {
23+
wg := &sync.WaitGroup{}
24+
wg.Add(1)
25+
26+
go func() {
27+
time.Sleep(10 * time.Millisecond)
28+
wg.Done()
29+
}()
30+
31+
err := waitTimeout(context.Background(), wg, 5*time.Second)
32+
require.NoError(t, err)
33+
}
34+
35+
// TestWaitTimeout_TimeoutFires verifies that waitTimeout returns an error when
36+
// the WaitGroup does not reach zero within the timeout.
37+
func TestWaitTimeout_TimeoutFires(t *testing.T) {
38+
wg := &sync.WaitGroup{}
39+
wg.Add(1) // never Done'd — simulates a stuck operation
40+
41+
err := waitTimeout(context.Background(), wg, 50*time.Millisecond)
42+
require.Error(t, err)
43+
44+
// Unblock the goroutine left behind so the test and runtime clean up.
45+
wg.Done()
46+
}
47+
48+
// TestWaitTimeout_ContextCancelled verifies that a cancelled context causes
49+
// waitTimeout to return immediately with context.Canceled, even when the
50+
// timeout has not expired.
51+
func TestWaitTimeout_ContextCancelled(t *testing.T) {
52+
wg := &sync.WaitGroup{}
53+
wg.Add(1) // never Done'd
54+
55+
ctx, cancel := context.WithCancel(context.Background())
56+
cancel() // cancel immediately
57+
58+
err := waitTimeout(ctx, wg, 10*time.Second) // long timeout — must not block
59+
require.ErrorIs(t, err, context.Canceled)
60+
61+
wg.Done() // clean up the stuck goroutine
62+
}
63+
64+
// TestWaitTimeout_NoGoroutineLeak is the regression test for issue #2124.
65+
//
66+
// Before the fix, waitTimeout used an unbuffered channel and a bare goroutine
67+
// that parked on wg.Wait() with no way to be signalled. When the timeout fired
68+
// the function returned but the goroutine remained alive forever, one per
69+
// timeout event.
70+
//
71+
// After the fix the channel is buffered (capacity 1), so the goroutine can
72+
// always send and exit once wg reaches zero, even if waitTimeout has already
73+
// returned. This test confirms that the goroutine count returns to its
74+
// pre-call baseline within a short grace period after the wg is released.
75+
func TestWaitTimeout_NoGoroutineLeak(t *testing.T) {
76+
// Stabilise the baseline goroutine count before the test.
77+
runtime.GC()
78+
time.Sleep(10 * time.Millisecond)
79+
before := runtime.NumGoroutine()
80+
81+
wg := &sync.WaitGroup{}
82+
wg.Add(1)
83+
84+
// Fire the timeout — this is the path that previously leaked.
85+
err := waitTimeout(context.Background(), wg, 20*time.Millisecond)
86+
require.Error(t, err, "expected timeout error")
87+
88+
// Release the WaitGroup so the background goroutine can exit.
89+
wg.Done()
90+
91+
// Give the runtime a moment to schedule and collect the goroutine.
92+
assert.Eventually(t, func() bool {
93+
runtime.GC()
94+
return runtime.NumGoroutine() <= before+1 // +1 for test framework variance
95+
}, 2*time.Second, 50*time.Millisecond,
96+
"goroutine leaked: count did not return to baseline after wg.Done()")
97+
}
98+
99+
// TestWaitTimeout_MultipleTimeouts verifies that repeated timeout events do not
100+
// accumulate goroutines — each goroutine exits once its wg is released.
101+
func TestWaitTimeout_MultipleTimeouts(t *testing.T) {
102+
const n = 10
103+
104+
runtime.GC()
105+
time.Sleep(10 * time.Millisecond)
106+
before := runtime.NumGoroutine()
107+
108+
wgs := make([]*sync.WaitGroup, n)
109+
for i := range n {
110+
wg := &sync.WaitGroup{}
111+
wg.Add(1)
112+
wgs[i] = wg
113+
err := waitTimeout(context.Background(), wg, 20*time.Millisecond)
114+
require.Error(t, err)
115+
}
116+
117+
// Release all WaitGroups so all background goroutines can exit.
118+
for _, wg := range wgs {
119+
wg.Done()
120+
}
121+
122+
assert.Eventually(t, func() bool {
123+
runtime.GC()
124+
return runtime.NumGoroutine() <= before+1
125+
}, 2*time.Second, 50*time.Millisecond,
126+
"goroutines leaked after %d timeout events", n)
127+
}

0 commit comments

Comments
 (0)