Skip to content

Commit

Permalink
wait for peers in wantmanager to all appear
Browse files Browse the repository at this point in the history
License: MIT
Signed-off-by: Jeromy <jeromyj@gmail.com>
  • Loading branch information
whyrusleeping committed Feb 8, 2016
1 parent e9cc395 commit c73da84
Showing 2 changed files with 29 additions and 2 deletions.
13 changes: 13 additions & 0 deletions exchange/bitswap/bitswap_test.go
Original file line number Diff line number Diff line change
@@ -158,6 +158,19 @@ func PerformDistributionTest(t *testing.T, numInstances, numBlocks int) {

t.Log("Give the blocks to the first instance")

nump := len(instances) - 1
// assert we're properly connected
for _, inst := range instances {
peers := inst.Exchange.wm.ConnectedPeers()
for i := 0; i < 10 && len(peers) != nump; i++ {
time.Sleep(time.Millisecond * 50)
peers = inst.Exchange.wm.ConnectedPeers()
}
if len(peers) != nump {
t.Fatal("not enough peers connected to instance")
}
}

var blkeys []key.Key
first := instances[0]
for _, b := range blocks {
18 changes: 16 additions & 2 deletions exchange/bitswap/wantmanager.go
Original file line number Diff line number Diff line change
@@ -16,8 +16,9 @@ import (
type WantManager struct {
// sync channels for Run loop
incoming chan []*bsmsg.Entry
connect chan peer.ID // notification channel for new peers connecting
disconnect chan peer.ID // notification channel for peers disconnecting
connect chan peer.ID // notification channel for new peers connecting
disconnect chan peer.ID // notification channel for peers disconnecting
peerReqs chan chan []peer.ID // channel to request connected peers on

// synchronized by Run loop, only touch inside there
peers map[peer.ID]*msgQueue
@@ -32,6 +33,7 @@ func NewWantManager(ctx context.Context, network bsnet.BitSwapNetwork) *WantMana
incoming: make(chan []*bsmsg.Entry, 10),
connect: make(chan peer.ID, 10),
disconnect: make(chan peer.ID, 10),
peerReqs: make(chan chan []peer.ID),
peers: make(map[peer.ID]*msgQueue),
wl: wantlist.NewThreadSafe(),
network: network,
@@ -88,6 +90,12 @@ func (pm *WantManager) addEntries(ks []key.Key, cancel bool) {
}
}

func (pm *WantManager) ConnectedPeers() []peer.ID {
resp := make(chan []peer.ID)
pm.peerReqs <- resp
return <-resp
}

func (pm *WantManager) SendBlock(ctx context.Context, env *engine.Envelope) {
// Blocks need to be sent synchronously to maintain proper backpressure
// throughout the network stack
@@ -242,6 +250,12 @@ func (pm *WantManager) Run() {
pm.startPeerHandler(p)
case p := <-pm.disconnect:
pm.stopPeerHandler(p)
case req := <-pm.peerReqs:
var peers []peer.ID
for p := range pm.peers {
peers = append(peers, p)
}
req <- peers
case <-pm.ctx.Done():
return
}

0 comments on commit c73da84

Please sign in to comment.