From 2605f59bc0e522c73c14c65471119703ec5bf10c Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Wed, 11 Apr 2018 14:28:44 +0200 Subject: [PATCH 1/3] swarm/network/stream: fix TestSyncerSimulation deadlock This change fixes the synchronization between simulations network shutdown and calling of rpc subscription Unsubscribe method. If the unsubscribe is called after net shutdown, rpc client is blocking. This change also addresses the panic in benchmarkDeliveryFromNodes function, when executing benchmarks from tests. --- swarm/network/stream/delivery_test.go | 30 +++++++++++++++--- swarm/network/stream/intervals_test.go | 14 ++++++-- swarm/network/stream/snapshot_sync_test.go | 37 ++++++++++++++++++---- swarm/network/stream/syncer_test.go | 14 ++++++-- swarm/network/stream/testing/testing.go | 22 ++++++++++--- 5 files changed, 99 insertions(+), 18 deletions(-) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index 0d92b040da..d4e31343c5 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -22,6 +22,7 @@ import ( crand "crypto/rand" "fmt" "io" + "sync" "testing" "time" @@ -318,7 +319,11 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } sim, teardown, err := streamTesting.NewSimulation(conf) - defer teardown() + var rpcSubscriptionsWg sync.WaitGroup + defer func() { + rpcSubscriptionsWg.Wait() + teardown() + }() if err != nil { t.Fatal(err.Error()) } @@ -347,6 +352,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck errc := make(chan error, 1) waitPeerErrC = make(chan error) quitC := make(chan struct{}) + defer close(quitC) action := func(ctx context.Context) error { // each node Subscribes to each other's swarmChunkServerStreamName @@ -369,10 +375,15 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck for j := 0; j < nodes-1; j++ { id := sim.IDs[j] err := sim.CallClient(id, func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(id, client, errc, quitC) + doneC, err := streamTesting.WatchDisconnections(id, client, errc, quitC) if err != nil { return err } + rpcSubscriptionsWg.Add(1) + go func() { + <-doneC + rpcSubscriptionsWg.Done() + }() ctx, cancel := context.WithTimeout(ctx, 1*time.Second) defer cancel() sid := sim.IDs[j+1] @@ -475,6 +486,8 @@ func BenchmarkDeliveryFromNodesWithCheck(b *testing.B) { func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skipCheck bool) { defaultSkipCheck = skipCheck toAddr = network.NewAddrFromNodeID + createStoreFunc = createTestLocalStorageFromSim + registries = make(map[discover.NodeID]*TestRegistry) timeout := 300 * time.Second ctx, cancel := context.WithTimeout(context.Background(), timeout) @@ -489,7 +502,11 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip EnableMsgEvents: false, } sim, teardown, err := streamTesting.NewSimulation(conf) - defer teardown() + var rpcSubscriptionsWg sync.WaitGroup + defer func() { + rpcSubscriptionsWg.Wait() + teardown() + }() if err != nil { b.Fatal(err.Error()) } @@ -539,10 +556,15 @@ func benchmarkDeliveryFromNodes(b *testing.B, nodes, conns, chunkCount int, skip for j := 0; j < nodes-1; j++ { id := sim.IDs[j] err = sim.CallClient(id, func(client *rpc.Client) error { - err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC) + doneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC) if err != nil { return err } + rpcSubscriptionsWg.Add(1) + go func() { + <-doneC + rpcSubscriptionsWg.Done() + }() ctx, cancel := context.WithTimeout(ctx, 1*time.Second) defer cancel() sid := sim.IDs[j+1] // the upstream peer's id diff --git a/swarm/network/stream/intervals_test.go b/swarm/network/stream/intervals_test.go index d0de039939..c65ac27fc4 100644 --- a/swarm/network/stream/intervals_test.go +++ b/swarm/network/stream/intervals_test.go @@ -22,6 +22,7 @@ import ( "encoding/binary" "fmt" "io" + "sync" "testing" "time" @@ -90,7 +91,11 @@ func testIntervals(t *testing.T, live bool, history *Range) { } sim, teardown, err := streamTesting.NewSimulation(conf) - defer teardown() + var rpcSubscriptionsWg sync.WaitGroup + defer func() { + rpcSubscriptionsWg.Wait() + teardown() + }() if err != nil { t.Fatal(err) } @@ -136,10 +141,15 @@ func testIntervals(t *testing.T, live bool, history *Range) { sid := sim.IDs[0] - err := streamTesting.WatchDisconnections(id, client, errc, quitC) + doneC, err := streamTesting.WatchDisconnections(id, client, errc, quitC) if err != nil { return err } + rpcSubscriptionsWg.Add(1) + go func() { + <-doneC + rpcSubscriptionsWg.Done() + }() ctx, cancel := context.WithTimeout(ctx, 100*time.Second) defer cancel() diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index fb54641c67..20dc13799c 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -199,15 +199,20 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error { if err != nil { return err } + var rpcSubscriptionsWg sync.WaitGroup //do cleanup after test is terminated defer func() { + // close quitC channel to signall all goroutines to clanup + // before calling simulation network shutdown. + close(quitC) + //wait for all rpc subscriptions to unsubscribe + rpcSubscriptionsWg.Wait() //shutdown the snapshot network net.Shutdown() //after the test, clean up local stores initialized with createLocalStoreForId localStoreCleanup() //finally clear all data directories datadirsCleanup() - close(quitC) }() //get the nodes of the network nodes := net.GetNodes() @@ -290,7 +295,15 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error { return err } - watchSubscriptionEvents(ctx, id, client, errc) + wsDoneC := watchSubscriptionEvents(ctx, id, client, errc, quitC) + // doneC is nil, the error happened which is sent to errc channel, already + if wsDoneC == nil { + continue + } + go func() { + <-wsDoneC + rpcSubscriptionsWg.Done() + }() if log.Lvl(*loglevel) >= log.LvlTrace { //this will print the kademlia tables of all nodes @@ -306,10 +319,15 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error { log.Debug(kt) } //watch for peers disconnecting - err = streamTesting.WatchDisconnections(id, client, disconnectC, quitC) + wdDoneC, err := streamTesting.WatchDisconnections(id, client, disconnectC, quitC) if err != nil { return err } + rpcSubscriptionsWg.Add(1) + go func() { + <-wdDoneC + rpcSubscriptionsWg.Done() + }() //start syncing! err = client.CallContext(ctx, nil, "stream_startSyncing") if err != nil { @@ -589,7 +607,7 @@ func initNetWithSnapshot(nodeCount int) (*simulations.Network, error) { //we want to wait for subscriptions to be established before uploading to test //that live syncing is working correctly -func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, errc chan error) { +func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) (doneC <-chan struct{}) { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { @@ -597,11 +615,18 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp errc <- fmt.Errorf("error getting peer events for node %v: %s", id, err) return } + c := make(chan struct{}) go func() { - defer sub.Unsubscribe() + defer func() { + log.Trace("watch subscription events: unsubscribe", "id", id) + sub.Unsubscribe() + close(c) + }() for { select { + case <-quitC: + return case <-ctx.Done(): errc <- ctx.Err() return @@ -618,7 +643,7 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp } } }() - return + return c } //create a local store for the given node diff --git a/swarm/network/stream/syncer_test.go b/swarm/network/stream/syncer_test.go index c2f679e594..4e443cbda4 100644 --- a/swarm/network/stream/syncer_test.go +++ b/swarm/network/stream/syncer_test.go @@ -22,6 +22,7 @@ import ( "fmt" "io" "math" + "sync" "testing" "time" @@ -69,7 +70,11 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck // create simulation network with the config sim, teardown, err := streamTesting.NewSimulation(conf) - defer teardown() + var rpcSubscriptionsWg sync.WaitGroup + defer func() { + rpcSubscriptionsWg.Wait() + teardown() + }() if err != nil { t.Fatal(err.Error()) } @@ -154,10 +159,15 @@ func testSyncBetweenNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck id := sim.IDs[j] err := sim.CallClient(id, func(client *rpc.Client) error { // report disconnect events to the error channel cos peers should not disconnect - err := streamTesting.WatchDisconnections(id, client, errc, quitC) + doneC, err := streamTesting.WatchDisconnections(id, client, errc, quitC) if err != nil { return err } + rpcSubscriptionsWg.Add(1) + go func() { + <-doneC + rpcSubscriptionsWg.Done() + }() ctx, cancel := context.WithTimeout(ctx, 1*time.Second) defer cancel() // start syncing, i.e., subscribe to upstream peers po 1 bin diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index 4367df8e35..b19ba4a6ee 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -143,7 +143,9 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { DefaultService: defaultService, }) teardown := func() { + log.Trace("simulation: teardown adapter") adapterTeardown() + log.Trace("simulation: teardown net") net.Shutdown() } ids := make([]discover.NodeID, nodes) @@ -162,8 +164,11 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { // set nodes number of Stores available stores, storeTeardown, err := SetStores(addrs...) teardown = func() { + log.Trace("simulation: teardown net") net.Shutdown() + log.Trace("simulation: teardown adapter") adapterTeardown() + log.Trace("simulation: teardown store") storeTeardown() } if err != nil { @@ -216,14 +221,23 @@ func (s *Simulation) Run(ctx context.Context, conf *RunConfig) (*simulations.Ste return result, nil } -func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) error { +// WatchDisconnections subscribes to admin peerEvents and sends peer event drop +// errors to the errc channel. Channel quitC signals the termination of the event loop. +// Returned doneC will be closed after the rpc subscription is unsubscribed, +// signaling that simulations network is safe to shutdown. +func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error, quitC chan struct{}) (doneC <-chan struct{}, err error) { events := make(chan *p2p.PeerEvent) sub, err := client.Subscribe(context.Background(), "admin", events, "peerEvents") if err != nil { - return fmt.Errorf("error getting peer events for node %v: %s", id, err) + return nil, fmt.Errorf("error getting peer events for node %v: %s", id, err) } + c := make(chan struct{}) go func() { - defer sub.Unsubscribe() + defer func() { + log.Trace("watch disconnections: unsubscribe", "id", id) + sub.Unsubscribe() + close(c) + }() for { select { case <-quitC: @@ -247,7 +261,7 @@ func WatchDisconnections(id discover.NodeID, client *rpc.Client, errc chan error } } }() - return nil + return c, nil } func Trigger(d time.Duration, quitC chan struct{}, ids ...discover.NodeID) chan discover.NodeID { From b3be75c537dbb9aa1729ee08abb9768c462c966c Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Thu, 12 Apr 2018 11:12:05 +0200 Subject: [PATCH 2/3] swarm/network/stream: add missing WaitGroup.Add and fix errc send blocking --- swarm/network/stream/snapshot_sync_test.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 20dc13799c..a498013e3c 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -300,6 +300,7 @@ func runSyncTest(chunkCount int, nodeCount int, live bool, history bool) error { if wsDoneC == nil { continue } + rpcSubscriptionsWg.Add(1) go func() { <-wsDoneC rpcSubscriptionsWg.Done() @@ -628,7 +629,10 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp case <-quitC: return case <-ctx.Done(): - errc <- ctx.Err() + select { + case errc <- ctx.Err(): + case <-quitC: + } return case e := <-events: //just catch SubscribeMsg @@ -637,7 +641,10 @@ func watchSubscriptionEvents(ctx context.Context, id discover.NodeID, client *rp } case err := <-sub.Err(): if err != nil { - errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err) + select { + case errc <- fmt.Errorf("error getting peer events for node %v: %v", id, err): + case <-quitC: + } return } } From 9f88c689ab2e0c855cdddcd1575ec1d1fea57a02 Mon Sep 17 00:00:00 2001 From: Janos Guljas Date: Thu, 12 Apr 2018 13:11:47 +0200 Subject: [PATCH 3/3] swarm/netork/stream/testing: remove trace log lines in teardown functions --- swarm/network/stream/testing/testing.go | 5 ----- 1 file changed, 5 deletions(-) diff --git a/swarm/network/stream/testing/testing.go b/swarm/network/stream/testing/testing.go index b19ba4a6ee..29d8026a22 100644 --- a/swarm/network/stream/testing/testing.go +++ b/swarm/network/stream/testing/testing.go @@ -143,9 +143,7 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { DefaultService: defaultService, }) teardown := func() { - log.Trace("simulation: teardown adapter") adapterTeardown() - log.Trace("simulation: teardown net") net.Shutdown() } ids := make([]discover.NodeID, nodes) @@ -164,11 +162,8 @@ func NewSimulation(conf *RunConfig) (*Simulation, func(), error) { // set nodes number of Stores available stores, storeTeardown, err := SetStores(addrs...) teardown = func() { - log.Trace("simulation: teardown net") net.Shutdown() - log.Trace("simulation: teardown adapter") adapterTeardown() - log.Trace("simulation: teardown store") storeTeardown() } if err != nil {