From 8cde53c8f11c4f5aa7ad656cd93acb30af19615e Mon Sep 17 00:00:00 2001 From: Alex Grintsvayg Date: Wed, 13 Jun 2018 12:45:47 -0400 Subject: [PATCH] fix some stoppers --- Gopkg.lock | 14 +++++++++++- dht/dht.go | 54 ++++++++++++++------------------------------ dht/dht_test.go | 25 ++++++++------------ dht/node.go | 31 +++++++++++++------------ dht/node_finder.go | 35 ++++++++++++++-------------- dht/routing_table.go | 35 ++++++++-------------------- 6 files changed, 84 insertions(+), 110 deletions(-) diff --git a/Gopkg.lock b/Gopkg.lock index 224395f..52d1dc2 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -73,6 +73,12 @@ packages = ["."] revision = "3287d94d4c6a48a63e16fffaabf27ab20203af2a" +[[projects]] + name = "github.com/gorilla/websocket" + packages = ["."] + revision = "ea4d1f681babbce9545c9c5f3d5194a789c89f5b" + version = "v1.2.0" + [[projects]] branch = "master" name = "github.com/hashicorp/errwrap" @@ -152,7 +158,7 @@ "stopOnce", "util" ] - revision = "2a6ea528bdd66de4f3c707304e26f69dcf003909" + revision = "f0762e9c57d41be10cb83edc4e9b7a7ce0891519" [[projects]] branch = "master" @@ -166,6 +172,12 @@ revision = "eac804ceef194db2da6ee80c728d7658c8c805ff" version = "v1.0.6" +[[projects]] + name = "github.com/nlopes/slack" + packages = ["."] + revision = "8ab4d0b364ef1e9af5d102531da20d5ec902b6c4" + version = "v0.2.0" + [[projects]] branch = "master" name = "github.com/sean-/seed" diff --git a/dht/dht.go b/dht/dht.go index e954e2e..bc069de 100644 --- a/dht/dht.go +++ b/dht/dht.go @@ -134,20 +134,12 @@ func (dht *DHT) join() { } // now call iterativeFind on yourself - nf := newContactFinder(dht.node, dht.node.id, false) - // stop if dht is stopped - go func() { - <-dht.stop.Ch() - if nf != nil { - nf.Cancel() - } - }() - _, err := nf.Find() + _, _, err := FindContacts(dht.node, dht.node.id, false, dht.stop.Ch()) if err != nil { log.Errorf("[%s] join: %s", dht.node.id.HexShort(), err.Error()) } - // TODO: after joining, refresh all the buckets all buckets further away than our closest neighbor + // TODO: after joining, refresh all buckets further away than our closest neighbor // http://xlattice.sourceforge.net/components/protocol/kademlia/specs.html#join } @@ -162,20 +154,12 @@ func (dht *DHT) Start() error { if err != nil { return err } - //Perform join in the background - dht.stop.Add(1) - go func() { - defer dht.stop.Done() - dht.join() - log.Debugf("[%s] DHT ready on %s (%d nodes found during join)", - dht.node.id.HexShort(), dht.contact.Addr().String(), dht.node.rt.Count()) - //Reannouncer can only be launched after join is complete. - dht.stop.Add(1) - go func() { - defer dht.stop.Done() - dht.startReannouncer() - }() - }() + + dht.join() + log.Debugf("[%s] DHT ready on %s (%d nodes found during join)", + dht.node.id.HexShort(), dht.contact.Addr().String(), dht.node.rt.Count()) + + go dht.startReannouncer() return nil } @@ -215,34 +199,32 @@ func (dht *DHT) Ping(addr string) error { // Get returns the list of nodes that have the blob for the given hash func (dht *DHT) Get(hash Bitmap) ([]Contact, error) { - nf := newContactFinder(dht.node, hash, true) - res, err := nf.Find() + contacts, found, err := FindContacts(dht.node, hash, true, dht.stop.Ch()) if err != nil { return nil, err } - if res.Found { - return res.Contacts, nil + if found { + return contacts, nil } return nil, nil } // Announce announces to the DHT that this node has the blob for the given hash func (dht *DHT) Announce(hash Bitmap) error { - nf := newContactFinder(dht.node, hash, false) - res, err := nf.Find() + contacts, _, err := FindContacts(dht.node, hash, false, dht.stop.Ch()) if err != nil { return err } // if we found less than K contacts, or current node is closer than farthest contact - if len(res.Contacts) < bucketSize || dht.node.id.Xor(hash).Less(res.Contacts[bucketSize-1].ID.Xor(hash)) { + if len(contacts) < bucketSize || dht.node.id.Xor(hash).Less(contacts[bucketSize-1].ID.Xor(hash)) { // pop last contact, and self-store instead - res.Contacts[bucketSize-1] = dht.contact + contacts[bucketSize-1] = dht.contact } wg := &sync.WaitGroup{} - for _, c := range res.Contacts { + for _, c := range contacts { wg.Add(1) go func(c Contact) { dht.storeOnNode(hash, c) @@ -271,7 +253,8 @@ func (dht *DHT) startReannouncer() { dht.stop.Add(1) go func(bm Bitmap) { defer dht.stop.Done() - if err := dht.Announce(bm); err != nil { + err := dht.Announce(bm) + if err != nil { log.Error("error re-announcing bitmap - ", err) } }(h) @@ -282,9 +265,6 @@ func (dht *DHT) startReannouncer() { } func (dht *DHT) storeOnNode(hash Bitmap, c Contact) { - dht.stop.Add(1) - defer dht.stop.Done() - // self-store if dht.contact.Equals(c) { dht.node.Store(hash, c) diff --git a/dht/dht_test.go b/dht/dht_test.go index 5b4bb00..15ab99b 100644 --- a/dht/dht_test.go +++ b/dht/dht_test.go @@ -16,26 +16,24 @@ func TestNodeFinder_FindNodes(t *testing.T) { bs.Shutdown() }() - nf := newContactFinder(dhts[2].node, RandomBitmapP(), false) - res, err := nf.Find() + contacts, found, err := FindContacts(dhts[2].node, RandomBitmapP(), false, nil) if err != nil { t.Fatal(err) } - foundNodes, found := res.Contacts, res.Found if found { t.Fatal("something was found, but it should not have been") } - if len(foundNodes) != 3 { - t.Errorf("expected 3 node, found %d", len(foundNodes)) + if len(contacts) != 3 { + t.Errorf("expected 3 node, found %d", len(contacts)) } foundBootstrap := false foundOne := false foundTwo := false - for _, n := range foundNodes { + for _, n := range contacts { if n.ID.Equals(bs.id) { foundBootstrap = true } @@ -66,8 +64,7 @@ func TestNodeFinder_FindNodes_NoBootstrap(t *testing.T) { } }() - nf := newContactFinder(dhts[2].node, RandomBitmapP(), false) - _, err := nf.Find() + _, _, err := FindContacts(dhts[2].node, RandomBitmapP(), false, nil) if err == nil { t.Fatal("contact finder should have errored saying that there are no contacts in the routing table") } @@ -86,23 +83,21 @@ func TestNodeFinder_FindValue(t *testing.T) { nodeToFind := Contact{ID: RandomBitmapP(), IP: net.IPv4(1, 2, 3, 4), Port: 5678} dhts[0].node.store.Upsert(blobHashToFind, nodeToFind) - nf := newContactFinder(dhts[2].node, blobHashToFind, true) - res, err := nf.Find() + contacts, found, err := FindContacts(dhts[2].node, blobHashToFind, true, nil) if err != nil { t.Fatal(err) } - foundNodes, found := res.Contacts, res.Found if !found { t.Fatal("node was not found") } - if len(foundNodes) != 1 { - t.Fatalf("expected one node, found %d", len(foundNodes)) + if len(contacts) != 1 { + t.Fatalf("expected one node, found %d", len(contacts)) } - if !foundNodes[0].ID.Equals(nodeToFind.ID) { - t.Fatalf("found node id %s, expected %s", foundNodes[0].ID.Hex(), nodeToFind.ID.Hex()) + if !contacts[0].ID.Equals(nodeToFind.ID) { + t.Fatalf("found node id %s, expected %s", contacts[0].ID.Hex(), nodeToFind.ID.Hex()) } } diff --git a/dht/node.go b/dht/node.go index 556b30a..d2e5ebc 100644 --- a/dht/node.go +++ b/dht/node.go @@ -89,12 +89,14 @@ func (n *Node) Connect(conn UDPConn) error { <-n.stop.Ch() n.tokens.Stop() n.connClosed = true - if err := n.conn.Close(); err != nil { + err := n.conn.Close() + if err != nil { log.Error("error closing node connection on shutdown - ", err) } }() packets := make(chan packet) + n.stop.Add(1) go func() { defer n.stop.Done() @@ -124,6 +126,7 @@ func (n *Node) Connect(conn UDPConn) error { } } }() + n.stop.Add(1) go func() { defer n.stop.Done() @@ -140,7 +143,11 @@ func (n *Node) Connect(conn UDPConn) error { } }() - n.startRoutingTableGrooming() + n.stop.Add(1) + go func() { + defer n.stop.Done() + n.startRoutingTableGrooming() + }() return nil } @@ -427,19 +434,15 @@ func (n *Node) CountActiveTransactions() int { } func (n *Node) startRoutingTableGrooming() { - n.stop.Add(1) - go func() { - defer n.stop.Done() - refreshTicker := time.NewTicker(tRefresh / 5) // how often to check for buckets that need to be refreshed - for { - select { - case <-refreshTicker.C: - RoutingTableRefresh(n, tRefresh, n.stop.Ch()) - case <-n.stop.Ch(): - return - } + refreshTicker := time.NewTicker(tRefresh / 5) // how often to check for buckets that need to be refreshed + for { + select { + case <-refreshTicker.C: + RoutingTableRefresh(n, tRefresh, n.stop.Ch()) + case <-n.stop.Ch(): + return } - }() + } } // Store stores a node contact in the node's contact store. diff --git a/dht/node_finder.go b/dht/node_finder.go index 6fbd35f..878c4ee 100644 --- a/dht/node_finder.go +++ b/dht/node_finder.go @@ -35,13 +35,8 @@ type contactFinder struct { outstandingRequests uint } -type findNodeResponse struct { - Found bool - Contacts []Contact -} - -func newContactFinder(node *Node, target Bitmap, findValue bool) *contactFinder { - return &contactFinder{ +func FindContacts(node *Node, target Bitmap, findValue bool, upstreamStop stopOnce.Chan) ([]Contact, bool, error) { + cf := &contactFinder{ node: node, target: target, findValue: findValue, @@ -52,14 +47,18 @@ func newContactFinder(node *Node, target Bitmap, findValue bool) *contactFinder stop: stopOnce.New(), outstandingRequestsMutex: &sync.RWMutex{}, } + if upstreamStop != nil { + cf.stop.Link(upstreamStop) + } + return cf.Find() } -func (cf *contactFinder) Cancel() { +func (cf *contactFinder) Stop() { cf.stop.Stop() cf.stop.Wait() } -func (cf *contactFinder) Find() (findNodeResponse, error) { +func (cf *contactFinder) Find() ([]Contact, bool, error) { if cf.findValue { log.Debugf("[%s] starting an iterative Find for the value %s", cf.node.id.HexShort(), cf.target.HexShort()) } else { @@ -67,7 +66,7 @@ func (cf *contactFinder) Find() (findNodeResponse, error) { } cf.appendNewToShortlist(cf.node.rt.GetClosest(cf.target, alpha)) if len(cf.shortlist) == 0 { - return findNodeResponse{}, errors.Err("no contacts in routing table") + return nil, false, errors.Err("no contacts in routing table") } for i := 0; i < alpha; i++ { @@ -83,18 +82,20 @@ func (cf *contactFinder) Find() (findNodeResponse, error) { // TODO: what to do if we have less than K active contacts, shortlist is empty, but we // TODO: have other contacts in our routing table whom we have not contacted. prolly contact them - result := findNodeResponse{} + var contacts []Contact + var found bool if cf.findValue && len(cf.findValueResult) > 0 { - result.Found = true - result.Contacts = cf.findValueResult + contacts = cf.findValueResult + found = true } else { - result.Contacts = cf.activeContacts - if len(result.Contacts) > bucketSize { - result.Contacts = result.Contacts[:bucketSize] + contacts = cf.activeContacts + if len(contacts) > bucketSize { + contacts = contacts[:bucketSize] } } - return result, nil + cf.Stop() + return contacts, found, nil } func (cf *contactFinder) iterationWorker(num int) { diff --git a/dht/routing_table.go b/dht/routing_table.go index 6507f9d..29ec855 100644 --- a/dht/routing_table.go +++ b/dht/routing_table.go @@ -12,6 +12,7 @@ import ( "time" "github.com/lbryio/lbry.go/errors" + "github.com/lbryio/lbry.go/stopOnce" "github.com/lyoshenka/bencode" log "github.com/sirupsen/logrus" @@ -445,40 +446,22 @@ func (rt *routingTable) UnmarshalJSON(b []byte) error { } // RoutingTableRefresh refreshes any buckets that need to be refreshed -// It returns a channel that will be closed when the refresh is done -func RoutingTableRefresh(n *Node, refreshInterval time.Duration, cancel <-chan struct{}) <-chan struct{} { - var wg sync.WaitGroup - done := make(chan struct{}) +func RoutingTableRefresh(n *Node, refreshInterval time.Duration, upstreamStop stopOnce.Chan) { + done := stopOnce.New() for _, id := range n.rt.GetIDsForRefresh(refreshInterval) { - wg.Add(1) + done.Add(1) go func(id Bitmap) { - defer wg.Done() - - nf := newContactFinder(n, id, false) - - if cancel != nil { - go func() { - select { - case <-cancel: - nf.Cancel() - case <-done: - } - }() - } - - if _, err := nf.Find(); err != nil { + defer done.Done() + _, _, err := FindContacts(n, id, false, upstreamStop) + if err != nil { log.Error("error finding contact during routing table refresh - ", err) } }(id) } - go func() { - wg.Wait() - close(done) - }() - - return done + done.Wait() + done.Stop() } func moveToBack(peers []peer, index int) {