diff options
Diffstat (limited to 'swarm')
-rw-r--r-- | swarm/network/simulation/simulation.go | 23 | ||||
-rw-r--r-- | swarm/network/simulation/simulation_test.go | 2 | ||||
-rw-r--r-- | swarm/network/stream/delivery_test.go | 10 | ||||
-rw-r--r-- | swarm/network/stream/snapshot_sync_test.go | 4 |
4 files changed, 29 insertions, 10 deletions
diff --git a/swarm/network/simulation/simulation.go b/swarm/network/simulation/simulation.go index 2241dfca2..74f9d98ee 100644 --- a/swarm/network/simulation/simulation.go +++ b/swarm/network/simulation/simulation.go @@ -62,6 +62,8 @@ type Simulation struct { // where all "global" state related to the service should be kept. // All cleanups needed for constructed service and any other constructed // objects should ne provided in a single returned cleanup function. +// Returned cleanup function will be called by Close function +// after network shutdown. type ServiceFunc func(ctx *adapters.ServiceContext, bucket *sync.Map) (s node.Service, cleanup func(), err error) // New creates a new Simulation instance with new @@ -161,6 +163,18 @@ var maxParallelCleanups = 10 // simulation. func (s *Simulation) Close() { close(s.done) + + // Close all connections before calling the Network Shutdown. + // It is possible that p2p.Server.Stop will block if there are + // existing connections. + for _, c := range s.Net.Conns { + if c.Up { + s.Net.Disconnect(c.One, c.Other) + } + } + s.shutdownWG.Wait() + s.Net.Shutdown() + sem := make(chan struct{}, maxParallelCleanups) s.mu.RLock() cleanupFuncs := make([]func(), len(s.cleanupFuncs)) @@ -170,16 +184,19 @@ func (s *Simulation) Close() { } } s.mu.RUnlock() + var cleanupWG sync.WaitGroup for _, cleanup := range cleanupFuncs { - s.shutdownWG.Add(1) + cleanupWG.Add(1) sem <- struct{}{} go func(cleanup func()) { - defer s.shutdownWG.Done() + defer cleanupWG.Done() defer func() { <-sem }() cleanup() }(cleanup) } + cleanupWG.Wait() + if s.httpSrv != nil { ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() @@ -189,8 +206,6 @@ func (s *Simulation) Close() { } close(s.runC) } - s.shutdownWG.Wait() - s.Net.Shutdown() } // Done returns a channel that is closed when the simulation diff --git a/swarm/network/simulation/simulation_test.go b/swarm/network/simulation/simulation_test.go index 803e0499a..8576732c9 100644 --- a/swarm/network/simulation/simulation_test.go +++ b/swarm/network/simulation/simulation_test.go @@ -68,7 +68,7 @@ func TestRun(t *testing.T) { defer cancel() r := sim.Run(ctx, func(ctx context.Context, sim *Simulation) error { - time.Sleep(100 * time.Millisecond) + time.Sleep(time.Second) return nil }) diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go index ae007e5b0..972cc859a 100644 --- a/swarm/network/stream/delivery_test.go +++ b/swarm/network/stream/delivery_test.go @@ -393,6 +393,11 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck return err } + log.Debug("Waiting for kademlia") + if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { + return err + } + //each of the nodes (except pivot node) subscribes to the stream of the next node for j, node := range nodeIDs[0 : nodes-1] { sid := nodeIDs[j+1] @@ -424,11 +429,6 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck } }() - log.Debug("Waiting for kademlia") - if _, err := sim.WaitTillHealthy(ctx, 2); err != nil { - return err - } - log.Debug("Watching for disconnections") disconnections := sim.PeerEvents( context.Background(), diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go index 2dfc5898f..6acab50af 100644 --- a/swarm/network/stream/snapshot_sync_test.go +++ b/swarm/network/stream/snapshot_sync_test.go @@ -246,6 +246,8 @@ func testSyncingViaGlobalSync(t *testing.T, chunkCount int, nodeCount int) { if err != nil { log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) localSuccess = false + // Do not get crazy with logging the warn message + time.Sleep(500 * time.Millisecond) } else { log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) } @@ -426,6 +428,8 @@ func testSyncingViaDirectSubscribe(chunkCount int, nodeCount int) error { if err != nil { log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id)) localSuccess = false + // Do not get crazy with logging the warn message + time.Sleep(500 * time.Millisecond) } else { log.Debug(fmt.Sprintf("Chunk %s IS FOUND for id %s", chunk, id)) } |