Ability to override fifos/

[btrtrc.git] / peerconn.go

diff --git a/peerconn.go b/peerconn.go
index 4d978f07f58d12ae2a3d39f4e78075fdb17547ba..322e7625ba37b59884dbc7051a23b897893dc1e4 100644 (file)
--- a/peerconn.go
+++ b/peerconn.go
@@ -20,13 +20,14 @@ import (
        "github.com/anacrolix/missinggo/iter"
        "github.com/anacrolix/missinggo/v2/bitmap"
        "github.com/anacrolix/multiless"
+       "golang.org/x/time/rate"
+
        "github.com/anacrolix/torrent/bencode"
        "github.com/anacrolix/torrent/metainfo"
        "github.com/anacrolix/torrent/mse"
        pp "github.com/anacrolix/torrent/peer_protocol"
        request_strategy "github.com/anacrolix/torrent/request-strategy"
        "github.com/anacrolix/torrent/typed-roaring"
-       "golang.org/x/time/rate"
 )
 
 type PeerSource string
@@ -170,8 +171,8 @@ type PeerConn struct {
        peerSentHaveAll bool
 }
 
-func (cn *PeerConn) connStatusString() string {
-       return fmt.Sprintf("%+-55q %s %s", cn.PeerID, cn.PeerExtensionBytes, cn.connString)
+func (cn *PeerConn) peerImplStatusLines() []string {
+       return []string{fmt.Sprintf("%+-55q %s %s", cn.PeerID, cn.PeerExtensionBytes, cn.connString)}
 }
 
 func (cn *Peer) updateExpectingChunks() {
@@ -282,6 +283,10 @@ func (cn *Peer) completedString() string {
        return fmt.Sprintf("%d/%d", have, cn.bestPeerNumPieces())
 }
 
+func (cn *Peer) CompletedString() string {
+       return cn.completedString()
+}
+
 func (cn *PeerConn) onGotInfo(info *metainfo.Info) {
        cn.setNumPieces(info.NumPieces())
 }
@@ -347,6 +352,10 @@ func (cn *Peer) statusFlags() (ret string) {
        return
 }
 
+func (cn *Peer) StatusFlags() string {
+       return cn.statusFlags()
+}
+
 func (cn *Peer) downloadRate() float64 {
        num := cn._stats.BytesReadUsefulData.Int64()
        if num == 0 {
@@ -362,6 +371,17 @@ func (cn *Peer) DownloadRate() float64 {
        return cn.downloadRate()
 }
 
+func (cn *Peer) UploadRate() float64 {
+       cn.locker().RLock()
+       defer cn.locker().RUnlock()
+       num := cn._stats.BytesWrittenData.Int64()
+       if num == 0 {
+               return 0
+       }
+       return float64(num) / time.Now().Sub(cn.completedHandshake).Seconds()
+}
+
+
 func (cn *Peer) iterContiguousPieceRequests(f func(piece pieceIndex, count int)) {
        var last Option[pieceIndex]
        var count int
@@ -388,14 +408,14 @@ func (cn *Peer) writeStatus(w io.Writer, t *Torrent) {
        if cn.closed.IsSet() {
                fmt.Fprint(w, "CLOSED: ")
        }
-       fmt.Fprintln(w, cn.connStatusString())
+       fmt.Fprintln(w, strings.Join(cn.peerImplStatusLines(), "\n"))
        prio, err := cn.peerPriority()
        prioStr := fmt.Sprintf("%08x", prio)
        if err != nil {
                prioStr += ": " + err.Error()
        }
-       fmt.Fprintf(w, "    bep40-prio: %v\n", prioStr)
-       fmt.Fprintf(w, "    last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
+       fmt.Fprintf(w, "bep40-prio: %v\n", prioStr)
+       fmt.Fprintf(w, "last msg: %s, connected: %s, last helpful: %s, itime: %s, etime: %s\n",
                eventAgeString(cn.lastMessageReceived),
                eventAgeString(cn.completedHandshake),
                eventAgeString(cn.lastHelpful()),
@@ -403,7 +423,7 @@ func (cn *Peer) writeStatus(w io.Writer, t *Torrent) {
                cn.totalExpectingTime(),
        )
        fmt.Fprintf(w,
-               "    %s completed, %d pieces touched, good chunks: %v/%v:%v reqq: %d+%v/(%d/%d):%d/%d, flags: %s, dr: %.1f KiB/s\n",
+               "%s completed, %d pieces touched, good chunks: %v/%v:%v reqq: %d+%v/(%d/%d):%d/%d, flags: %s, dr: %.1f KiB/s\n",
                cn.completedString(),
                len(cn.peerTouchedPieces),
                &cn._stats.ChunksReadUseful,
@@ -418,7 +438,7 @@ func (cn *Peer) writeStatus(w io.Writer, t *Torrent) {
                cn.statusFlags(),
                cn.downloadRate()/(1<<10),
        )
-       fmt.Fprintf(w, "    requested pieces:")
+       fmt.Fprintf(w, "requested pieces:")
        cn.iterContiguousPieceRequests(func(piece pieceIndex, count int) {
                fmt.Fprintf(w, " %v(%v)", piece, count)
        })
@@ -595,6 +615,10 @@ type messageWriter func(pp.Message) bool
 // This function seems to only used by Peer.request. It's all logic checks, so maybe we can no-op it
 // when we want to go fast.
 func (cn *Peer) shouldRequest(r RequestIndex) error {
+       err := cn.t.checkValidReceiveChunk(cn.t.requestIndexToRequest(r))
+       if err != nil {
+               return err
+       }
        pi := cn.t.pieceIndexOfRequestIndex(r)
        if cn.requestState.Cancelled.Contains(r) {
                return errors.New("request is cancelled and waiting acknowledgement")
@@ -1424,8 +1448,13 @@ func (c *Peer) receiveChunk(msg *pp.Message) error {
        chunksReceived.Add("total", 1)
 
        ppReq := newRequestFromMessage(msg)
-       req := c.t.requestIndexFromRequest(ppReq)
        t := c.t
+       err := t.checkValidReceiveChunk(ppReq)
+       if err != nil {
+               err = log.WithLevel(log.Warning, err)
+               return err
+       }
+       req := c.t.requestIndexFromRequest(ppReq)
 
        if c.bannableAddr.Ok {
                t.smartBanCache.RecordBlock(c.bannableAddr.Value, req, msg.Piece)
@@ -1507,7 +1536,7 @@ func (c *Peer) receiveChunk(msg *pp.Message) error {
                p.cancel(req)
        }
 
-       err := func() error {
+       err = func() error {
                cl.unlock()
                defer cl.lock()
                concurrentChunkWrites.Add(1)
@@ -1826,6 +1855,10 @@ func (cn *Peer) stats() *ConnStats {
        return &cn._stats
 }
 
+func (cn *Peer) Stats() *ConnStats {
+       return cn.stats()
+}
+
 func (p *Peer) TryAsPeerConn() (*PeerConn, bool) {
        pc, ok := p.peerImpl.(*PeerConn)
        return pc, ok