dlss
/
FedP2P
4
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Track chunks read for webseed peers too

This commit is contained in:
Matt Joiner 2021-05-21 11:49:57 +10:00
parent 33d3a75cd3
commit c895a21a16
3 changed files with 15 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
conn_stats.go
View File

@ -79,14 +79,9 @@ func (cs *ConnStats) wroteMsg(msg *pp.Message) {
} }
} }
func (cs *ConnStats) readMsg(msg *pp.Message) { func (cs *ConnStats) receivedChunk(size int64) {
// We want to also handle extended metadata pieces here, but we wouldn't
// have decoded the extended payload yet.
switch msg.Type {
case pp.Piece:
cs.ChunksRead.Add(1) cs.ChunksRead.Add(1)
cs.BytesReadData.Add(int64(len(msg.Piece))) cs.BytesReadData.Add(size)
}
} }
func (cs *ConnStats) incrementPiecesDirtiedGood() { func (cs *ConnStats) incrementPiecesDirtiedGood() {

15
peerconn.go
View File

@ -856,10 +856,6 @@ func (cn *PeerConn) wroteMsg(msg *pp.Message) {
cn.allStats(func(cs *ConnStats) { cs.wroteMsg(msg) }) cn.allStats(func(cs *ConnStats) { cs.wroteMsg(msg) })
} }
func (cn *PeerConn) readMsg(msg *pp.Message) {
cn.allStats(func(cs *ConnStats) { cs.readMsg(msg) })
}
// After handshake, we know what Torrent and Client stats to include for a // After handshake, we know what Torrent and Client stats to include for a
// connection. // connection.
func (cn *Peer) postHandshakeStats(f func(*ConnStats)) { func (cn *Peer) postHandshakeStats(f func(*ConnStats)) {
@ -1065,7 +1061,6 @@ func (c *PeerConn) mainReadLoop() (err error) {
if err != nil { if err != nil {
return err return err
} }
c.readMsg(&msg)
c.lastMessageReceived = time.Now() c.lastMessageReceived = time.Now()
if msg.Keepalive { if msg.Keepalive {
receivedKeepalives.Add(1) receivedKeepalives.Add(1)
@ -1105,6 +1100,7 @@ func (c *PeerConn) mainReadLoop() (err error) {
r := newRequestFromMessage(&msg) r := newRequestFromMessage(&msg)
err = c.onReadRequest(r) err = c.onReadRequest(r)
case pp.Piece: case pp.Piece:
c.doChunkReadStats(int64(len(msg.Piece)))
err = c.receiveChunk(&msg) err = c.receiveChunk(&msg)
if len(msg.Piece) == int(t.chunkSize) { if len(msg.Piece) == int(t.chunkSize) {
t.chunkPool.Put(&msg.Piece) t.chunkPool.Put(&msg.Piece)
@ -1253,10 +1249,12 @@ func (cn *PeerConn) rw() io.ReadWriter {
}{cn.r, cn.w} }{cn.r, cn.w}
} }
func (c *Peer) doChunkReadStats(size int64) {
c.allStats(func(cs *ConnStats) { cs.receivedChunk(size) })
}
// Handle a received chunk from a peer. // Handle a received chunk from a peer.
func (c *Peer) receiveChunk(msg *pp.Message) error { func (c *Peer) receiveChunk(msg *pp.Message) error {
t := c.t
cl := t.cl
chunksReceived.Add("total", 1) chunksReceived.Add("total", 1)
req := newRequestFromMessage(msg) req := newRequestFromMessage(msg)
@ -1296,6 +1294,9 @@ func (c *Peer) receiveChunk(msg *pp.Message) error {
} }
} }
t := c.t
cl := t.cl
// Do we actually want this chunk? // Do we actually want this chunk?
if t.haveChunk(req) { if t.haveChunk(req) {
chunksReceived.Add("wasted", 1) chunksReceived.Add("wasted", 1)

4
webseed-peer.go
View File

@ -110,6 +110,10 @@ func (ws *webseedPeer) onClose() {
func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) { func (ws *webseedPeer) requestResultHandler(r Request, webseedRequest webseed.Request) {
result := <-webseedRequest.Result result := <-webseedRequest.Result
// We do this here rather than inside receiveChunk, since we want to count errors too. I'm not
// sure if we can divine which errors indicate cancellation on our end without hitting the
// network though.
ws.peer.doChunkReadStats(int64(len(result.Bytes)))
ws.peer.t.cl.lock() ws.peer.t.cl.lock()
defer ws.peer.t.cl.unlock() defer ws.peer.t.cl.unlock()
if result.Err != nil { if result.Err != nil {