diff --git a/common/deliverclient/blocksprovider/stoppable_sleeper_test.go b/common/deliverclient/blocksprovider/stoppable_sleeper_test.go index b61591beb91..901d76bdfa0 100644 --- a/common/deliverclient/blocksprovider/stoppable_sleeper_test.go +++ b/common/deliverclient/blocksprovider/stoppable_sleeper_test.go @@ -24,11 +24,11 @@ func SetSleeper(d sleeperSetter, sleeper customSleeper) { } type testSleeper struct { - c int32 + c atomic.Int32 } func (s *testSleeper) Sleep(duration time.Duration) { - atomic.AddInt32(&s.c, 1) + s.c.Add(1) } func TestSleeper(t *testing.T) { @@ -75,7 +75,7 @@ func TestSleeper(t *testing.T) { require.Eventually(t, func() bool { - return atomic.LoadInt32(&s.c) == 10 + return s.c.Load() == 10 }, 10*time.Second, time.Millisecond) }) diff --git a/discovery/client/signer_test.go b/discovery/client/signer_test.go index 952afe51a77..92649b70997 100644 --- a/discovery/client/signer_test.go +++ b/discovery/client/signer_test.go @@ -40,9 +40,9 @@ func TestSameMessage(t *testing.T) { func TestDifferentMessages(t *testing.T) { var n uint = 50 - var signedInvokedCount uint32 + var signedInvokedCount atomic.Uint32 sign := func(msg []byte) ([]byte, error) { - atomic.AddUint32(&signedInvokedCount, 1) + signedInvokedCount.Add(1) return msg, nil } @@ -64,21 +64,21 @@ func TestDifferentMessages(t *testing.T) { // Query once parallelSignRange(0, n) - require.Equal(t, uint32(n), atomic.LoadUint32(&signedInvokedCount)) + require.Equal(t, uint32(n), signedInvokedCount.Load()) // Query twice parallelSignRange(0, n) - require.Equal(t, uint32(n), atomic.LoadUint32(&signedInvokedCount)) + require.Equal(t, uint32(n), signedInvokedCount.Load()) // Query thrice on a disjoint range for i := n + 1; i < 2*n; i++ { parallelSignRange(i, i+1) } - oldSignedInvokedCount := atomic.LoadUint32(&signedInvokedCount) + oldSignedInvokedCount := signedInvokedCount.Load() // Ensure that some of the early messages 0-n were purged from memory parallelSignRange(0, n) - require.True(t, oldSignedInvokedCount < atomic.LoadUint32(&signedInvokedCount)) + require.True(t, oldSignedInvokedCount < signedInvokedCount.Load()) } func TestFailure(t *testing.T) { diff --git a/discovery/test/integration_test.go b/discovery/test/integration_test.go index c546e29c165..e7db4212aca 100644 --- a/discovery/test/integration_test.go +++ b/discovery/test/integration_test.go @@ -291,7 +291,7 @@ func TestRevocation(t *testing.T) { res, err := client.Send(context.Background(), req, client.AuthInfo) require.NoError(t, err) // Record number of times we deserialized the identity - firstCount := atomic.LoadUint32(&service.sup.deserializeIdentityCount) + firstCount := service.sup.deserializeIdentityCount.Load() // Do the same query again peers, err := res.ForChannel("mychannel").Peers() @@ -302,7 +302,7 @@ func TestRevocation(t *testing.T) { require.NoError(t, err) // The amount of times deserializeIdentity was called should not have changed // because requests should have hit the cache - secondCount := atomic.LoadUint32(&service.sup.deserializeIdentityCount) + secondCount := service.sup.deserializeIdentityCount.Load() require.Equal(t, firstCount, secondCount) // Now, increment the config sequence @@ -312,14 +312,14 @@ func TestRevocation(t *testing.T) { service.sup.sequenceWrapper.instance.Store(v) // Revoke all identities inside the MSP manager - atomic.AddUint32(&service.sup.mspWrapper.blocks, uint32(1)) + service.sup.mspWrapper.blocks.Add(uint32(1)) // Send the query for the third time res, err = client.Send(context.Background(), req, client.AuthInfo) require.NoError(t, err) // The cache should have been purged, thus deserializeIdentity should have been // called an additional time - thirdCount := atomic.LoadUint32(&service.sup.deserializeIdentityCount) + thirdCount := service.sup.deserializeIdentityCount.Load() require.NotEqual(t, thirdCount, secondCount) // We should be denied access @@ -339,15 +339,15 @@ func (c *client) newConnection() (*grpc.ClientConn, error) { } type mspWrapper struct { - deserializeIdentityCount uint32 + deserializeIdentityCount atomic.Uint32 msp.MSPManager mspConfigs map[string]*msprotos.FabricMSPConfig - blocks uint32 + blocks atomic.Uint32 } func (w *mspWrapper) DeserializeIdentity(serializedIdentity []byte) (msp.Identity, error) { - atomic.AddUint32(&w.deserializeIdentityCount, 1) - if atomic.LoadUint32(&w.blocks) == uint32(1) { + w.deserializeIdentityCount.Add(1) + if w.blocks.Load() == uint32(1) { return nil, errors.New("failed deserializing identity") } return w.MSPManager.DeserializeIdentity(serializedIdentity) diff --git a/gossip/comm/comm_test.go b/gossip/comm/comm_test.go index 637179ae03a..d0fb9fc6c83 100644 --- a/gossip/comm/comm_test.go +++ b/gossip/comm/comm_test.go @@ -1077,7 +1077,7 @@ func waitForMessages(t *testing.T, msgChan chan uint64, count int, errMsg string } func TestConcurrentCloseSend(t *testing.T) { - var stopping int32 + var stopping atomic.Int32 comm1, _ := newCommInstance(t, naiveSec) comm2, port2 := newCommInstance(t, naiveSec) @@ -1092,12 +1092,12 @@ func TestConcurrentCloseSend(t *testing.T) { comm1.Send(createGossipMsg(), remotePeer(port2)) close(ready) - for atomic.LoadInt32(&stopping) == int32(0) { + for stopping.Load() == int32(0) { comm1.Send(createGossipMsg(), remotePeer(port2)) } }() <-ready comm2.Stop() - atomic.StoreInt32(&stopping, int32(1)) + stopping.Store(int32(1)) <-done } diff --git a/gossip/comm/metrics_test.go b/gossip/comm/metrics_test.go index 6835943f3cd..5af8e69efcc 100644 --- a/gossip/comm/metrics_test.go +++ b/gossip/comm/metrics_test.go @@ -26,9 +26,9 @@ func newCommInstanceWithMetrics(t *testing.T, sec *naiveSecProvider, metrics *me func TestMetrics(t *testing.T) { testMetricProvider := mocks.TestUtilConstructMetricProvider() - var overflown uint32 + var overflown atomic.Uint32 testMetricProvider.FakeBufferOverflow.AddStub = func(delta float64) { - atomic.StoreUint32(&overflown, uint32(1)) + overflown.Store(uint32(1)) } fakeCommMetrics := metrics.NewGossipMetrics(testMetricProvider.FakeProvider).CommMetrics @@ -59,7 +59,7 @@ func TestMetrics(t *testing.T) { // Send messages until the buffer overflow event emission is detected for { comm1.Send(createGossipMsg(), remotePeer(port2)) - if atomic.LoadUint32(&overflown) == uint32(1) { + if overflown.Load() == uint32(1) { t.Log("Buffer overflow detected") break } @@ -80,5 +80,5 @@ func TestMetrics(t *testing.T) { testMetricProvider.FakeBufferOverflow.AddArgsForCall(0), ) - require.Equal(t, uint32(1), atomic.LoadUint32(&overflown)) + require.Equal(t, uint32(1), overflown.Load()) } diff --git a/gossip/discovery/discovery_test.go b/gossip/discovery/discovery_test.go index 01b80fc4fcf..dbe5fe1ca7b 100644 --- a/gossip/discovery/discovery_test.go +++ b/gossip/discovery/discovery_test.go @@ -94,8 +94,8 @@ func (m *mockAnchorPeerTracker) IsAnchorPeer(endpoint string) bool { type dummyCommModule struct { validatedMessages chan *protoext.SignedGossipMessage - msgsReceived uint32 - msgsSent uint32 + msgsReceived atomic.Uint32 + msgsSent atomic.Uint32 id string identitySwitch chan common.PKIidType presumeDead chan common.PKIidType @@ -108,7 +108,7 @@ type dummyCommModule struct { shouldGossip bool disableComm bool mock *mock.Mock - signCount uint32 + signCount atomic.Uint32 } type gossipInstance struct { @@ -145,7 +145,7 @@ func (comm *dummyCommModule) recordValidation(validatedMessages chan *protoext.S } func (comm *dummyCommModule) SignMessage(am *proto.GossipMessage, internalEndpoint string) *proto.Envelope { - atomic.AddUint32(&comm.signCount, 1) + comm.signCount.Add(1) protoext.NoopSign(am) secret := &proto.Secret{ @@ -207,7 +207,7 @@ func (comm *dummyCommModule) SendToPeer(peer *NetworkMember, msg *protoext.Signe s, _ := protoext.NoopSign(msg.GossipMessage) comm.streams[peer.Endpoint].Send(s.Envelope) comm.lock.Unlock() - atomic.AddUint32(&comm.msgsSent, 1) + comm.msgsSent.Add(1) } func (comm *dummyCommModule) Ping(peer *NetworkMember) bool { @@ -262,11 +262,11 @@ func (comm *dummyCommModule) CloseConn(peer *NetworkMember) { } func (g *gossipInstance) receivedMsgCount() int { - return int(atomic.LoadUint32(&g.comm.msgsReceived)) + return int(g.comm.msgsReceived.Load()) } func (g *gossipInstance) sentMsgCount() int { - return int(atomic.LoadUint32(&g.comm.msgsSent)) + return int(g.comm.msgsSent.Load()) } func (g *gossipInstance) discoveryImpl() *gossipDiscoveryImpl { @@ -313,7 +313,7 @@ func (g *gossipInstance) GossipStream(stream proto.Gossip_GossipStreamServer) er ID: common.PKIidType("testID"), }, } - atomic.AddUint32(&g.comm.msgsReceived, 1) + g.comm.msgsReceived.Add(1) if aliveMsg := gMsg.GetAliveMsg(); aliveMsg != nil { g.tryForwardMessage(gMsg) @@ -598,10 +598,10 @@ func TestNoSigningIfNoMembership(t *testing.T) { inst := createDiscoveryInstance(8931, "foreveralone", nil) defer inst.Stop() time.Sleep(defaultTestConfig.AliveTimeInterval * 10) - assert.Zero(t, atomic.LoadUint32(&inst.comm.signCount)) + assert.Zero(t, inst.comm.signCount.Load()) inst.InitiateSync(10000) - assert.Zero(t, atomic.LoadUint32(&inst.comm.signCount)) + assert.Zero(t, inst.comm.signCount.Load()) } func TestValidation(t *testing.T) { @@ -1167,14 +1167,14 @@ func TestCertificateChange(t *testing.T) { // Shutdown the second peer waitUntilOrFailBlocking(t, p2.Stop) - var pingCountFrom1 uint32 - var pingCountFrom3 uint32 + var pingCountFrom1 atomic.Uint32 + var pingCountFrom3 atomic.Uint32 // Program mocks to increment ping counters p1.comm.lock.Lock() p1.comm.mock = &mock.Mock{} p1.comm.mock.On("SendToPeer", mock.Anything, mock.Anything) p1.comm.mock.On("Ping").Run(func(arguments mock.Arguments) { - atomic.AddUint32(&pingCountFrom1, 1) + pingCountFrom1.Add(1) }) p1.comm.lock.Unlock() @@ -1182,16 +1182,16 @@ func TestCertificateChange(t *testing.T) { p3.comm.mock = &mock.Mock{} p3.comm.mock.On("SendToPeer", mock.Anything, mock.Anything) p3.comm.mock.On("Ping").Run(func(arguments mock.Arguments) { - atomic.AddUint32(&pingCountFrom3, 1) + pingCountFrom3.Add(1) }) p3.comm.lock.Unlock() pingCount1 := func() uint32 { - return atomic.LoadUint32(&pingCountFrom1) + return pingCountFrom1.Load() } pingCount3 := func() uint32 { - return atomic.LoadUint32(&pingCountFrom3) + return pingCountFrom3.Load() } c1 := pingCount1() diff --git a/gossip/election/election.go b/gossip/election/election.go index 2b9e13ad3cd..4c3db109acc 100644 --- a/gossip/election/election.go +++ b/gossip/election/election.go @@ -175,9 +175,9 @@ type leaderElectionSvcImpl struct { stopChan chan struct{} interruptChan chan struct{} stopWG sync.WaitGroup - isLeader int32 - leaderExists int32 - yield int32 + isLeader atomic.Int32 + leaderExists atomic.Int32 + yield atomic.Int32 sleeping bool adapter LeaderElectionAdapter logger util.Logger @@ -224,7 +224,7 @@ func (le *leaderElectionSvcImpl) handleMessage(msg Msg) { if msg.IsProposal() { le.proposals.Add(string(msg.SenderID())) } else if msg.IsDeclaration() { - atomic.StoreInt32(&le.leaderExists, int32(1)) + le.leaderExists.Store(int32(1)) if le.sleeping && len(le.interruptChan) == 0 { le.interruptChan <- struct{}{} } @@ -317,7 +317,7 @@ func (le *leaderElectionSvcImpl) leaderElection() { // If we got here, there is no one that proposed being a leader // that's a better candidate than us. le.beLeader() - atomic.StoreInt32(&le.leaderExists, int32(1)) + le.leaderExists.Store(int32(1)) } // propose sends a leadership proposal message to remote peers @@ -333,7 +333,7 @@ func (le *leaderElectionSvcImpl) follower() { defer le.logger.Debug(le.id, ": Exiting") le.proposals.Clear() - atomic.StoreInt32(&le.leaderExists, int32(0)) + le.leaderExists.Store(int32(0)) le.adapter.ReportMetrics(false) select { case <-time.After(le.config.LeaderAliveThreshold): @@ -384,25 +384,25 @@ func (le *leaderElectionSvcImpl) isAlive(id peerID) bool { } func (le *leaderElectionSvcImpl) isLeaderExists() bool { - return atomic.LoadInt32(&le.leaderExists) == int32(1) + return le.leaderExists.Load() == int32(1) } // IsLeader returns whether this peer is a leader func (le *leaderElectionSvcImpl) IsLeader() bool { - isLeader := atomic.LoadInt32(&le.isLeader) == int32(1) + isLeader := le.isLeader.Load() == int32(1) le.logger.Debug(le.id, ": Returning", isLeader) return isLeader } func (le *leaderElectionSvcImpl) beLeader() { le.logger.Info(le.id, ": Becoming a leader") - atomic.StoreInt32(&le.isLeader, int32(1)) + le.isLeader.Store(int32(1)) le.callback(true) } func (le *leaderElectionSvcImpl) stopBeingLeader() { le.logger.Info(le.id, "Stopped being a leader") - atomic.StoreInt32(&le.isLeader, int32(0)) + le.isLeader.Store(int32(0)) le.callback(false) } @@ -416,14 +416,14 @@ func (le *leaderElectionSvcImpl) shouldStop() bool { } func (le *leaderElectionSvcImpl) isYielding() bool { - return atomic.LoadInt32(&le.yield) == int32(1) + return le.yield.Load() == int32(1) } func (le *leaderElectionSvcImpl) stopYielding() { le.logger.Debug("Stopped yielding") le.Lock() defer le.Unlock() - atomic.StoreInt32(&le.yield, int32(0)) + le.yield.Store(int32(0)) le.yieldTimer.Stop() } @@ -436,14 +436,14 @@ func (le *leaderElectionSvcImpl) Yield() { return } // Turn on the yield flag - atomic.StoreInt32(&le.yield, int32(1)) + le.yield.Store(int32(1)) // Stop being a leader le.stopBeingLeader() // Clear the leader exists flag since it could be that we are the leader - atomic.StoreInt32(&le.leaderExists, int32(0)) + le.leaderExists.Store(int32(0)) // Clear the yield flag in any case afterwards le.yieldTimer = time.AfterFunc(le.config.LeaderAliveThreshold*6, func() { - atomic.StoreInt32(&le.yield, int32(0)) + le.yield.Store(int32(0)) }) } diff --git a/gossip/election/election_test.go b/gossip/election/election_test.go index 946ce3bd040..f33a1f4b128 100644 --- a/gossip/election/election_test.go +++ b/gossip/election/election_test.go @@ -265,11 +265,11 @@ func TestStop(t *testing.T) { // and then are stopped. We count the number of Gossip() invocations they invoke // after they stop, and it should not increase after they are stopped peers := createPeers(0, 3, 2, 1, 0) - var gossipCounter int32 + var gossipCounter atomic.Int32 for i, p := range peers { p.On("Gossip", mock.Anything).Run(func(args mock.Arguments) { msg := args.Get(0).(Msg) - atomic.AddInt32(&gossipCounter, int32(1)) + gossipCounter.Add(int32(1)) for j := range peers { if i == j { continue @@ -283,9 +283,9 @@ func TestStop(t *testing.T) { p.Stop() } time.Sleep(testLeaderAliveThreshold) - gossipCounterAfterStop := atomic.LoadInt32(&gossipCounter) + gossipCounterAfterStop := gossipCounter.Load() time.Sleep(testLeaderAliveThreshold * 5) - require.Equal(t, gossipCounterAfterStop, atomic.LoadInt32(&gossipCounter)) + require.Equal(t, gossipCounterAfterStop, gossipCounter.Load()) } func TestConvergence(t *testing.T) { diff --git a/gossip/gossip/anchor_test.go b/gossip/gossip/anchor_test.go index e53cc497af4..5b1e64812a9 100644 --- a/gossip/gossip/anchor_test.go +++ b/gossip/gossip/anchor_test.go @@ -33,7 +33,7 @@ type peerMock struct { gRGCserv *grpc.Server finishedSignal sync.WaitGroup expectedMsgs2Receive uint32 - msgReceivedCount uint32 + msgReceivedCount atomic.Uint32 msgAssertions []msgInspection t *testing.T } @@ -61,8 +61,8 @@ func (p *peerMock) GossipStream(stream proto.Gossip_GossipStreamServer) error { } p.t.Log("sessionCounter:", sessionCounter, string(p.pkiID), "got msg:", gMsg) sessionCounter++ - atomic.AddUint32(&p.msgReceivedCount, uint32(1)) - if atomic.LoadUint32(&p.msgReceivedCount) == p.expectedMsgs2Receive { + p.msgReceivedCount.Add(uint32(1)) + if p.msgReceivedCount.Load() == p.expectedMsgs2Receive { p.finishedSignal.Done() } } diff --git a/gossip/gossip/channel/channel.go b/gossip/gossip/channel/channel.go index f8e93de9d34..3a035461051 100644 --- a/gossip/gossip/channel/channel.go +++ b/gossip/gossip/channel/channel.go @@ -160,7 +160,7 @@ type gossipChannel struct { memFilter *membershipFilter ledgerHeight uint64 incTime uint64 - leftChannel int32 + leftChannel atomic.Int32 membershipTracker *membershipTracker } @@ -349,7 +349,7 @@ func (gc *gossipChannel) LeaveChannel() { gc.Lock() defer gc.Unlock() - atomic.StoreInt32(&gc.leftChannel, 1) + gc.leftChannel.Store(1) var chaincodes []*proto.Chaincode var height uint64 @@ -362,7 +362,7 @@ func (gc *gossipChannel) LeaveChannel() { } func (gc *gossipChannel) hasLeftChannel() bool { - return atomic.LoadInt32(&gc.leftChannel) == 1 + return gc.leftChannel.Load() == 1 } // GetPeers returns a list of peers with metadata as published by them diff --git a/gossip/gossip/channel/channel_test.go b/gossip/gossip/channel/channel_test.go index f350d883299..3bd94da37c0 100644 --- a/gossip/gossip/channel/channel_test.go +++ b/gossip/gossip/channel/channel_test.go @@ -193,7 +193,7 @@ func (m *receivedMsg) GetConnectionInfo() *protoext.ConnectionInfo { } type gossipAdapterMock struct { - signCallCount uint32 + signCallCount atomic.Uint32 mock.Mock sync.RWMutex } @@ -205,7 +205,7 @@ func (ga *gossipAdapterMock) On(methodName string, arguments ...any) *mock.Call } func (ga *gossipAdapterMock) Sign(msg *proto.GossipMessage) (*protoext.SignedGossipMessage, error) { - atomic.AddUint32(&ga.signCallCount, 1) + ga.signCallCount.Add(1) return protoext.NoopSign(msg) } @@ -478,14 +478,14 @@ func TestLeaveChannel(t *testing.T) { require.Len(t, gc.GetPeers(), 1) // Ensure peer in org1 remained and peer in org2 is skipped require.Equal(t, pkiIDInOrg1, gc.GetPeers()[0].PKIid) - var digestSendTime int32 + var digestSendTime atomic.Int32 var DigestSentWg sync.WaitGroup DigestSentWg.Add(1) hello := createHelloMsg(pkiIDInOrg1) hello.On("Respond", mock.Anything).Run(func(arguments mock.Arguments) { - atomic.AddInt32(&digestSendTime, 1) + digestSendTime.Add(1) // Ensure we only respond with digest before we leave the channel - require.Equal(t, int32(1), atomic.LoadInt32(&digestSendTime)) + require.Equal(t, int32(1), digestSendTime.Load()) DigestSentWg.Done() }) // Wait until we send a hello pull message @@ -1120,14 +1120,14 @@ func TestNoGossipOrSigningWhenEmptyMembership(t *testing.T) { gc := NewGossipChannel(pkiIDInOrg1, orgInChannelA, cs, channelA, adapter, &joinChanMsg{}, disabledMetrics, nil) // We have signed only once at creation time - assert.Equal(t, uint32(1), atomic.LoadUint32(&adapter.signCallCount)) + assert.Equal(t, uint32(1), adapter.signCallCount.Load()) defer gc.Stop() gc.UpdateLedgerHeight(1) // The first time we have membership, so we should gossip and sign gossipedWG.Wait() // So far we have signed twice: Once at creation time, and once before we gossiped - assert.Equal(t, uint32(2), atomic.LoadUint32(&adapter.signCallCount)) + assert.Equal(t, uint32(2), adapter.signCallCount.Load()) // Membership is now empty dynamicMembership.Store(emptyMembership) @@ -1136,14 +1136,14 @@ func TestNoGossipOrSigningWhenEmptyMembership(t *testing.T) { // Wait some time and ensure we do not sign because membership is now empty time.Sleep(conf.PublishStateInfoInterval * 3) // We haven't signed anything - assert.Equal(t, uint32(2), atomic.LoadUint32(&adapter.signCallCount)) + assert.Equal(t, uint32(2), adapter.signCallCount.Load()) assert.Empty(t, gc.Self().GetStateInfo().Properties.Chaincodes) gossipedWG.Add(1) // Now, update chaincodes and check our chaincode information was indeed updated gc.UpdateChaincodes([]*proto.Chaincode{{Name: "mycc"}}) // We should have signed regardless! - assert.Equal(t, uint32(3), atomic.LoadUint32(&adapter.signCallCount)) + assert.Equal(t, uint32(3), adapter.signCallCount.Load()) assert.Equal(t, "mycc", gc.Self().GetStateInfo().Properties.Chaincodes[0].Name) } @@ -1453,22 +1453,22 @@ func TestChannelStop(t *testing.T) { cs := &cryptoService{} cs.On("VerifyBlock", mock.Anything).Return(nil) adapter := new(gossipAdapterMock) - var sendCount int32 + var sendCount atomic.Int32 configureAdapter(adapter, discovery.NetworkMember{PKIid: pkiIDInOrg1}) adapter.On("Send", mock.Anything, mock.Anything).Run(func(mock.Arguments) { - atomic.AddInt32(&sendCount, int32(1)) + sendCount.Add(int32(1)) }) gc := NewGossipChannel(pkiIDInOrg1, orgInChannelA, cs, channelA, adapter, &joinChanMsg{}, disabledMetrics, nil) time.Sleep(time.Second) gc.Stop() - oldCount := atomic.LoadInt32(&sendCount) + oldCount := sendCount.Load() t1 := time.Now() for { if time.Since(t1).Nanoseconds() > (time.Second * 15).Nanoseconds() { t.Fatal("Stop failed") } time.Sleep(time.Second) - newCount := atomic.LoadInt32(&sendCount) + newCount := sendCount.Load() if newCount == oldCount { break } @@ -2336,11 +2336,11 @@ func TestMembershiptrackerStopWhenGCStops(t *testing.T) { waitForHandleMsgChan <- struct{}{} }).Once() - var check uint32 - atomic.StoreUint32(&check, 0) + var check atomic.Uint32 + check.Store(0) logger := util.GetLogger(util.ChannelLogger, adapter.GetConf().ID) logger = logger.(*flogging.FabricLogger).WithOptions(zap.Hooks(func(entry zapcore.Entry) error { - if atomic.LoadUint32(&check) == 1 { + if check.Load() == 1 { if !strings.Contains(entry.Message, "Membership view has changed. peers went offline: [[a]] , peers went online: [[b]] , current view: [[b]]") { return nil } @@ -2364,7 +2364,7 @@ func TestMembershiptrackerStopWhenGCStops(t *testing.T) { }).Once() flogging.ActivateSpec("info") - atomic.StoreUint32(&check, 1) + check.Store(1) <-membershipReported wg.Wait() diff --git a/gossip/gossip/gossip_test.go b/gossip/gossip/gossip_test.go index 148ce4bea97..ea5f6118677 100644 --- a/gossip/gossip/gossip_test.go +++ b/gossip/gossip/gossip_test.go @@ -1383,19 +1383,19 @@ func TestSendByCriteria(t *testing.T) { } f() } - var messagesSent uint32 + var messagesSent atomic.Uint32 go waitForMessage(ackChan2, func() { - atomic.AddUint32(&messagesSent, 1) + messagesSent.Add(1) }) go waitForMessage(ackChan3, func() { - atomic.AddUint32(&messagesSent, 1) + messagesSent.Add(1) }) err = g1.SendByCriteria(msg, criteria) require.Error(t, err) require.Contains(t, err.Error(), "timed out") // Check how many messages were sent. // Only 1 should have been sent - require.Equal(t, uint32(1), atomic.LoadUint32(&messagesSent)) + require.Equal(t, uint32(1), messagesSent.Load()) } func TestIdentityExpiration(t *testing.T) { diff --git a/gossip/privdata/distributor.go b/gossip/privdata/distributor.go index 3430097b542..b7ec3f8e673 100644 --- a/gossip/privdata/distributor.go +++ b/gossip/privdata/distributor.go @@ -386,7 +386,7 @@ func (d *distributorImpl) eligiblePeersOfChannel(routingFilter filter.RoutingFil } func (d *distributorImpl) disseminate(disseminationPlan []*dissemination) error { - var failures uint32 + var failures atomic.Uint32 var wg sync.WaitGroup wg.Add(len(disseminationPlan)) start := time.Now() @@ -396,14 +396,14 @@ func (d *distributorImpl) disseminate(disseminationPlan []*dissemination) error defer d.reportSendDuration(start) err := d.SendByCriteria(dis.msg, dis.criteria) if err != nil { - atomic.AddUint32(&failures, 1) + failures.Add(1) m := dis.msg.GetPrivateData().Payload d.logger.Error("Failed disseminating private RWSet for TxID", m.TxId, ", namespace", m.Namespace, "collection", m.CollectionName, ":", err) } }(dis) } wg.Wait() - failureCount := atomic.LoadUint32(&failures) + failureCount := failures.Load() if failureCount != 0 { return errors.Errorf("Failed disseminating %d out of %d private dissemination plans", failureCount, len(disseminationPlan)) } diff --git a/gossip/state/state_test.go b/gossip/state/state_test.go index d5bc3ec2870..a01c5a2ada2 100644 --- a/gossip/state/state_test.go +++ b/gossip/state/state_test.go @@ -969,7 +969,7 @@ func TestLedgerHeightFromProperties(t *testing.T) { // Returns whether the given networkMember was selected or not wasNetworkMemberSelected := func(t *testing.T, networkMember discovery.NetworkMember) bool { - var wasGivenNetworkMemberSelected int32 + var wasGivenNetworkMemberSelected atomic.Int32 finChan := make(chan struct{}) g := &mocks.GossipMock{} g.On("Send", mock.Anything, mock.Anything).Run(func(arguments mock.Arguments) { @@ -977,7 +977,7 @@ func TestLedgerHeightFromProperties(t *testing.T) { require.NotNil(t, msg.GetStateRequest()) peer := arguments.Get(1).([]*comm.RemotePeer)[0] if bytes.Equal(networkMember.PKIid, peer.PKIID) { - atomic.StoreInt32(&wasGivenNetworkMemberSelected, 1) + wasGivenNetworkMemberSelected.Store(1) } finChan <- struct{}{} }) @@ -1003,7 +1003,7 @@ func TestLedgerHeightFromProperties(t *testing.T) { t.Fatal("Didn't send a request within a timely manner") case <-finChan: } - return atomic.LoadInt32(&wasGivenNetworkMemberSelected) == 1 + return wasGivenNetworkMemberSelected.Load() == 1 } peerWithProperties := discovery.NetworkMember{ diff --git a/internal/pkg/gateway/submit.go b/internal/pkg/gateway/submit.go index d47d2145664..8031b0890e3 100644 --- a/internal/pkg/gateway/submit.go +++ b/internal/pkg/gateway/submit.go @@ -98,7 +98,7 @@ loop: func (gs *Server) broadcastToAll(orderers []*orderer, txn *common.Envelope, waitCh chan<- *gp.ErrorDetail, logger *flogging.FabricLogger) { everyoneSubmitted := make(chan struct{}) - var numFinishedSend uint32 + var numFinishedSend atomic.Uint32 broadcastContext, broadcastCancel := context.WithCancel(context.Background()) defer broadcastCancel() @@ -109,7 +109,7 @@ func (gs *Server) broadcastToAll(orderers []*orderer, txn *common.Envelope, wait defer cancel() response, err := gs.broadcast(ctx, ord, txn) // If I'm the last to submit, notify this - if atomic.AddUint32(&numFinishedSend, 1) == uint32(len(orderers)) { + if numFinishedSend.Add(1) == uint32(len(orderers)) { close(everyoneSubmitted) } if err != nil { diff --git a/msp/cache/second_chance.go b/msp/cache/second_chance.go index 0d5b3c9e47d..32b05e69472 100644 --- a/msp/cache/second_chance.go +++ b/msp/cache/second_chance.go @@ -35,7 +35,7 @@ type cacheItem struct { key string value any // set to 1 when get() is called. set to 0 when victim scan - referenced int32 + referenced atomic.Int32 } func newSecondChanceCache(cacheSize int) *secondChanceCache { @@ -64,7 +64,7 @@ func (cache *secondChanceCache) get(key string) (any, bool) { } // referenced bit is set to true to indicate that this item is recently accessed. - atomic.StoreInt32(&item.referenced, 1) + item.referenced.Store(1) return item.value, true } @@ -75,7 +75,7 @@ func (cache *secondChanceCache) add(key string, value any) { if old, ok := cache.table[key]; ok { old.value = value - atomic.StoreInt32(&old.referenced, 1) + old.referenced.Store(1) return } @@ -96,7 +96,7 @@ func (cache *secondChanceCache) add(key string, value any) { for { // checks whether this item is recently accessed or not victim := cache.items[cache.position] - if atomic.LoadInt32(&victim.referenced) == 0 { + if victim.referenced.Load() == 0 { // a victim is found. delete it, and store the new item here. delete(cache.table, victim.key) cache.table[key] = &item @@ -107,7 +107,7 @@ func (cache *secondChanceCache) add(key string, value any) { // referenced bit is set to false so that this item will be get purged // unless it is accessed until a next victim scan - atomic.StoreInt32(&victim.referenced, 0) + victim.referenced.Store(0) cache.position = (cache.position + 1) % size } }