forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/newtork: WIP Span request span until delivery and put

Browse source
This commit is contained in:
lash 2019-02-08 17:52:49 +01:00
parent 27e3f96819
commit f8001a6891
2 changed files with 23 additions and 14 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
swarm/network/stream/delivery.go
View file

@ -209,15 +209,22 @@ type ChunkDeliveryMsgSyncing ChunkDeliveryMsg
// TODO: Fix context SNAFU // TODO: Fix context SNAFU
func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error { func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *ChunkDeliveryMsg) error {
var osp opentracing.Span // var osp opentracing.Span
ctx, osp = spancontext.StartSpan( // ctx, osp = spancontext.StartSpan(
ctx, // ctx,
"chunk.delivery") // "chunk.delivery")
spanId := fmt.Sprintf("request.%v.%v", sp.ID(), req.Addr)
span, spanOk := sp.spans.Load(spanId)
sp.spans.Delete(spanId)
processReceivedChunksCount.Inc(1) processReceivedChunksCount.Inc(1)
go func() { go func() {
defer osp.Finish() //defer osp.Finish()
if spanOk {
defer span.(opentracing.Span).Finish()
}
req.peer = sp req.peer = sp
err := d.chunkStore.Put(ctx, storage.NewChunk(req.Addr, req.SData)) err := d.chunkStore.Put(ctx, storage.NewChunk(req.Addr, req.SData))
@ -272,7 +279,7 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) (
Addr: req.Addr, Addr: req.Addr,
SkipCheck: req.SkipCheck, SkipCheck: req.SkipCheck,
HopCount: req.HopCount, HopCount: req.HopCount,
}, Top, "request.from.peers") }, Top, fmt.Sprintf("request.%v.%v", sp.ID(), req.Addr))
if err != nil { if err != nil {
return nil, nil, err return nil, nil, err
} }

18
swarm/network/stream/peer.go
View file

@ -88,11 +88,11 @@ func NewPeer(peer *protocols.Peer, streamer *Registry) *Peer {
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
go p.pq.Run(ctx, func(i interface{}) { go p.pq.Run(ctx, func(i interface{}) {
wmsg := i.(WrappedPriorityMsg) wmsg := i.(WrappedPriorityMsg)
defer p.spans.Delete(wmsg.Context) // defer p.spans.Delete(wmsg.Context)
sp, ok := p.spans.Load(wmsg.Context) // sp, ok := p.spans.Load(wmsg.Context)
if ok { // if ok {
defer sp.(opentracing.Span).Finish() // defer sp.(opentracing.Span).Finish()
} // }
err := p.Send(wmsg.Context, wmsg.Msg) err := p.Send(wmsg.Context, wmsg.Msg)
if err != nil { if err != nil {
log.Error("Message send error, dropping peer", "peer", p.ID(), "err", err) log.Error("Message send error, dropping peer", "peer", p.ID(), "err", err)
@ -158,7 +158,8 @@ func (p *Peer) Deliver(ctx context.Context, chunk storage.Chunk, priority uint8,
spanName += ".retrieval" spanName += ".retrieval"
} }
return p.SendPriority(ctx, msg, priority, spanName) //return p.SendPriority(ctx, msg, priority, spanName)
return p.SendPriority(ctx, msg, priority, "")
} }
// SendPriority sends message to the peer using the outgoing priority queue // SendPriority sends message to the peer using the outgoing priority queue
@ -171,7 +172,7 @@ func (p *Peer) SendPriority(ctx context.Context, msg interface{}, priority uint8
ctx, ctx,
traceId, traceId,
) )
p.spans.Store(ctx, sp) p.spans.Store(traceId, sp)
} }
wmsg := WrappedPriorityMsg{ wmsg := WrappedPriorityMsg{
Context: ctx, Context: ctx,
@ -190,7 +191,8 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
var sp opentracing.Span var sp opentracing.Span
ctx, sp := spancontext.StartSpan( ctx, sp := spancontext.StartSpan(
context.TODO(), context.TODO(),
"send.offered.hashes") "",
)
defer sp.Finish() defer sp.Finish()
hashes, from, to, proof, err := s.setNextBatch(f, t) hashes, from, to, proof, err := s.setNextBatch(f, t)