opensource/kubo
1
0
Fork 0
mirror of https://github.com/ipfs/kubo.git synced 2025-07-01 10:49:24 +08:00
Code Issues Projects Releases Wiki Activity

contextify peermanager

Browse Source
This commit is contained in:
Jeromy
2015-05-13 16:35:08 -07:00
committed by Juan Batiz-Benet
parent 8443b99c1d
commit ef967ceeef
4 changed files with 14 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
exchange/bitswap/bitswap.go
View File

@ -316,8 +316,6 @@ func (bs *Bitswap) sendWantlistToProviders(ctx context.Context, entries []wantli
// TODO(brian): handle errors // TODO(brian): handle errors
func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) error { func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) error {
//defer log.EventBegin(ctx, "receiveMessage", p, incoming).Done()
// This call records changes to wantlists, blocks received, // This call records changes to wantlists, blocks received,
// and number of bytes transfered. // and number of bytes transfered.
bs.engine.MessageReceived(p, incoming) bs.engine.MessageReceived(p, incoming)

2
exchange/bitswap/decision/engine.go
View File

@ -206,7 +206,7 @@ func (e *Engine) MessageReceived(p peer.ID, m bsmsg.BitSwapMessage) error {
l.CancelWant(entry.Key) l.CancelWant(entry.Key)
e.peerRequestQueue.Remove(entry.Key, p) e.peerRequestQueue.Remove(entry.Key, p)
} else { } else {
log.Debugf("wants %s", entry.Key, entry.Priority) log.Debugf("wants %s - %d", entry.Key, entry.Priority)
l.Wants(entry.Key, entry.Priority) l.Wants(entry.Key, entry.Priority)
if exists, err := e.bs.Has(entry.Key); err == nil && exists { if exists, err := e.bs.Has(entry.Key); err == nil && exists {
e.peerRequestQueue.Push(entry.Entry, p) e.peerRequestQueue.Push(entry.Entry, p)

22
exchange/bitswap/peermanager.go
View File

@ -53,24 +53,24 @@ type msgQueue struct {
done chan struct{} done chan struct{}
} }
func (pm *PeerManager) SendBlock(env *engine.Envelope) { func (pm *PeerManager) SendBlock(ctx context.Context, env *engine.Envelope) {
// Blocks need to be sent synchronously to maintain proper backpressure // Blocks need to be sent synchronously to maintain proper backpressure
// throughout the network stack // throughout the network stack
defer env.Sent() defer env.Sent()
msg := bsmsg.New() msg := bsmsg.New()
msg.AddBlock(env.Block) msg.AddBlock(env.Block)
err := pm.network.SendMessage(context.TODO(), env.Peer, msg) err := pm.network.SendMessage(ctx, env.Peer, msg)
if err != nil { if err != nil {
log.Error(err) log.Error(err)
} }
} }
func (pm *PeerManager) startPeerHandler(p peer.ID) { func (pm *PeerManager) startPeerHandler(ctx context.Context, p peer.ID) *msgQueue {
_, ok := pm.peers[p] _, ok := pm.peers[p]
if ok { if ok {
// TODO: log an error? // TODO: log an error?
return return nil
} }
mq := new(msgQueue) mq := new(msgQueue)
@ -79,7 +79,8 @@ func (pm *PeerManager) startPeerHandler(p peer.ID) {
mq.p = p mq.p = p
pm.peers[p] = mq pm.peers[p] = mq
go pm.runQueue(mq) go pm.runQueue(ctx, mq)
return mq
} }
func (pm *PeerManager) stopPeerHandler(p peer.ID) { func (pm *PeerManager) stopPeerHandler(p peer.ID) {
@ -93,14 +94,14 @@ func (pm *PeerManager) stopPeerHandler(p peer.ID) {
delete(pm.peers, p) delete(pm.peers, p)
} }
func (pm *PeerManager) runQueue(mq *msgQueue) { func (pm *PeerManager) runQueue(ctx context.Context, mq *msgQueue) {
for { for {
select { select {
case <-mq.work: // there is work to be done case <-mq.work: // there is work to be done
// TODO: this might not need to be done every time, figure out // TODO: this might not need to be done every time, figure out
// a good heuristic // a good heuristic
err := pm.network.ConnectTo(context.TODO(), mq.p) err := pm.network.ConnectTo(ctx, mq.p)
if err != nil { if err != nil {
log.Error(err) log.Error(err)
// TODO: cant connect, what now? // TODO: cant connect, what now?
@ -114,7 +115,7 @@ func (pm *PeerManager) runQueue(mq *msgQueue) {
if wlm != nil && !wlm.Empty() { if wlm != nil && !wlm.Empty() {
// send wantlist updates // send wantlist updates
err = pm.network.SendMessage(context.TODO(), mq.p, wlm) err = pm.network.SendMessage(ctx, mq.p, wlm)
if err != nil { if err != nil {
log.Error("bitswap send error: ", err) log.Error("bitswap send error: ", err)
// TODO: what do we do if this fails? // TODO: what do we do if this fails?
@ -162,13 +163,12 @@ func (pm *PeerManager) Run(ctx context.Context) {
p, ok := pm.peers[msgp.to] p, ok := pm.peers[msgp.to]
if !ok { if !ok {
//TODO: decide, drop message? or dial? //TODO: decide, drop message? or dial?
pm.startPeerHandler(msgp.to) p = pm.startPeerHandler(ctx, msgp.to)
p = pm.peers[msgp.to]
} }
p.addMessage(msgp.msg) p.addMessage(msgp.msg)
case p := <-pm.connect: case p := <-pm.connect:
pm.startPeerHandler(p) pm.startPeerHandler(ctx, p)
case p := <-pm.disconnect: case p := <-pm.disconnect:
pm.stopPeerHandler(p) pm.stopPeerHandler(p)
case <-ctx.Done(): case <-ctx.Done():

4
exchange/bitswap/workers.go
View File

@ -46,6 +46,7 @@ func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
bs.rebroadcastWorker(ctx) bs.rebroadcastWorker(ctx)
}) })
// Start up a worker to manage sending out provides messages
px.Go(func(px process.Process) { px.Go(func(px process.Process) {
bs.provideCollector(ctx) bs.provideCollector(ctx)
}) })
@ -71,8 +72,7 @@ func (bs *Bitswap) taskWorker(ctx context.Context) {
continue continue
} }
//log.Event(ctx, "deliverBlocks", envelope.Message, envelope.Peer) bs.pm.SendBlock(ctx, envelope)
bs.pm.SendBlock(envelope)
case <-ctx.Done(): case <-ctx.Done():
return return
} }