cmd/btrtrc/fifos.go

   1 package main
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "encoding/hex"
   7         "fmt"
   8         "log"
   9         "os"
  10         "path"
  11         "sort"
  12         "strconv"
  13         "strings"
  14         "sync"
  15         "syscall"
  16         "time"
  17
  18         "github.com/anacrolix/dht/v2"
  19         "github.com/anacrolix/torrent"
  20         "github.com/anacrolix/torrent/metainfo"
  21         "github.com/anacrolix/torrent/types/infohash"
  22         "github.com/dustin/go-humanize"
  23 )
  24
  25 const (
  26         MaxListNameWidth = 40
  27         FIFOsDir         = "fifos"
  28         PeersDir         = "peers"
  29         FilesDir         = "files"
  30 )
  31
  32 type TorrentStat struct {
  33         stats   torrent.ConnStats
  34         rxSpeed int64
  35         txSpeed int64
  36 }
  37
  38 var (
  39         TorrentStats  = map[metainfo.Hash]TorrentStat{}
  40         TorrentStatsM sync.RWMutex
  41 )
  42
  43 func recreateFIFO(pth string) {
  44         os.Remove(pth)
  45         if err := syscall.Mkfifo(pth, 0666); err != nil {
  46                 log.Fatalln(err)
  47         }
  48 }
  49
  50 func shortenName(name string) string {
  51         s := []rune(name)
  52         if len(s) > MaxListNameWidth {
  53                 s = s[:MaxListNameWidth]
  54         }
  55         return string(s)
  56 }
  57
  58 func fifoList(c *torrent.Client) {
  59         pth := path.Join(FIFOsDir, "list")
  60         recreateFIFO(pth)
  61         for {
  62                 fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
  63                 if err != nil {
  64                         log.Println("OpenFile:", pth, err)
  65                         time.Sleep(time.Second)
  66                         continue
  67                 }
  68                 ts := c.Torrents()
  69                 sort.Sort(ByInfoHash(ts))
  70                 for _, t := range ts {
  71                         if t.Info() == nil {
  72                                 fmt.Fprintf(fd, "%s not ready\n", t.Name())
  73                                 continue
  74                         }
  75                         stats := t.Stats()
  76                         done := t.BytesCompleted() * 100 / t.Length()
  77                         percColour := Red
  78                         if done == 100 {
  79                                 percColour = Green
  80                         }
  81                         tx := stats.BytesWrittenData.Int64()
  82                         tx += TxStats[t.InfoHash()]
  83                         ratio := float64(tx) / float64(t.Length())
  84                         TorrentStatsM.RLock()
  85                         prev := TorrentStats[t.InfoHash()]
  86                         TorrentStatsM.RUnlock()
  87                         var eta string
  88                         if done < 100 && prev.rxSpeed > 0 {
  89                                 etaRaw := time.Duration((t.Length() - t.BytesCompleted()) / prev.rxSpeed)
  90                                 etaRaw *= time.Second
  91                                 eta = etaRaw.String()
  92                         }
  93                         fmt.Fprintf(fd,
  94                                 "%s%s%s %s%40s%s %8s %s%3d%%%s %4.1f %s%d%s/%s%d%s %d/%d/%d/%d %s\n",
  95                                 Blue, t.InfoHash().HexString(), Reset,
  96                                 Green, shortenName(t.Name()), Reset,
  97                                 humanize.IBytes(uint64(t.Length())),
  98                                 percColour, done, Reset,
  99                                 ratio,
 100                                 Green, prev.rxSpeed/1024, Reset,
 101                                 Magenta, prev.txSpeed/1024, Reset,
 102                                 stats.TotalPeers,
 103                                 stats.PendingPeers,
 104                                 stats.ActivePeers,
 105                                 stats.ConnectedSeeders,
 106                                 eta,
 107                         )
 108                 }
 109                 fd.Close()
 110                 time.Sleep(time.Second)
 111         }
 112 }
 113
 114 func mustParseInt(s string) int {
 115         i, err := strconv.Atoi(s)
 116         if err != nil {
 117                 log.Fatalln(err)
 118         }
 119         return i
 120 }
 121
 122 func fifoPeerList(t *torrent.Torrent) {
 123         pth := path.Join(FIFOsDir, PeersDir, t.InfoHash().HexString())
 124         recreateFIFO(pth)
 125         for {
 126                 fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
 127                 if err != nil {
 128                         if os.IsNotExist(err) {
 129                                 break
 130                         }
 131                         log.Println("OpenFile:", pth, err)
 132                         time.Sleep(time.Second)
 133                         continue
 134                 }
 135                 pcs := t.PeerConns()
 136                 sort.Sort(ByPeerID(pcs))
 137                 for _, pc := range pcs {
 138                         cols := strings.Split(pc.CompletedString(), "/")
 139                         done := (mustParseInt(cols[0]) * 100) / mustParseInt(cols[1])
 140                         doneColour := Red
 141                         if done == 100 {
 142                                 doneColour = Green
 143                         }
 144                         stats := pc.Peer.Stats()
 145                         fmt.Fprintf(fd,
 146                                 "%s%s%s %10s %s%3d%%%s %s%d%s/%s%d%s %s / %s | %s%s%s %q\n",
 147                                 Blue, hex.EncodeToString(pc.PeerID[:]), Reset,
 148                                 pc.StatusFlags(),
 149                                 doneColour, done, Reset,
 150                                 Green, int(pc.DownloadRate()/1024), Reset,
 151                                 Magenta, int(pc.UploadRate()/1024), Reset,
 152                                 humanize.IBytes(uint64(stats.BytesReadData.Int64())),
 153                                 humanize.IBytes(uint64(stats.BytesWrittenData.Int64())),
 154                                 Green, pc.RemoteAddr, Reset,
 155                                 pc.PeerClientName,
 156                         )
 157                 }
 158                 fd.Close()
 159                 time.Sleep(time.Second)
 160         }
 161 }
 162
 163 func fifoFileList(t *torrent.Torrent) {
 164         pth := path.Join(FIFOsDir, FilesDir, t.InfoHash().HexString())
 165         recreateFIFO(pth)
 166         for {
 167                 fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
 168                 if err != nil {
 169                         if os.IsNotExist(err) {
 170                                 break
 171                         }
 172                         log.Println("OpenFile:", pth, err)
 173                         time.Sleep(time.Second)
 174                         continue
 175                 }
 176                 for n, f := range t.Files() {
 177                         var done int64
 178                         if f.Length() > 0 {
 179                                 done = (f.BytesCompleted() * 100) / f.Length()
 180                         }
 181                         percColour := Green
 182                         if done < 100 {
 183                                 percColour = Red
 184                         }
 185                         fmt.Fprintf(fd,
 186                                 "%5d %8s %3d%% | %s%s%s\n",
 187                                 n, humanize.IBytes(uint64(f.Length())), done,
 188                                 percColour, f.Path(), Reset,
 189                         )
 190                 }
 191                 fd.Close()
 192                 time.Sleep(time.Second)
 193         }
 194 }
 195
 196 func fifoDHTList(c *torrent.Client) {
 197         pth := path.Join(FIFOsDir, "dht")
 198         recreateFIFO(pth)
 199         for {
 200                 fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
 201                 if err != nil {
 202                         if os.IsNotExist(err) {
 203                                 break
 204                         }
 205                         log.Println("OpenFile:", pth, err)
 206                         time.Sleep(time.Second)
 207                         continue
 208                 }
 209                 for _, s := range c.DhtServers() {
 210                         stats := s.Stats().(dht.ServerStats)
 211                         fmt.Fprintf(
 212                                 fd, "%s%s%s all:%d good:%d await:%d succ:%d bad:%d\n",
 213                                 Green, s.Addr().String(), Reset,
 214                                 stats.Nodes,
 215                                 stats.GoodNodes,
 216                                 stats.OutstandingTransactions,
 217                                 stats.SuccessfulOutboundAnnouncePeerQueries,
 218                                 stats.BadNodes,
 219                         )
 220                 }
 221                 fd.Close()
 222                 time.Sleep(time.Second)
 223         }
 224 }
 225
 226 type topTorrent struct {
 227         infoHash metainfo.Hash
 228         name     string
 229         tx       int64
 230 }
 231
 232 func fifoTopSeed(c *torrent.Client) {
 233         pth := path.Join(FIFOsDir, "top-seed")
 234         recreateFIFO(pth)
 235         for {
 236                 fd, err := os.OpenFile(pth, os.O_WRONLY|os.O_APPEND, os.FileMode(0666))
 237                 if err != nil {
 238                         log.Println("OpenFile:", pth, err)
 239                         time.Sleep(time.Second)
 240                         continue
 241                 }
 242                 var ts []*topTorrent
 243                 for _, t := range c.Torrents() {
 244                         if t.Info() == nil {
 245                                 continue
 246                         }
 247                         stats := t.Stats()
 248                         ts = append(ts, &topTorrent{
 249                                 infoHash: t.InfoHash(),
 250                                 name:     t.Name(),
 251                                 tx:       stats.BytesWrittenData.Int64() + TxStats[t.InfoHash()],
 252                         })
 253                 }
 254                 sort.Sort(ByTxTraffic(ts))
 255                 for _, t := range ts {
 256                         fmt.Fprintf(fd,
 257                                 "%s%s%s %s%40s%s %s\n",
 258                                 Blue, t.infoHash.HexString(), Reset,
 259                                 Green, shortenName(t.name), Reset,
 260                                 humanize.IBytes(uint64(t.tx)),
 261                         )
 262                 }
 263                 fd.Close()
 264                 time.Sleep(time.Second)
 265         }
 266 }
 267
 268 type stringAddr string
 269
 270 func (stringAddr) Network() string   { return "" }
 271 func (me stringAddr) String() string { return string(me) }
 272
 273 func resolveTestPeers(addrs []string) (ret []torrent.PeerInfo) {
 274         for _, ta := range addrs {
 275                 ret = append(ret, torrent.PeerInfo{Addr: stringAddr(ta)})
 276         }
 277         return
 278 }
 279
 280 func readLinesFromFIFO(pth string) []string {
 281         fd, err := os.OpenFile(pth, os.O_RDONLY, os.FileMode(0666))
 282         if err != nil {
 283                 log.Println("OpenFile:", pth, err)
 284                 time.Sleep(time.Second)
 285                 return nil
 286         }
 287         var lines []string
 288         scanner := bufio.NewScanner(fd)
 289         for scanner.Scan() {
 290                 t := scanner.Text()
 291                 if len(t) > 0 {
 292                         lines = append(lines, t)
 293                 }
 294         }
 295         fd.Close()
 296         return lines
 297 }
 298
 299 func saveTorrent(t *torrent.Torrent) error {
 300         pth := t.Name() + TorrentExt
 301         if _, err := os.Stat(pth); err == nil {
 302                 return nil
 303         }
 304         var b bytes.Buffer
 305         t.Metainfo().Write(&b)
 306         return os.WriteFile(pth, b.Bytes(), 0666)
 307 }
 308
 309 func fifoAdd(c *torrent.Client) {
 310         pth := path.Join(FIFOsDir, "add")
 311         recreateFIFO(pth)
 312         for {
 313                 for _, what := range readLinesFromFIFO(pth) {
 314                         cols := strings.Fields(what)
 315                         what = cols[0]
 316                         var t *torrent.Torrent
 317                         var err error
 318                         if strings.HasPrefix(what, "magnet:") {
 319                                 t, err = c.AddMagnet(what)
 320                                 if err != nil {
 321                                         log.Println("AddMagnet:", what, err)
 322                                         continue
 323                                 }
 324                         } else {
 325                                 metaInfo, err := metainfo.LoadFromFile(what)
 326                                 if err != nil {
 327                                         log.Println("LoadFromFile:", what, err)
 328                                         continue
 329                                 }
 330                                 t, err = c.AddTorrent(metaInfo)
 331                                 if err != nil {
 332                                         log.Println("AddTorrent:", what, err)
 333                                         continue
 334                                 }
 335                         }
 336                         if len(cols) > 1 {
 337                                 t.AddPeers(resolveTestPeers(cols[1:]))
 338                         }
 339                         go fifoPeerList(t)
 340                         go fifoFileList(t)
 341                         log.Println("added:", t.InfoHash().HexString(), t.Name())
 342                         go func() {
 343                                 <-t.GotInfo()
 344                                 if err = saveTorrent(t); err != nil {
 345                                         log.Println("saveTorrent:", err)
 346                                 }
 347                                 txStatsLoad(t.InfoHash())
 348                                 t.DownloadAll()
 349                         }()
 350                 }
 351                 time.Sleep(time.Second)
 352         }
 353 }
 354
 355 func fifoDel(c *torrent.Client) {
 356         pth := path.Join(FIFOsDir, "del")
 357         recreateFIFO(pth)
 358         for {
 359                 for _, what := range readLinesFromFIFO(pth) {
 360                         raw, err := hex.DecodeString(what)
 361                         if err != nil {
 362                                 log.Println(err)
 363                                 continue
 364                         }
 365                         if len(raw) != infohash.Size {
 366                                 log.Println("bad length")
 367                                 continue
 368                         }
 369                         var i infohash.T
 370                         copy(i[:], raw)
 371                         t, ok := c.Torrent(i)
 372                         if !ok {
 373                                 log.Println("no suck torrent", what)
 374                                 continue
 375                         }
 376                         txStatsDump(t)
 377                         txStatsDel(t.InfoHash())
 378                         t.Drop()
 379                         for _, where := range []string{"files", "peers"} {
 380                                 pth := path.Join(where, t.InfoHash().HexString())
 381                                 os.Remove(pth)
 382                                 fd, err := os.Open(pth)
 383                                 if err == nil {
 384                                         fd.Close()
 385                                 }
 386                         }
 387                         log.Println("deleted:", what, t.Name())
 388                 }
 389                 time.Sleep(time.Second)
 390         }
 391 }
 392
 393 func fifosPrepare() {
 394         os.MkdirAll(path.Join(FIFOsDir, PeersDir), 0777)
 395         os.MkdirAll(path.Join(FIFOsDir, FilesDir), 0777)
 396 }
 397
 398 func fifosCleanup() {
 399         os.Remove(path.Join(FIFOsDir, "list"))
 400         os.Remove(path.Join(FIFOsDir, "dht"))
 401         os.Remove(path.Join(FIFOsDir, "add"))
 402         os.Remove(path.Join(FIFOsDir, "del"))
 403         os.Remove(path.Join(FIFOsDir, "top-seed"))
 404         os.RemoveAll(path.Join(FIFOsDir, PeersDir))
 405         os.RemoveAll(path.Join(FIFOsDir, FilesDir))
 406 }