mirror of
https://github.com/ethereum/go-ethereum.git
synced 2026-08-20 10:52:25 +00:00
swarm/network, swarm/storage: Remove redundant id to sendpriority
This commit is contained in:
parent
241119a970
commit
ea260a178d
6 changed files with 8 additions and 12 deletions
|
|
@ -213,8 +213,6 @@ func (d *Delivery) handleChunkDeliveryMsg(ctx context.Context, sp *Peer, req *Ch
|
||||||
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 {
|
||||||
|
|
@ -268,7 +266,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, fmt.Sprintf("request.%v.%v", sp.ID(), req.Addr))
|
}, Top)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, nil, err
|
return nil, nil, err
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -300,7 +300,7 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
|
log.Trace("sending want batch", "peer", p.ID(), "stream", msg.Stream, "from", msg.From, "to", msg.To)
|
||||||
err := p.SendPriority(ctx, msg, c.priority, "")
|
err := p.SendPriority(ctx, msg, c.priority)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Warn("SendPriority error", "err", err)
|
log.Warn("SendPriority error", "err", err)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -151,12 +151,11 @@ 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)
|
||||||
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
|
||||||
func (p *Peer) SendPriority(ctx context.Context, msg interface{}, priority uint8, traceId string) error {
|
func (p *Peer) SendPriority(ctx context.Context, msg interface{}, priority uint8) error {
|
||||||
defer metrics.GetOrRegisterResettingTimer(fmt.Sprintf("peer.sendpriority_t.%d", priority), nil).UpdateSince(time.Now())
|
defer metrics.GetOrRegisterResettingTimer(fmt.Sprintf("peer.sendpriority_t.%d", priority), nil).UpdateSince(time.Now())
|
||||||
metrics.GetOrRegisterCounter(fmt.Sprintf("peer.sendpriority.%d", priority), nil).Inc(1)
|
metrics.GetOrRegisterCounter(fmt.Sprintf("peer.sendpriority.%d", priority), nil).Inc(1)
|
||||||
wmsg := WrappedPriorityMsg{
|
wmsg := WrappedPriorityMsg{
|
||||||
|
|
@ -202,7 +201,7 @@ func (p *Peer) SendOfferedHashes(s *server, f, t uint64) error {
|
||||||
Stream: s.stream,
|
Stream: s.stream,
|
||||||
}
|
}
|
||||||
log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
|
log.Trace("Swarm syncer offer batch", "peer", p.ID(), "stream", s.stream, "len", len(hashes), "from", from, "to", to)
|
||||||
return p.SendPriority(ctx, msg, s.priority, "send.offered.hashes")
|
return p.SendPriority(ctx, msg, s.priority)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *Peer) getServer(s Stream) (*server, error) {
|
func (p *Peer) getServer(s Stream) (*server, error) {
|
||||||
|
|
|
||||||
|
|
@ -359,7 +359,7 @@ func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8
|
||||||
}
|
}
|
||||||
log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)
|
log.Debug("Subscribe ", "peer", peerId, "stream", s, "history", h)
|
||||||
|
|
||||||
return peer.SendPriority(context.TODO(), msg, priority, "")
|
return peer.SendPriority(context.TODO(), msg, priority)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *Registry) Unsubscribe(peerId enode.ID, s Stream) error {
|
func (r *Registry) Unsubscribe(peerId enode.ID, s Stream) error {
|
||||||
|
|
@ -730,7 +730,7 @@ func (c *client) batchDone(p *Peer, req *OfferedHashesMsg, hashes []byte) error
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
if err := p.SendPriority(context.TODO(), tp, c.priority, ""); err != nil {
|
if err := p.SendPriority(context.TODO(), tp, c.priority); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if c.to > 0 && tp.Takeover.End >= c.to {
|
if c.to > 0 && tp.Takeover.End >= c.to {
|
||||||
|
|
|
||||||
|
|
@ -177,7 +177,6 @@ func testFileStoreCapacity(toEncrypt bool, t *testing.T) {
|
||||||
// TestGetAllReferences only tests that GetAllReferences returns an expected
|
// TestGetAllReferences only tests that GetAllReferences returns an expected
|
||||||
// number of references for a given file
|
// number of references for a given file
|
||||||
func TestGetAllReferences(t *testing.T) {
|
func TestGetAllReferences(t *testing.T) {
|
||||||
t.Skip("sometimes fails with chunk count 247 instead of 248")
|
|
||||||
tdb, cleanup, err := newTestDbStore(false, false)
|
tdb, cleanup, err := newTestDbStore(false, false)
|
||||||
defer cleanup()
|
defer cleanup()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,4 @@
|
||||||
//// Copyright 2016 The go-ethereum Authors
|
// Copyright 2016 The go-ethereum Authors
|
||||||
// This file is part of the go-ethereum library.
|
// This file is part of the go-ethereum library.
|
||||||
//
|
//
|
||||||
// The go-ethereum library is free software: you can redistribute it and/or modify
|
// The go-ethereum library is free software: you can redistribute it and/or modify
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue