mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
swarm/storage: fix tracing
This commit is contained in:
parent
002cee3868
commit
6f4dcb14c6
3 changed files with 7 additions and 9 deletions
|
|
@ -23,7 +23,6 @@ import (
|
||||||
|
|
||||||
"github.com/ethereum/go-ethereum/log"
|
"github.com/ethereum/go-ethereum/log"
|
||||||
"github.com/ethereum/go-ethereum/p2p/enode"
|
"github.com/ethereum/go-ethereum/p2p/enode"
|
||||||
"github.com/ethereum/go-ethereum/swarm/spancontext"
|
|
||||||
"github.com/ethereum/go-ethereum/swarm/storage"
|
"github.com/ethereum/go-ethereum/swarm/storage"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
@ -120,7 +119,6 @@ func (f *FetcherFactory) New(ctx context.Context, source storage.Address, peers
|
||||||
|
|
||||||
// NewFetcher creates a new Fetcher for the given chunk address using the given request function.
|
// NewFetcher creates a new Fetcher for the given chunk address using the given request function.
|
||||||
func NewFetcher(ctx context.Context, addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher {
|
func NewFetcher(ctx context.Context, addr storage.Address, rf RequestFunc, skipCheck bool) *Fetcher {
|
||||||
sctx, sp := spancontext.StartSpan(ctx, "fetcher")
|
|
||||||
return &Fetcher{
|
return &Fetcher{
|
||||||
addr: addr,
|
addr: addr,
|
||||||
protoRequestFunc: rf,
|
protoRequestFunc: rf,
|
||||||
|
|
@ -128,9 +126,8 @@ func NewFetcher(ctx context.Context, addr storage.Address, rf RequestFunc, skipC
|
||||||
requestC: make(chan uint8),
|
requestC: make(chan uint8),
|
||||||
searchTimeout: defaultSearchTimeout,
|
searchTimeout: defaultSearchTimeout,
|
||||||
skipCheck: skipCheck,
|
skipCheck: skipCheck,
|
||||||
ctx: sctx,
|
ctx: ctx,
|
||||||
quitFunc: func() {
|
quitFunc: func() {
|
||||||
sp.Finish()
|
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -213,11 +213,12 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *Ch
|
||||||
ctx, osp = spancontext.StartSpan(
|
ctx, osp = spancontext.StartSpan(
|
||||||
ctx,
|
ctx,
|
||||||
"chunk.delivery")
|
"chunk.delivery")
|
||||||
defer osp.Finish()
|
|
||||||
|
|
||||||
processReceivedChunksCount.Inc(1)
|
processReceivedChunksCount.Inc(1)
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
|
defer osp.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))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -150,7 +150,7 @@ func (n *NetStore) get(ctx context.Context, ref Address) (Chunk, func(context.Co
|
||||||
}
|
}
|
||||||
// The chunk is not available in the LocalStore, let's get the fetcher for it, or create a new one
|
// The chunk is not available in the LocalStore, let's get the fetcher for it, or create a new one
|
||||||
// if it doesn't exist yet
|
// if it doesn't exist yet
|
||||||
f := n.getOrCreateFetcher(ref)
|
f := n.getOrCreateFetcher(ctx, ref)
|
||||||
// If the caller needs the chunk, it has to use the returned fetch function to get it
|
// If the caller needs the chunk, it has to use the returned fetch function to get it
|
||||||
return nil, f.Fetch, nil
|
return nil, f.Fetch, nil
|
||||||
}
|
}
|
||||||
|
|
@ -161,7 +161,7 @@ func (n *NetStore) get(ctx context.Context, ref Address) (Chunk, func(context.Co
|
||||||
// getOrCreateFetcher attempts at retrieving an existing fetchers
|
// getOrCreateFetcher attempts at retrieving an existing fetchers
|
||||||
// if none exists, creates one and saves it in the fetchers cache
|
// if none exists, creates one and saves it in the fetchers cache
|
||||||
// caller must hold the lock
|
// caller must hold the lock
|
||||||
func (n *NetStore) getOrCreateFetcher(ref Address) *fetcher {
|
func (n *NetStore) getOrCreateFetcher(ctx context.Context, ref Address) *fetcher {
|
||||||
if f := n.getFetcher(ref); f != nil {
|
if f := n.getFetcher(ref); f != nil {
|
||||||
return f
|
return f
|
||||||
}
|
}
|
||||||
|
|
@ -169,7 +169,7 @@ func (n *NetStore) getOrCreateFetcher(ref Address) *fetcher {
|
||||||
// no fetcher for the given address, we have to create a new one
|
// no fetcher for the given address, we have to create a new one
|
||||||
key := hex.EncodeToString(ref)
|
key := hex.EncodeToString(ref)
|
||||||
// create the context during which fetching is kept alive
|
// create the context during which fetching is kept alive
|
||||||
ctx, cancel := context.WithTimeout(context.Background(), fetcherTimeout)
|
cctx, cancel := context.WithTimeout(ctx, fetcherTimeout)
|
||||||
// destroy is called when all requests finish
|
// destroy is called when all requests finish
|
||||||
destroy := func() {
|
destroy := func() {
|
||||||
// remove fetcher from fetchers
|
// remove fetcher from fetchers
|
||||||
|
|
@ -183,7 +183,7 @@ func (n *NetStore) getOrCreateFetcher(ref Address) *fetcher {
|
||||||
// the peers which requested the chunk should not be requested to deliver it.
|
// the peers which requested the chunk should not be requested to deliver it.
|
||||||
peers := &sync.Map{}
|
peers := &sync.Map{}
|
||||||
|
|
||||||
fetcher := newFetcher(ref, n.NewNetFetcherFunc(ctx, ref, peers), destroy, peers, n.closeC)
|
fetcher := newFetcher(ref, n.NewNetFetcherFunc(cctx, ref, peers), destroy, peers, n.closeC)
|
||||||
n.fetchers.Add(key, fetcher)
|
n.fetchers.Add(key, fetcher)
|
||||||
|
|
||||||
return fetcher
|
return fetcher
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue