Add miner lookup using DHT #2
Conversation
kasteph
force-pushed
the
feat/add-routing
branch
from
da4de08 to
0920414
Compare
main.go
Outdated
| } | ||
|
|
||
| func getActiveMiners(db *pg.DB, fp *filecoinPeer) (<-chan minerInfo, error) { | ||
| go fp.bootstrap() |
There was a problem hiding this comment.
I believe this function already does the connecting in the background and returns, so we shouldn't need to launch it as goroutine. It might be desirable to wait for it to finish connecting to bootstrappers before making dht queries though.
| i, activeMiner := i, activeMiner | ||
| wg.Add(1) | ||
|
|
||
| go func() { |
There was a problem hiding this comment.
I am guessing that if we have 2000 active miners this is going to launch 2000 goroutines which are going to launch 2000 parallel dht queries.
Might be better to send that list of peer IDs to a channel and launch 50 workers to read from it and process each peer ID, so that we don't have more than 50 queries on flight.
| } | ||
| found = append(found, f) | ||
| }() | ||
| } | ||
|
|
||
| go func() { | ||
| wg.Wait() |
There was a problem hiding this comment.
Waiting in a goroutine makes the waiting useless as nothing blocks or is meant to happen after waiting.
| var ( | ||
| found = make([]minerInfo, len(activeMiners)) | ||
| peerIDs = make(chan minerWithDecodedPeerID, 50) | ||
| wg = sync.WaitGroup{} | ||
| maxWorkers = runtime.GOMAXPROCS(0) | ||
| sem = semaphore.NewWeighted(int64(maxWorkers)) | ||
| ) | ||
| for i := range activeMiners { | ||
|
|
||
| for i, activeMiner := range activeMiners { | ||
| i, activeMiner := i, activeMiner | ||
| wg.Add(1) | ||
| miner := activeMiners[i] | ||
|
|
||
| decodedPid, err := peer.Decode(miner.PeerID) | ||
| if err != nil { | ||
| log.Warnw("could not decode peer ID ", "ID", miner.PeerID, "error", err) | ||
| } | ||
| peerIDs <- minerWithDecodedPeerID{miner: miner, decodedPeerID: decodedPid} | ||
|
|
||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
|
|
||
| ctx, cancel := context.WithTimeout(context.Background(), time.Duration(15*time.Second)) | ||
| defer cancel() | ||
|
|
||
| decodedPid, err := peer.Decode(activeMiner.PeerID) | ||
| if err != nil { | ||
| log.Warnw("could not decode peer ID ", "ID", activeMiner.PeerID, "error", err) | ||
| return | ||
| if err := sem.Acquire(fp.ctx, 1); err != nil { | ||
| log.Warnw("Failed to acquire semaphore: ", err) |
There was a problem hiding this comment.
This is probably not a good pattern (I have not seen anything like this anywhere). It is overly complex in that it uses all a channel, a semaphore and a waitgroup, when a worker pattern can be solved with a channel.
It might work, in the sense that only 50 things are querying the DHT, but there will be also 50 items in the channel (I guess) and 50 goroutines waiting to access the semaphore (those that managed to put something in the channel), and another 50 executing (those that managed to Acquire the semaphore). Overly complex.
This PR does the following:
sourceto thelocations.minerstableI would like feedback particularly on these lines:
locations.minerstable to fill in the heights of miners found via DHT for now?