Repository: libp2p/go-libp2p
Pull request #2096 (2023-02-14 13:27:15)

use atomic.Int32 and atomic.Int64
Fixes: https://github.com/libp2p/go-libp2p/issues/2095


Patch:
diff --git a/p2p/host/eventbus/basic.go b/p2p/host/eventbus/basic.go
index ae477887..42365a79 100644
--- a/p2p/host/eventbus/basic.go
+++ b/p2p/host/eventbus/basic.go
@@ -50,7 +50,7 @@ func (e *emitter) Close() error {
 	if !e.closed.CompareAndSwap(false, true) {
 		return fmt.Errorf("closed an emitter more than once")
 	}
-	if atomic.AddInt32(&e.n.nEmitters, -1) == 0 {
+	if e.n.nEmitters.Add(-1) == 0 {
 		e.dropper(e.typ)
 	}
 	return nil
@@ -100,7 +100,7 @@ func (b *basicBus) tryDropNode(typ reflect.Type) {
 	}
 
 	n.lk.Lock()
-	if atomic.LoadInt32(&n.nEmitters) > 0 || len(n.sinks) > 0 {
+	if n.nEmitters.Load() > 0 || len(n.sinks) > 0 {
 		n.lk.Unlock()
 		b.lk.Unlock()
 		return // still in use
@@ -178,7 +178,7 @@ func (s *sub) Close() error {
 			}
 		}
 
-		tryDrop := len(n.sinks) == 0 && atomic.LoadInt32(&n.nEmitters) == 0
+		tryDrop := len(n.sinks) == 0 && n.nEmitters.Load() == 0
 
 		n.lk.Unlock()
 
@@ -294,7 +294,7 @@ func (b *basicBus) Emitter(evtType interface{}, opts ...event.EmitterOpt) (e eve
 	typ = typ.Elem()
 
 	b.withNode(typ, func(n *node) {
-		atomic.AddInt32(&n.nEmitters, 1)
+		n.nEmitters.Add(1)
 		n.keepLast = n.keepLast || settings.makeStateful
 		e = &emitter{n: n, typ: typ, dropper: b.tryDropNode, w: b.wildcard, metricsTracer: b.metricsTracer}
 	}, nil)
@@ -319,13 +319,13 @@ func (b *basicBus) GetAllEventTypes() []reflect.Type {
 
 type wildcardNode struct {
 	sync.RWMutex
-	nSinks        int32
+	nSinks        atomic.Int32
 	sinks         []*namedSink
 	metricsTracer MetricsTracer
 }
 
 func (n *wildcardNode) addSink(sink *namedSink) {
-	atomic.AddInt32(&n.nSinks, 1) // ok to do outside the lock
+	n.nSinks.Add(1) // ok to do outside the lock
 	n.Lock()
 	n.sinks = append(n.sinks, sink)
 	n.Unlock()
@@ -336,7 +336,7 @@ func (n *wildcardNode) addSink(sink *namedSink) {
 }
 
 func (n *wildcardNode) removeSink(ch chan interface{}) {
-	atomic.AddInt32(&n.nSinks, -1) // ok to do outside the lock
+	n.nSinks.Add(-1) // ok to do outside the lock
 	n.Lock()
 	for i := 0; i < len(n.sinks); i++ {
 		if n.sinks[i].ch == ch {
@@ -349,7 +349,7 @@ func (n *wildcardNode) removeSink(ch chan interface{}) {
 }
 
 func (n *wildcardNode) emit(evt interface{}) {
-	if atomic.LoadInt32(&n.nSinks) == 0 {
+	if n.nSinks.Load() == 0 {
 		return
 	}
 
@@ -372,7 +372,7 @@ type node struct {
 	typ reflect.Type
 
 	// emitter ref count
-	nEmitters int32
+	nEmitters atomic.Int32
 
 	keepLast bool
 	last     interface{}
diff --git a/p2p/host/eventbus/opts.go b/p2p/host/eventbus/opts.go
index 3922be0a..837a0683 100644
--- a/p2p/host/eventbus/opts.go
+++ b/p2p/host/eventbus/opts.go
@@ -12,7 +12,7 @@ type subSettings struct {
 	name   string
 }
 
-var subCnt int64
+var subCnt atomic.Int64
 
 var subSettingsDefault = subSettings{
 	buffer: 16,
@@ -34,7 +34,7 @@ func newSubSettings() subSettings {
 		}
 		settings.name = fmt.Sprintf("%s-L%d", file, line)
 	} else {
-		settings.name = fmt.Sprintf("subscriber-%d", atomic.AddInt64(&subCnt, 1))
+		settings.name = fmt.Sprintf("subscriber-%d", subCnt.Add(1))
 	}
 	return settings
 }
diff --git a/p2p/net/connmgr/connmgr.go b/p2p/net/connmgr/connmgr.go
index c124c6bb..df3d96d5 100644
--- a/p2p/net/connmgr/connmgr.go
+++ b/p2p/net/connmgr/connmgr.go
@@ -38,7 +38,7 @@ type BasicConnMgr struct {
 
 	// channel-based semaphore that enforces only a single trim is in progress
 	trimMutex sync.Mutex
-	connCount int32
+	connCount atomic.Int32
 	// to be accessed atomically. This is mimicking the implementation of a sync.Once.
 	// Take care of correct alignment when modifying this struct.
 	trimCount uint64
@@ -158,7 +158,7 @@ func NewConnManager(low, hi int, opts ...Option) (*BasicConnMgr, error) {
 // We don't pay attention to the silence period or the grace period.
 // We try to not kill protected connections, but if that turns out to be necessary, not connection is safe!
 func (cm *BasicConnMgr) memoryEmergency() {
-	connCount := int(atomic.LoadInt32(&cm.connCount))
+	connCount := int(cm.connCount.Load())
 	target := connCount - cm.cfg.lowWater
 	if target < 0 {
 		log.Warnw("Low on memory, but we only have a few connections", "num", connCount, "low watermark", cm.cfg.lowWater)
@@ -346,7 +346,7 @@ func (cm *BasicConnMgr) background() {
 	for {
 		select {
 		case <-ticker.C:
-			if atomic.LoadInt32(&cm.connCount) < int32(cm.cfg.highWater) {
+			if cm.connCount.Load() < int32(cm.cfg.highWater) {
 				// Below high water, skip.
 				continue
 			}
@@ -456,7 +456,7 @@ func (cm *BasicConnMgr) getConnsToClose() []network.Conn {
 		return nil
 	}
 
-	if int(atomic.LoadInt32(&cm.connCount)) <= cm.cfg.lowWater {
+	if int(cm.connCount.Load()) <= cm.cfg.lowWater {
 		log.Info("open connection count below limit")
 		return nil
 	}
@@ -632,7 +632,7 @@ func (cm *BasicConnMgr) GetInfo() CMInfo {
 		LowWater:    cm.cfg.lowWater,
 		LastTrim:    lastTrim,
 		GracePeriod: cm.cfg.gracePeriod,
-		ConnCount:   int(atomic.LoadInt32(&cm.connCount)),
+		ConnCount:   int(cm.connCount.Load()),
 	}
 }
 
@@ -686,7 +686,7 @@ func (nn *cmNotifee) Connected(n network.Network, c network.Conn) {
 	}
 
 	pinfo.conns[c] = cm.clock.Now()
-	atomic.AddInt32(&cm.connCount, 1)
+	cm.connCount.Add(1)
 }
 
 // Disconnected is called by notifiers to inform that an existing connection has been closed or terminated.
@@ -715,7 +715,7 @@ func (nn *cmNotifee) Disconnected(n network.Network, c network.Conn) {
 	if len(cinf.conns) == 0 {
 		delete(s.peers, p)
 	}
-	atomic.AddInt32(&cm.connCount, -1)
+	cm.connCount.Add(-1)
 }
 
 // Listen is no-op in this implementation.
diff --git a/p2p/net/mock/mock_conn.go b/p2p/net/mock/mock_conn.go
index 487db8c7..b6fe8f2d 100644
--- a/p2p/net/mock/mock_conn.go
+++ b/p2p/net/mock/mock_conn.go
@@ -14,7 +14,7 @@ import (
 	manet "github.com/multiformats/go-multiaddr/net"
 )
 
-var connCounter int64
+var connCounter atomic.Int64
 
 // conn represents one side's perspective of a
 // live connection between two peers.
@@ -49,7 +49,7 @@ func newConn(ln, rn *peernet, l *link, dir network.Direction) *conn {
 	c.local = ln.peer
 	c.remote = rn.peer
 	c.stat.Direction = dir
-	c.id = atomic.AddInt64(&connCounter, 1)
+	c.id = connCounter.Add(1)
 
 	c.localAddr = ln.ps.Addrs(ln.peer)[0]
 	for _, a := range rn.ps.Addrs(rn.peer) {
diff --git a/p2p/net/mock/mock_stream.go b/p2p/net/mock/mock_stream.go
index f6cb842e..c85cca54 100644
--- a/p2p/net/mock/mock_stream.go
+++ b/p2p/net/mock/mock_stream.go
@@ -13,7 +13,7 @@ import (
 	"github.com/libp2p/go-libp2p/core/protocol"
 )
 
-var streamCounter int64
+var streamCounter atomic.Int64
 
 // stream implements network.Stream
 type stream struct {
@@ -57,7 +57,7 @@ func newStream(w *io.PipeWriter, r *io.PipeReader, dir network.Direction) *strea
 	s := &stream{
 		read:      r,
 		write:     w,
-		id:        atomic.AddInt64(&streamCounter, 1),
+		id:        streamCounter.Add(1),
 		reset:     make(chan struct{}, 1),
 		close:     make(chan struct{}, 1),
 		closed:    make(chan struct{}),
diff --git a/p2p/protocol/circuitv1/relay/relay.go b/p2p/protocol/circuitv1/relay/relay.go
index 296ef306..6e6e5b03 100644
--- a/p2p/protocol/circuitv1/relay/relay.go
+++ b/p2p/protocol/circuitv1/relay/relay.go
@@ -309,10 +309,10 @@ func (r *Relay) handleHopStream(s network.Stream, msg *pb.CircuitRelay) {
 	// reset deadline
 	bs.SetDeadline(time.Time{})
 
-	goroutines := new(int32)
-	*goroutines = 2
+	var goroutines atomic.Int32
+	goroutines.Store(2)
 	done := func() {
-		if atomic.AddInt32(goroutines, -1) == 0 {
+		if goroutines.Add(-1) == 0 {
 			s.Close()
 			bs.Close()
 			cleanup()
diff --git a/p2p/protocol/circuitv2/relay/relay.go b/p2p/protocol/circuitv2/relay/relay.go
index e24cd7f3..ca50d424 100644
--- a/p2p/protocol/circuitv2/relay/relay.go
+++ b/p2p/protocol/circuitv2/relay/relay.go
@@ -377,11 +377,11 @@ func (r *Relay) handleConnect(s network.Stream, msg *pbv2.HopMessage) {
 
 	log.Infof("relaying connection from %s to %s", src, dest.ID)
 
-	goroutines := new(int32)
-	*goroutines = 2
+	var goroutines atomic.Int32
+	goroutines.Store(2)
 
 	done := func() {
-		if atomic.AddInt32(goroutines, -1) == 0 {
+		if goroutines.Add(-1) == 0 {
 			s.Close()
 			bs.Close()
 			cleanup()
diff --git a/p2p/protocol/internal/circuitv1-deprecated/relay.go b/p2p/protocol/internal/circuitv1-deprecated/relay.go
index f3a491d2..3d67bc07 100644
--- a/p2p/protocol/internal/circuitv1-deprecated/relay.go
+++ b/p2p/protocol/internal/circuitv1-deprecated/relay.go
@@ -54,8 +54,8 @@ type Relay struct {
 	incoming chan *Conn
 
 	// atomic counters
-	streamCount  int32
-	liveHopCount int32
+	streamCount  atomic.Int32
+	liveHopCount atomic.Int32
 
 	// per peer hop counters
 	mx       sync.Mutex
@@ -130,7 +130,7 @@ func NewRelay(h host.Host, upgrader transport.Upgrader, opts ...RelayOpt) (*Rela
 // sides of the hop stream. This ensures that connections with many hop streams will be protected
 // from pruning, thus minimizing disruption from connection trimming in a relay node.
 func (r *Relay) addLiveHop(from, to peer.ID) {
-	atomic.AddInt32(&r.liveHopCount, 1)
+	r.liveHopCount.Add(1)
 	r.host.ConnManager().UpsertTag(from, "relay-hop-stream", incrementTag)
 	r.host.ConnManager().UpsertTag(to, "relay-hop-stream", incrementTag)
 }
@@ -138,14 +138,14 @@ func (r *Relay) addLiveHop(from, to peer.ID) {
 // Decrement the live hpo count and decrement the connection manager tags for the two sides
 // of the hop stream.
 func (r *Relay) rmLiveHop(from, to peer.ID) {
-	atomic.AddInt32(&r.liveHopCount, -1)
+	r.liveHopCount.Add(-1)
 	r.host.ConnManager().UpsertTag(from, "relay-hop-stream", decrementTag)
 	r.host.ConnManager().UpsertTag(to, "relay-hop-stream", decrementTag)
 
 }
 
 func (r *Relay) GetActiveHops() int32 {
-	return atomic.LoadInt32(&r.liveHopCount)
+	return r.liveHopCount.Load()
 }
 
 func (r *Relay) DialPeer(ctx context.Context, relay peer.AddrInfo, dest peer.AddrInfo) (*Conn, error) {
@@ -280,9 +280,9 @@ func (r *Relay) handleHopStream(s network.Stream, msg *pb.CircuitRelay) {
 		return
 	}
 
-	streamCount := atomic.AddInt32(&r.streamCount, 1)
-	liveHopCount := atomic.LoadInt32(&r.liveHopCount)
-	defer atomic.AddInt32(&r.streamCount, -1)
+	streamCount := r.streamCount.Add(1)
+	liveHopCount := r.liveHopCount.Load()
+	defer r.streamCount.Add(-1)
 
 	if (streamCount + liveHopCount) > int32(HopStreamLimit) {
 		log.Warn("hop stream limit exceeded; resetting stream")
@@ -391,10 +391,10 @@ func (r *Relay) handleHopStream(s network.Stream, msg *pb.CircuitRelay) {
 
 	r.addLiveHop(src.ID, dst.ID)
 
-	goroutines := new(int32)
-	*goroutines = 2
+	var goroutines atomic.Int32
+	goroutines.Store(2)
 	done := func() {
-		if atomic.AddInt32(goroutines, -1) == 0 {
+		if goroutines.Add(-1) == 0 {
 			s.Close()
 			bs.Close()
 			r.rmLiveHop(src.ID, dst.ID)


Test patch:
diff --git a/core/test/crypto.go b/core/test/crypto.go
index 5032485b..ff09cfac 100644
--- a/core/test/crypto.go
+++ b/core/test/crypto.go
@@ -7,11 +7,11 @@ import (
 	ci "github.com/libp2p/go-libp2p/core/crypto"
 )
 
-var globalSeed int64
+var globalSeed atomic.Int64
 
 func RandTestKeyPair(typ, bits int) (ci.PrivKey, ci.PubKey, error) {
 	// workaround for low time resolution
-	seed := atomic.AddInt64(&globalSeed, 1)
+	seed := globalSeed.Add(1)
 	return SeededTestKeyPair(typ, bits, seed)
 }
 
diff --git a/p2p/host/autorelay/autorelay_test.go b/p2p/host/autorelay/autorelay_test.go
index 768c8ec2..53d7d67e 100644
--- a/p2p/host/autorelay/autorelay_test.go
+++ b/p2p/host/autorelay/autorelay_test.go
@@ -251,17 +251,17 @@ func TestBackoff(t *testing.T) {
 	)
 	require.NoError(t, err)
 	defer r.Close()
-	var reservations int32
+	var reservations atomic.Int32
 	r.SetStreamHandler(protoIDv2, func(str network.Stream) {
-		atomic.AddInt32(&reservations, 1)
+		reservations.Add(1)
 		str.Reset()
 	})
 
-	var counter int32 // to be used atomically
+	var counter atomic.Int32
 	h := newPrivateNode(t,
 		func(context.Context, int) <-chan peer.AddrInfo {
 			// always return the same node, and make sure we don't try to connect to it too frequently
-			atomic.AddInt32(&counter, 1)
+			counter.Add(1)
 			peerChan := make(chan peer.AddrInfo, 1)
 			peerChan <- peer.AddrInfo{ID: r.ID(), Addrs: r.Addrs()}
 			close(peerChan)
@@ -275,16 +275,16 @@ func TestBackoff(t *testing.T) {
 	)
 	defer h.Close()
 
-	require.Eventually(t, func() bool { return atomic.LoadInt32(&reservations) == 1 }, 3*time.Second, 20*time.Millisecond)
+	require.Eventually(t, func() bool { return reservations.Load() == 1 }, 3*time.Second, 20*time.Millisecond)
 	// make sure we don't add any relays yet
 	for i := 0; i < 2; i++ {
 		cl.Add(backoff / 3)
-		require.Equal(t, 1, int(atomic.LoadInt32(&reservations)))
+		require.Equal(t, 1, int(reservations.Load()))
 	}
 	cl.Add(backoff / 2)
-	require.Eventually(t, func() bool { return atomic.LoadInt32(&reservations) == 2 }, 3*time.Second, 20*time.Millisecond)
-	require.Less(t, int(atomic.LoadInt32(&counter)), 100) // just make sure we're not busy-looping
-	require.Equal(t, 2, int(atomic.LoadInt32(&reservations)))
+	require.Eventually(t, func() bool { return reservations.Load() == 2 }, 3*time.Second, 20*time.Millisecond)
+	require.Less(t, int(counter.Load()), 100) // just make sure we're not busy-looping
+	require.Equal(t, 2, int(reservations.Load()))
 }
 
 func TestStaticRelays(t *testing.T) {
diff --git a/p2p/host/eventbus/basic_test.go b/p2p/host/eventbus/basic_test.go
index d2ede36a..57362ce9 100644
--- a/p2p/host/eventbus/basic_test.go
+++ b/p2p/host/eventbus/basic_test.go
@@ -198,7 +198,7 @@ func TestClosingRaces(t *testing.T) {
 func TestSubMany(t *testing.T) {
 	bus := NewBus()
 
-	var r int32
+	var r atomic.Int32
 
 	n := getN()
 	var wait sync.WaitGroup
@@ -215,7 +215,7 @@ func TestSubMany(t *testing.T) {
 			defer sub.Close()
 
 			ready.Done()
-			atomic.AddInt32(&r, int32((<-sub.Out()).(EventB)))
+			r.Add(int32((<-sub.Out()).(EventB)))
 			wait.Done()
 		}()
 	}
@@ -231,7 +231,7 @@ func TestSubMany(t *testing.T) {
 	em.Emit(EventB(7))
 	wait.Wait()
 
-	if int(r) != 7*n {
+	if int(r.Load()) != 7*n {
 		t.Error("got wrong result")
 	}
 }
@@ -488,7 +488,7 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 
 	bus := NewBus()
 
-	var r int64
+	var r atomic.Int64
 
 	var wait sync.WaitGroup
 	var ready sync.WaitGroup
@@ -509,7 +509,7 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 				if !ok {
 					panic("wat")
 				}
-				atomic.AddInt64(&r, int64(e.(EventB)))
+				r.Add(int64(e.(EventB)))
 			}
 			wait.Done()
 		}()
@@ -538,7 +538,7 @@ func testMany(t testing.TB, subs, emits, msgs int, stateful bool) {
 
 	wait.Wait()
 
-	if int(r) != 97*subs*emits*msgs {
+	if int(r.Load()) != 97*subs*emits*msgs {
 		t.Fatal("got wrong result")
 	}
 }
diff --git a/p2p/net/connmgr/connmgr_test.go b/p2p/net/connmgr/connmgr_test.go
index 39db7b54..7468f671 100644
--- a/p2p/net/connmgr/connmgr_test.go
+++ b/p2p/net/connmgr/connmgr_test.go
@@ -367,7 +367,7 @@ func TestDoubleConnection(t *testing.T) {
 	not.Connected(nil, conn)
 	cm.TagPeer(conn.RemotePeer(), "foo", 10)
 	not.Connected(nil, conn)
-	if cm.connCount != 1 {
+	if cm.connCount.Load() != 1 {
 		t.Fatal("unexpected number of connections")
 	}
 	if cm.segments.get(conn.RemotePeer()).peers[conn.RemotePeer()].value != 10 {
@@ -386,7 +386,7 @@ func TestDisconnected(t *testing.T) {
 	cm.TagPeer(conn.RemotePeer(), "foo", 10)
 
 	not.Disconnected(nil, randConn(t, nil))
-	if cm.connCount != 1 {
+	if cm.connCount.Load() != 1 {
 		t.Fatal("unexpected number of connections")
 	}
 	if cm.segments.get(conn.RemotePeer()).peers[conn.RemotePeer()].value != 10 {
@@ -394,7 +394,7 @@ func TestDisconnected(t *testing.T) {
 	}
 
 	not.Disconnected(nil, &tconn{peer: conn.RemotePeer()})
-	if cm.connCount != 1 {
+	if cm.connCount.Load() != 1 {
 		t.Fatal("unexpected number of connections")
 	}
 	if cm.segments.get(conn.RemotePeer()).peers[conn.RemotePeer()].value != 10 {
@@ -402,7 +402,7 @@ func TestDisconnected(t *testing.T) {
 	}
 
 	not.Disconnected(nil, conn)
-	if cm.connCount != 0 {
+	if cm.connCount.Load() != 0 {
 		t.Fatal("unexpected number of connections")
 	}
 	if cm.segments.countPeers() != 0 {
@@ -495,7 +495,7 @@ func TestQuickBurstRespectsSilencePeriod(t *testing.T) {
 	if closed > 20 {
 		t.Fatalf("should have closed at most 20 connections, closed: %d", closed)
 	}
-	if total := closed + int(cm.connCount); total != 30 {
+	if total := closed + int(cm.connCount.Load()); total != 30 {
 		t.Fatalf("expected closed connections + open conn count to equal 30, value: %d", total)
 	}
 }
diff --git a/p2p/net/upgrader/listener_test.go b/p2p/net/upgrader/listener_test.go
index c6b1b885..be4a1f46 100644
--- a/p2p/net/upgrader/listener_test.go
+++ b/p2p/net/upgrader/listener_test.go
@@ -261,11 +261,11 @@ func TestAcceptQueueBacklogged(t *testing.T) {
 	defer ln.Close()
 
 	// setup AcceptQueueLength connections, but don't accept any of them
-	var counter int32 // to be used atomically
+	var counter atomic.Int32
 	doDial := func() {
 		conn, err := dial(t, u, ln.Multiaddr(), id, &network.NullScope{})
 		require.NoError(err)
-		atomic.AddInt32(&counter, 1)
+		counter.Add(1)
 		t.Cleanup(func() { conn.Close() })
 	}
 
@@ -273,20 +273,20 @@ func TestAcceptQueueBacklogged(t *testing.T) {
 		go doDial()
 	}
 
-	require.Eventually(func() bool { return int(atomic.LoadInt32(&counter)) == upgrader.AcceptQueueLength }, 2*time.Second, 50*time.Millisecond)
+	require.Eventually(func() bool { return int(counter.Load()) == upgrader.AcceptQueueLength }, 2*time.Second, 50*time.Millisecond)
 
 	// dial a new connection. This connection should not complete setup, since the queue is full
 	go doDial()
 
 	time.Sleep(100 * time.Millisecond)
-	require.Equal(int(atomic.LoadInt32(&counter)), upgrader.AcceptQueueLength)
+	require.Equal(int(counter.Load()), upgrader.AcceptQueueLength)
 
 	// accept a single connection. Now the new connection should be set up, and fill the queue again
 	conn, err := ln.Accept()
 	require.NoError(err)
 	require.NoError(conn.Close())
 
-	require.Eventually(func() bool { return int(atomic.LoadInt32(&counter)) == upgrader.AcceptQueueLength+1 }, 2*time.Second, 50*time.Millisecond)
+	require.Eventually(func() bool { return int(counter.Load()) == upgrader.AcceptQueueLength+1 }, 2*time.Second, 50*time.Millisecond)
 }
 
 func TestListenerConnectionGater(t *testing.T) {
diff --git a/p2p/test/resource-manager/rcmgr_test.go b/p2p/test/resource-manager/rcmgr_test.go
index 017b4929..c953c5c7 100644
--- a/p2p/test/resource-manager/rcmgr_test.go
+++ b/p2p/test/resource-manager/rcmgr_test.go
@@ -265,10 +265,10 @@ func TestResourceManagerServicePeerInbound(t *testing.T) {
 
 func waitForBarrier(count int32, timeout time.Duration) func() error {
 	ready := make(chan struct{})
-	wait := new(int32)
-	*wait = count
+	var wait atomic.Int32
+	wait.Store(count)
 	return func() error {
-		if atomic.AddInt32(wait, -1) == 0 {
+		if wait.Add(-1) == 0 {
 			close(ready)
 		}