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) {