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