Skip to content

Commit

Permalink
les: use atomic type (#27168)
Browse files Browse the repository at this point in the history
  • Loading branch information
stephenfire authored Apr 26, 2023
1 parent f8aa623 commit 25f9977
Show file tree
Hide file tree
Showing 5 changed files with 21 additions and 25 deletions.
5 changes: 2 additions & 3 deletions les/client_handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
"math/big"
"math/rand"
"sync"
"sync/atomic"
"time"

"github.com/ethereum/go-ethereum/common"
Expand Down Expand Up @@ -144,8 +143,8 @@ func (h *clientHandler) handle(p *serverPeer, noInitAnnounce bool) error {
}

// Mark the peer starts to be served.
atomic.StoreUint32(&p.serving, 1)
defer atomic.StoreUint32(&p.serving, 0)
p.serving.Store(true)
defer p.serving.Store(false)

// Spawn a main loop to handle all incoming messages.
for {
Expand Down
26 changes: 13 additions & 13 deletions les/peer.go
Original file line number Diff line number Diff line change
Expand Up @@ -121,13 +121,13 @@ type peerCommons struct {
*p2p.Peer
rw p2p.MsgReadWriter

id string // Peer identity.
version int // Protocol version negotiated.
network uint64 // Network ID being on.
frozen uint32 // Flag whether the peer is frozen.
announceType uint64 // New block announcement type.
serving uint32 // The status indicates the peer is served.
headInfo blockInfo // Last announced block information.
id string // Peer identity.
version int // Protocol version negotiated.
network uint64 // Network ID being on.
frozen atomic.Bool // Flag whether the peer is frozen.
announceType uint64 // New block announcement type.
serving atomic.Bool // The status indicates the peer is served.
headInfo blockInfo // Last announced block information.

// Background task queue for caching peer tasks and executing in order.
sendQueue *utils.ExecQueue
Expand All @@ -143,7 +143,7 @@ type peerCommons struct {
// isFrozen returns true if the client is frozen or the server has put our
// client in frozen state
func (p *peerCommons) isFrozen() bool {
return atomic.LoadUint32(&p.frozen) != 0
return p.frozen.Load()
}

// canQueue returns an indicator whether the peer can queue an operation.
Expand Down Expand Up @@ -398,15 +398,15 @@ func (p *serverPeer) rejectUpdate(size uint64) bool {
// freeze processes Stop messages from the given server and set the status as
// frozen.
func (p *serverPeer) freeze() {
if atomic.CompareAndSwapUint32(&p.frozen, 0, 1) {
if p.frozen.CompareAndSwap(false, true) {
p.sendQueue.Clear()
}
}

// unfreeze processes Resume messages from the given server and set the status
// as unfrozen.
func (p *serverPeer) unfreeze() {
atomic.StoreUint32(&p.frozen, 0)
p.frozen.Store(false)
}

// sendRequest send a request to the server based on the given message type
Expand Down Expand Up @@ -823,11 +823,11 @@ func (p *clientPeer) freeze() {
if p.version < lpv3 {
// if Stop/Resume is not supported then just drop the peer after setting
// its frozen status permanently
atomic.StoreUint32(&p.frozen, 1)
p.frozen.Store(true)
p.Peer.Disconnect(p2p.DiscUselessPeer)
return
}
if atomic.SwapUint32(&p.frozen, 1) == 0 {
if !p.frozen.Swap(true) {
go func() {
p.sendStop()
time.Sleep(freezeTimeBase + time.Duration(rand.Int63n(int64(freezeTimeRandom))))
Expand All @@ -840,7 +840,7 @@ func (p *clientPeer) freeze() {
time.Sleep(freezeCheckPeriod)
continue
}
atomic.StoreUint32(&p.frozen, 0)
p.frozen.Store(false)
p.sendResume(bufValue)
return
}
Expand Down
5 changes: 2 additions & 3 deletions les/server_handler.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,7 +19,6 @@ package les
import (
"errors"
"sync"
"sync/atomic"
"time"

"github.com/ethereum/go-ethereum/common"
Expand Down Expand Up @@ -164,8 +163,8 @@ func (h *serverHandler) handle(p *clientPeer) error {
}()

// Mark the peer as being served.
atomic.StoreUint32(&p.serving, 1)
defer atomic.StoreUint32(&p.serving, 0)
p.serving.Store(true)
defer p.serving.Store(false)

// Spawn a main loop to handle all incoming messages.
for {
Expand Down
7 changes: 3 additions & 4 deletions les/test_helper.go
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ import (
"crypto/rand"
"fmt"
"math/big"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -380,7 +379,7 @@ func newTestPeerPair(name string, version int, server *serverHandler, client *cl
return nil, nil, fmt.Errorf("failed to establish protocol connection %v", err)
default:
}
if atomic.LoadUint32(&peer1.serving) == 1 && atomic.LoadUint32(&peer2.serving) == 1 {
if peer1.serving.Load() && peer2.serving.Load() {
break
}
time.Sleep(50 * time.Millisecond)
Expand Down Expand Up @@ -441,7 +440,7 @@ func (client *testClient) newRawPeer(t *testing.T, name string, version int, rec
return nil, nil, nil
default:
}
if atomic.LoadUint32(&peer.serving) == 1 {
if peer.serving.Load() {
break
}
time.Sleep(50 * time.Millisecond)
Expand Down Expand Up @@ -505,7 +504,7 @@ func (server *testServer) newRawPeer(t *testing.T, name string, version int) (*t
return nil, nil, nil
default:
}
if atomic.LoadUint32(&peer.serving) == 1 {
if peer.serving.Load() {
break
}
time.Sleep(50 * time.Millisecond)
Expand Down
3 changes: 1 addition & 2 deletions les/ulc_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,6 @@ import (
"crypto/rand"
"fmt"
"net"
"sync/atomic"
"testing"
"time"

Expand Down Expand Up @@ -136,7 +135,7 @@ func connect(server *serverHandler, serverId enode.ID, client *clientHandler, pr
return nil, nil, fmt.Errorf("failed to establish protocol connection %v", err)
default:
}
if atomic.LoadUint32(&peer1.serving) == 1 && atomic.LoadUint32(&peer2.serving) == 1 {
if peer1.serving.Load() && peer2.serving.Load() {
break
}
time.Sleep(50 * time.Millisecond)
Expand Down

0 comments on commit 25f9977

Please sign in to comment.