package swarm import ( "context" "fmt" "sync" "testing" "time" "github.com/libp2p/go-libp2p-core/peer" ) func getMockDialFunc() (dialWorkerFunc, func(), context.Context, <-chan struct{}) { dfcalls := make(chan struct{}, 512) // buffer it large enough that we won't care dialctx, cancel := context.WithCancel(context.Background()) ch := make(chan struct{}) f := func(ctx context.Context, p peer.ID, reqch <-chan dialRequest) error { dfcalls <- struct{}{} go func() { defer cancel() for { select { case req, ok := <-reqch: if !ok { return } select { case <-ch: req.resch <- dialResponse{conn: new(Conn)} case <-ctx.Done(): req.resch <- dialResponse{err: ctx.Err()} return } case <-ctx.Done(): return } } }() return nil } o := new(sync.Once) return f, func() { o.Do(func() { close(ch) }) }, dialctx, dfcalls } func TestBasicDialSync(t *testing.T) { df, done, _, callsch := getMockDialFunc() dsync := newDialSync(df) p := peer.ID("testpeer") ctx := context.Background() finished := make(chan struct{}) go func() { _, err := dsync.DialLock(ctx, p) if err != nil { t.Error(err) } finished <- struct{}{} }() go func() { _, err := dsync.DialLock(ctx, p) if err != nil { t.Error(err) } finished <- struct{}{} }() // short sleep just to make sure we've moved around in the scheduler time.Sleep(time.Millisecond * 20) done() <-finished <-finished if len(callsch) > 1 { t.Fatal("should only have called dial func once!") } } func TestDialSyncCancel(t *testing.T) { df, done, _, dcall := getMockDialFunc() dsync := newDialSync(df) p := peer.ID("testpeer") ctx1, cancel1 := context.WithCancel(context.Background()) finished := make(chan struct{}) go func() { _, err := dsync.DialLock(ctx1, p) if err != ctx1.Err() { t.Error("should have gotten context error") } finished <- struct{}{} }() // make sure the above makes it through the wait code first select { case <-dcall: case <-time.After(time.Second): t.Fatal("timed out waiting for dial to start") } // Add a second dialwait in so two actors are waiting on the same dial go func() { _, err := dsync.DialLock(context.Background(), p) if err != nil { t.Error(err) } finished <- struct{}{} }() time.Sleep(time.Millisecond * 20) // cancel the first dialwait, it should not affect the second at all cancel1() select { case <-finished: case <-time.After(time.Second): t.Fatal("timed out waiting for wait to exit") } // short sleep just to make sure we've moved around in the scheduler time.Sleep(time.Millisecond * 20) done() <-finished } func TestDialSyncAllCancel(t *testing.T) { df, done, dctx, _ := getMockDialFunc() dsync := newDialSync(df) p := peer.ID("testpeer") ctx1, cancel1 := context.WithCancel(context.Background()) finished := make(chan struct{}) go func() { _, err := dsync.DialLock(ctx1, p) if err != ctx1.Err() { t.Error("should have gotten context error") } finished <- struct{}{} }() // Add a second dialwait in so two actors are waiting on the same dial go func() { _, err := dsync.DialLock(ctx1, p) if err != ctx1.Err() { t.Error("should have gotten context error") } finished <- struct{}{} }() cancel1() for i := 0; i < 2; i++ { select { case <-finished: case <-time.After(time.Second): t.Fatal("timed out waiting for wait to exit") } } // the dial should have exited now select { case <-dctx.Done(): case <-time.After(time.Second): t.Fatal("timed out waiting for dial to return") } // should be able to successfully dial that peer again done() _, err := dsync.DialLock(context.Background(), p) if err != nil { t.Fatal(err) } } func TestFailFirst(t *testing.T) { var count int f := func(ctx context.Context, p peer.ID, reqch <-chan dialRequest) error { go func() { for { select { case req, ok := <-reqch: if !ok { return } if count > 0 { req.resch <- dialResponse{conn: new(Conn)} } else { req.resch <- dialResponse{err: fmt.Errorf("gophers ate the modem")} } count++ case <-ctx.Done(): return } } }() return nil } ds := newDialSync(f) p := peer.ID("testing") ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) defer cancel() _, err := ds.DialLock(ctx, p) if err == nil { t.Fatal("expected gophers to have eaten the modem") } c, err := ds.DialLock(ctx, p) if err != nil { t.Fatal(err) } if c == nil { t.Fatal("should have gotten a 'real' conn back") } } func TestStressActiveDial(t *testing.T) { ds := newDialSync(func(ctx context.Context, p peer.ID, reqch <-chan dialRequest) error { go func() { for { select { case req, ok := <-reqch: if !ok { return } req.resch <- dialResponse{} case <-ctx.Done(): return } } }() return nil }) wg := sync.WaitGroup{} pid := peer.ID("foo") makeDials := func() { for i := 0; i < 10000; i++ { ds.DialLock(context.Background(), pid) } wg.Done() } for i := 0; i < 100; i++ { wg.Add(1) go makeDials() } wg.Wait() } func TestDialSelf(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() self := peer.ID("ABC") s := NewSwarm(ctx, self, nil, nil) defer s.Close() // this should fail _, err := s.dsync.DialLock(ctx, self) if err != ErrDialToSelf { t.Fatal("expected error from self dial") } // do it twice to make sure we get a new active dial object that fails again _, err = s.dsync.DialLock(ctx, self) if err != ErrDialToSelf { t.Fatal("expected error from self dial") } }