client.go

   1 package torrent
   2
   3 import (
   4         "bufio"
   5         "bytes"
   6         "context"
   7         "crypto/rand"
   8         "encoding/binary"
   9         "errors"
  10         "fmt"
  11         "io"
  12         "net"
  13         "net/http"
  14         "net/url"
  15         "strconv"
  16         "strings"
  17         "time"
  18
  19         "github.com/anacrolix/dht"
  20         "github.com/anacrolix/dht/krpc"
  21         "github.com/anacrolix/log"
  22         "github.com/anacrolix/missinggo"
  23         "github.com/anacrolix/missinggo/bitmap"
  24         "github.com/anacrolix/missinggo/conntrack"
  25         "github.com/anacrolix/missinggo/perf"
  26         "github.com/anacrolix/missinggo/pproffd"
  27         "github.com/anacrolix/missinggo/pubsub"
  28         "github.com/anacrolix/missinggo/slices"
  29         "github.com/anacrolix/sync"
  30         "github.com/anacrolix/torrent/bencode"
  31         "github.com/anacrolix/torrent/iplist"
  32         "github.com/anacrolix/torrent/metainfo"
  33         "github.com/anacrolix/torrent/mse"
  34         pp "github.com/anacrolix/torrent/peer_protocol"
  35         "github.com/anacrolix/torrent/storage"
  36         "github.com/davecgh/go-spew/spew"
  37         "github.com/dustin/go-humanize"
  38         "github.com/google/btree"
  39         "golang.org/x/time/rate"
  40 )
  41
  42 // Clients contain zero or more Torrents. A Client manages a blocklist, the
  43 // TCP/UDP protocol ports, and DHT as desired.
  44 type Client struct {
  45         // An aggregate of stats over all connections. First in struct to ensure
  46         // 64-bit alignment of fields. See #262.
  47         stats ConnStats
  48
  49         _mu    sync.RWMutex
  50         event  sync.Cond
  51         closed missinggo.Event
  52
  53         config *ClientConfig
  54         logger *log.Logger
  55
  56         peerID         PeerID
  57         defaultStorage *storage.Client
  58         onClose        []func()
  59         conns          []socket
  60         dhtServers     []*dht.Server
  61         ipBlockList    iplist.Ranger
  62         // Our BitTorrent protocol extension bytes, sent in our BT handshakes.
  63         extensionBytes pp.PeerExtensionBits
  64
  65         // Set of addresses that have our client ID. This intentionally will
  66         // include ourselves if we end up trying to connect to our own address
  67         // through legitimate channels.
  68         dopplegangerAddrs map[string]struct{}
  69         badPeerIPs        map[string]struct{}
  70         torrents          map[InfoHash]*Torrent
  71
  72         acceptLimiter   map[ipStr]int
  73         dialRateLimiter *rate.Limiter
  74 }
  75
  76 type ipStr string
  77
  78 func (cl *Client) BadPeerIPs() []string {
  79         cl.rLock()
  80         defer cl.rUnlock()
  81         return cl.badPeerIPsLocked()
  82 }
  83
  84 func (cl *Client) badPeerIPsLocked() []string {
  85         return slices.FromMapKeys(cl.badPeerIPs).([]string)
  86 }
  87
  88 func (cl *Client) PeerID() PeerID {
  89         return cl.peerID
  90 }
  91
  92 func (cl *Client) LocalPort() (port int) {
  93         cl.eachListener(func(l socket) bool {
  94                 _port := missinggo.AddrPort(l.Addr())
  95                 if _port == 0 {
  96                         panic(l)
  97                 }
  98                 if port == 0 {
  99                         port = _port
 100                 } else if port != _port {
 101                         panic("mismatched ports")
 102                 }
 103                 return true
 104         })
 105         return
 106 }
 107
 108 func writeDhtServerStatus(w io.Writer, s *dht.Server) {
 109         dhtStats := s.Stats()
 110         fmt.Fprintf(w, "\t# Nodes: %d (%d good, %d banned)\n", dhtStats.Nodes, dhtStats.GoodNodes, dhtStats.BadNodes)
 111         fmt.Fprintf(w, "\tServer ID: %x\n", s.ID())
 112         fmt.Fprintf(w, "\tAnnounces: %d\n", dhtStats.SuccessfulOutboundAnnouncePeerQueries)
 113         fmt.Fprintf(w, "\tOutstanding transactions: %d\n", dhtStats.OutstandingTransactions)
 114 }
 115
 116 // Writes out a human readable status of the client, such as for writing to a
 117 // HTTP status page.
 118 func (cl *Client) WriteStatus(_w io.Writer) {
 119         cl.rLock()
 120         defer cl.rUnlock()
 121         w := bufio.NewWriter(_w)
 122         defer w.Flush()
 123         fmt.Fprintf(w, "Listen port: %d\n", cl.LocalPort())
 124         fmt.Fprintf(w, "Peer ID: %+q\n", cl.PeerID())
 125         fmt.Fprintf(w, "Announce key: %x\n", cl.announceKey())
 126         fmt.Fprintf(w, "Banned IPs: %d\n", len(cl.badPeerIPsLocked()))
 127         cl.eachDhtServer(func(s *dht.Server) {
 128                 fmt.Fprintf(w, "%s DHT server at %s:\n", s.Addr().Network(), s.Addr().String())
 129                 writeDhtServerStatus(w, s)
 130         })
 131         spew.Fdump(w, cl.stats)
 132         fmt.Fprintf(w, "# Torrents: %d\n", len(cl.torrentsAsSlice()))
 133         fmt.Fprintln(w)
 134         for _, t := range slices.Sort(cl.torrentsAsSlice(), func(l, r *Torrent) bool {
 135                 return l.InfoHash().AsString() < r.InfoHash().AsString()
 136         }).([]*Torrent) {
 137                 if t.name() == "" {
 138                         fmt.Fprint(w, "<unknown name>")
 139                 } else {
 140                         fmt.Fprint(w, t.name())
 141                 }
 142                 fmt.Fprint(w, "\n")
 143                 if t.info != nil {
 144                         fmt.Fprintf(w, "%f%% of %d bytes (%s)", 100*(1-float64(t.bytesMissingLocked())/float64(t.info.TotalLength())), t.length, humanize.Bytes(uint64(t.info.TotalLength())))
 145                 } else {
 146                         w.WriteString("<missing metainfo>")
 147                 }
 148                 fmt.Fprint(w, "\n")
 149                 t.writeStatus(w)
 150                 fmt.Fprintln(w)
 151         }
 152 }
 153
 154 const debugLogValue = "debug"
 155
 156 func (cl *Client) debugLogFilter(m *log.Msg) bool {
 157         if !cl.config.Debug {
 158                 _, ok := m.Values()[debugLogValue]
 159                 return !ok
 160         }
 161         return true
 162 }
 163
 164 func (cl *Client) initLogger() {
 165         cl.logger = log.Default.Clone().AddValue(cl).AddFilter(log.NewFilter(cl.debugLogFilter))
 166 }
 167
 168 func (cl *Client) announceKey() int32 {
 169         return int32(binary.BigEndian.Uint32(cl.peerID[16:20]))
 170 }
 171
 172 func NewClient(cfg *ClientConfig) (cl *Client, err error) {
 173         if cfg == nil {
 174                 cfg = NewDefaultClientConfig()
 175         }
 176         defer func() {
 177                 if err != nil {
 178                         cl = nil
 179                 }
 180         }()
 181         cl = &Client{
 182                 config:            cfg,
 183                 dopplegangerAddrs: make(map[string]struct{}),
 184                 torrents:          make(map[metainfo.Hash]*Torrent),
 185                 dialRateLimiter:   rate.NewLimiter(10, 10),
 186         }
 187         go cl.acceptLimitClearer()
 188         cl.initLogger()
 189         defer func() {
 190                 if err == nil {
 191                         return
 192                 }
 193                 cl.Close()
 194         }()
 195         cl.extensionBytes = defaultPeerExtensionBytes()
 196         cl.event.L = cl.locker()
 197         storageImpl := cfg.DefaultStorage
 198         if storageImpl == nil {
 199                 // We'd use mmap but HFS+ doesn't support sparse files.
 200                 storageImpl = storage.NewFile(cfg.DataDir)
 201                 cl.onClose = append(cl.onClose, func() {
 202                         if err := storageImpl.Close(); err != nil {
 203                                 log.Printf("error closing default storage: %s", err)
 204                         }
 205                 })
 206         }
 207         cl.defaultStorage = storage.NewClient(storageImpl)
 208         if cfg.IPBlocklist != nil {
 209                 cl.ipBlockList = cfg.IPBlocklist
 210         }
 211
 212         if cfg.PeerID != "" {
 213                 missinggo.CopyExact(&cl.peerID, cfg.PeerID)
 214         } else {
 215                 o := copy(cl.peerID[:], cfg.Bep20)
 216                 _, err = rand.Read(cl.peerID[o:])
 217                 if err != nil {
 218                         panic("error generating peer id")
 219                 }
 220         }
 221
 222         if cl.config.HTTPProxy == nil && cl.config.ProxyURL != "" {
 223                 if fixedURL, err := url.Parse(cl.config.ProxyURL); err == nil {
 224                         cl.config.HTTPProxy = http.ProxyURL(fixedURL)
 225                 }
 226         }
 227
 228         cl.conns, err = listenAll(cl.listenNetworks(), cl.config.ListenHost, cl.config.ListenPort, cl.config.ProxyURL, cl.firewallCallback)
 229         if err != nil {
 230                 return
 231         }
 232         // Check for panics.
 233         cl.LocalPort()
 234
 235         for _, s := range cl.conns {
 236                 if peerNetworkEnabled(parseNetworkString(s.Addr().Network()), cl.config) {
 237                         go cl.acceptConnections(s)
 238                 }
 239         }
 240
 241         go cl.forwardPort()
 242         if !cfg.NoDHT {
 243                 for _, s := range cl.conns {
 244                         if pc, ok := s.(net.PacketConn); ok {
 245                                 ds, err := cl.newDhtServer(pc)
 246                                 if err != nil {
 247                                         panic(err)
 248                                 }
 249                                 cl.dhtServers = append(cl.dhtServers, ds)
 250                         }
 251                 }
 252         }
 253
 254         return
 255 }
 256
 257 func (cl *Client) firewallCallback(net.Addr) bool {
 258         cl.rLock()
 259         block := !cl.wantConns()
 260         cl.rUnlock()
 261         if block {
 262                 torrent.Add("connections firewalled", 1)
 263         } else {
 264                 torrent.Add("connections not firewalled", 1)
 265         }
 266         return block
 267 }
 268
 269 func (cl *Client) enabledPeerNetworks() (ns []network) {
 270         for _, n := range allPeerNetworks {
 271                 if peerNetworkEnabled(n, cl.config) {
 272                         ns = append(ns, n)
 273                 }
 274         }
 275         return
 276 }
 277
 278 func (cl *Client) listenOnNetwork(n network) bool {
 279         if n.Ipv4 && cl.config.DisableIPv4 {
 280                 return false
 281         }
 282         if n.Ipv6 && cl.config.DisableIPv6 {
 283                 return false
 284         }
 285         if n.Tcp && cl.config.DisableTCP {
 286                 return false
 287         }
 288         if n.Udp && cl.config.DisableUTP && cl.config.NoDHT {
 289                 return false
 290         }
 291         return true
 292 }
 293
 294 func (cl *Client) listenNetworks() (ns []network) {
 295         for _, n := range allPeerNetworks {
 296                 if cl.listenOnNetwork(n) {
 297                         ns = append(ns, n)
 298                 }
 299         }
 300         return
 301 }
 302
 303 func (cl *Client) newDhtServer(conn net.PacketConn) (s *dht.Server, err error) {
 304         cfg := dht.ServerConfig{
 305                 IPBlocklist:    cl.ipBlockList,
 306                 Conn:           conn,
 307                 OnAnnouncePeer: cl.onDHTAnnouncePeer,
 308                 PublicIP: func() net.IP {
 309                         if connIsIpv6(conn) && cl.config.PublicIp6 != nil {
 310                                 return cl.config.PublicIp6
 311                         }
 312                         return cl.config.PublicIp4
 313                 }(),
 314                 StartingNodes:      cl.config.DhtStartingNodes,
 315                 ConnectionTracking: cl.config.ConnTracker,
 316         }
 317         s, err = dht.NewServer(&cfg)
 318         if err == nil {
 319                 go func() {
 320                         ts, err := s.Bootstrap()
 321                         if err != nil {
 322                                 log.Printf("error bootstrapping dht: %s", err)
 323                         }
 324                         log.Printf("%s: dht bootstrap: %v", s, ts)
 325                 }()
 326         }
 327         return
 328 }
 329
 330 func (cl *Client) Closed() <-chan struct{} {
 331         cl.lock()
 332         defer cl.unlock()
 333         return cl.closed.C()
 334 }
 335
 336 func (cl *Client) eachDhtServer(f func(*dht.Server)) {
 337         for _, ds := range cl.dhtServers {
 338                 f(ds)
 339         }
 340 }
 341
 342 func (cl *Client) closeSockets() {
 343         cl.eachListener(func(l socket) bool {
 344                 l.Close()
 345                 return true
 346         })
 347         cl.conns = nil
 348 }
 349
 350 // Stops the client. All connections to peers are closed and all activity will
 351 // come to a halt.
 352 func (cl *Client) Close() {
 353         cl.lock()
 354         defer cl.unlock()
 355         cl.closed.Set()
 356         cl.eachDhtServer(func(s *dht.Server) { s.Close() })
 357         cl.closeSockets()
 358         for _, t := range cl.torrents {
 359                 t.close()
 360         }
 361         for _, f := range cl.onClose {
 362                 f()
 363         }
 364         cl.event.Broadcast()
 365 }
 366
 367 func (cl *Client) ipBlockRange(ip net.IP) (r iplist.Range, blocked bool) {
 368         if cl.ipBlockList == nil {
 369                 return
 370         }
 371         return cl.ipBlockList.Lookup(ip)
 372 }
 373
 374 func (cl *Client) ipIsBlocked(ip net.IP) bool {
 375         _, blocked := cl.ipBlockRange(ip)
 376         return blocked
 377 }
 378
 379 func (cl *Client) wantConns() bool {
 380         for _, t := range cl.torrents {
 381                 if t.wantConns() {
 382                         return true
 383                 }
 384         }
 385         return false
 386 }
 387
 388 func (cl *Client) waitAccept() {
 389         for {
 390                 if cl.closed.IsSet() {
 391                         return
 392                 }
 393                 if cl.wantConns() {
 394                         return
 395                 }
 396                 cl.event.Wait()
 397         }
 398 }
 399
 400 func (cl *Client) rejectAccepted(conn net.Conn) bool {
 401         ra := conn.RemoteAddr()
 402         rip := missinggo.AddrIP(ra)
 403         if cl.config.DisableIPv4Peers && rip.To4() != nil {
 404                 return true
 405         }
 406         if cl.config.DisableIPv4 && len(rip) == net.IPv4len {
 407                 return true
 408         }
 409         if cl.config.DisableIPv6 && len(rip) == net.IPv6len && rip.To4() == nil {
 410                 return true
 411         }
 412         if cl.rateLimitAccept(rip) {
 413                 return true
 414         }
 415         return cl.badPeerIPPort(rip, missinggo.AddrPort(ra))
 416 }
 417
 418 func (cl *Client) acceptConnections(l net.Listener) {
 419         for {
 420                 conn, err := l.Accept()
 421                 conn = pproffd.WrapNetConn(conn)
 422                 cl.rLock()
 423                 closed := cl.closed.IsSet()
 424                 reject := false
 425                 if conn != nil {
 426                         reject = cl.rejectAccepted(conn)
 427                 }
 428                 cl.rUnlock()
 429                 if closed {
 430                         if conn != nil {
 431                                 conn.Close()
 432                         }
 433                         return
 434                 }
 435                 if err != nil {
 436                         log.Printf("error accepting connection: %s", err)
 437                         continue
 438                 }
 439                 go func() {
 440                         if reject {
 441                                 torrent.Add("rejected accepted connections", 1)
 442                                 conn.Close()
 443                         } else {
 444                                 go cl.incomingConnection(conn)
 445                         }
 446                         log.Fmsg("accepted %s connection from %s", conn.RemoteAddr().Network(), conn.RemoteAddr()).AddValue(debugLogValue).Log(cl.logger)
 447                         torrent.Add(fmt.Sprintf("accepted conn remote IP len=%d", len(missinggo.AddrIP(conn.RemoteAddr()))), 1)
 448                         torrent.Add(fmt.Sprintf("accepted conn network=%s", conn.RemoteAddr().Network()), 1)
 449                         torrent.Add(fmt.Sprintf("accepted on %s listener", l.Addr().Network()), 1)
 450                 }()
 451         }
 452 }
 453
 454 func (cl *Client) incomingConnection(nc net.Conn) {
 455         defer nc.Close()
 456         if tc, ok := nc.(*net.TCPConn); ok {
 457                 tc.SetLinger(0)
 458         }
 459         c := cl.newConnection(nc, false, missinggo.IpPortFromNetAddr(nc.RemoteAddr()), nc.RemoteAddr().Network())
 460         c.Discovery = peerSourceIncoming
 461         cl.runReceivedConn(c)
 462 }
 463
 464 // Returns a handle to the given torrent, if it's present in the client.
 465 func (cl *Client) Torrent(ih metainfo.Hash) (t *Torrent, ok bool) {
 466         cl.lock()
 467         defer cl.unlock()
 468         t, ok = cl.torrents[ih]
 469         return
 470 }
 471
 472 func (cl *Client) torrent(ih metainfo.Hash) *Torrent {
 473         return cl.torrents[ih]
 474 }
 475
 476 type dialResult struct {
 477         Conn    net.Conn
 478         Network string
 479 }
 480
 481 func countDialResult(err error) {
 482         if err == nil {
 483                 torrent.Add("successful dials", 1)
 484         } else {
 485                 torrent.Add("unsuccessful dials", 1)
 486         }
 487 }
 488
 489 func reducedDialTimeout(minDialTimeout, max time.Duration, halfOpenLimit int, pendingPeers int) (ret time.Duration) {
 490         ret = max / time.Duration((pendingPeers+halfOpenLimit)/halfOpenLimit)
 491         if ret < minDialTimeout {
 492                 ret = minDialTimeout
 493         }
 494         return
 495 }
 496
 497 // Returns whether an address is known to connect to a client with our own ID.
 498 func (cl *Client) dopplegangerAddr(addr string) bool {
 499         _, ok := cl.dopplegangerAddrs[addr]
 500         return ok
 501 }
 502
 503 // Returns a connection over UTP or TCP, whichever is first to connect.
 504 func (cl *Client) dialFirst(ctx context.Context, addr string) dialResult {
 505         ctx, cancel := context.WithCancel(ctx)
 506         // As soon as we return one connection, cancel the others.
 507         defer cancel()
 508         left := 0
 509         resCh := make(chan dialResult, left)
 510         func() {
 511                 cl.lock()
 512                 defer cl.unlock()
 513                 cl.eachListener(func(s socket) bool {
 514                         network := s.Addr().Network()
 515                         if peerNetworkEnabled(parseNetworkString(network), cl.config) {
 516                                 left++
 517                                 go func() {
 518                                         cte := cl.config.ConnTracker.Wait(
 519                                                 conntrack.Entry{network, s.Addr().String(), addr},
 520                                                 "dial torrent client",
 521                                                 0,
 522                                         )
 523                                         c, err := s.dial(ctx, addr)
 524                                         // This is a bit optimistic, but it looks non-trivial to thread
 525                                         // this through the proxy code. Set it now in case we close the
 526                                         // connection forthwith.
 527                                         if tc, ok := c.(*net.TCPConn); ok {
 528                                                 tc.SetLinger(0)
 529                                         }
 530                                         countDialResult(err)
 531                                         dr := dialResult{c, network}
 532                                         if c == nil {
 533                                                 cte.Done()
 534                                         } else {
 535                                                 dr.Conn = closeWrapper{c, func() error {
 536                                                         err := c.Close()
 537                                                         cte.Done()
 538                                                         return err
 539                                                 }}
 540                                         }
 541                                         resCh <- dr
 542                                 }()
 543                         }
 544                         return true
 545                 })
 546         }()
 547         var res dialResult
 548         // Wait for a successful connection.
 549         func() {
 550                 defer perf.ScopeTimer()()
 551                 for ; left > 0 && res.Conn == nil; left-- {
 552                         res = <-resCh
 553                 }
 554         }()
 555         // There are still incompleted dials.
 556         go func() {
 557                 for ; left > 0; left-- {
 558                         conn := (<-resCh).Conn
 559                         if conn != nil {
 560                                 conn.Close()
 561                         }
 562                 }
 563         }()
 564         if res.Conn != nil {
 565                 go torrent.Add(fmt.Sprintf("network dialed first: %s", res.Conn.RemoteAddr().Network()), 1)
 566         }
 567         return res
 568 }
 569
 570 func (cl *Client) noLongerHalfOpen(t *Torrent, addr string) {
 571         if _, ok := t.halfOpen[addr]; !ok {
 572                 panic("invariant broken")
 573         }
 574         delete(t.halfOpen, addr)
 575         t.openNewConns()
 576 }
 577
 578 // Performs initiator handshakes and returns a connection. Returns nil
 579 // *connection if no connection for valid reasons.
 580 func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torrent, encryptHeader bool, remoteAddr IpPort, network string) (c *connection, err error) {
 581         c = cl.newConnection(nc, true, remoteAddr, network)
 582         c.headerEncrypted = encryptHeader
 583         ctx, cancel := context.WithTimeout(ctx, cl.config.HandshakesTimeout)
 584         defer cancel()
 585         dl, ok := ctx.Deadline()
 586         if !ok {
 587                 panic(ctx)
 588         }
 589         err = nc.SetDeadline(dl)
 590         if err != nil {
 591                 panic(err)
 592         }
 593         ok, err = cl.initiateHandshakes(c, t)
 594         if !ok {
 595                 c = nil
 596         }
 597         return
 598 }
 599
 600 // Returns nil connection and nil error if no connection could be established
 601 // for valid reasons.
 602 func (cl *Client) establishOutgoingConnEx(t *Torrent, addr IpPort, ctx context.Context, obfuscatedHeader bool) (c *connection, err error) {
 603         dr := cl.dialFirst(ctx, addr.String())
 604         nc := dr.Conn
 605         if nc == nil {
 606                 return
 607         }
 608         defer func() {
 609                 if c == nil || err != nil {
 610                         nc.Close()
 611                 }
 612         }()
 613         return cl.handshakesConnection(ctx, nc, t, obfuscatedHeader, addr, dr.Network)
 614 }
 615
 616 // Returns nil connection and nil error if no connection could be established
 617 // for valid reasons.
 618 func (cl *Client) establishOutgoingConn(t *Torrent, addr IpPort) (c *connection, err error) {
 619         torrent.Add("establish outgoing connection", 1)
 620         ctx, cancel := context.WithTimeout(context.Background(), func() time.Duration {
 621                 cl.rLock()
 622                 defer cl.rUnlock()
 623                 return t.dialTimeout()
 624         }())
 625         defer cancel()
 626         obfuscatedHeaderFirst := !cl.config.DisableEncryption && !cl.config.PreferNoEncryption
 627         c, err = cl.establishOutgoingConnEx(t, addr, ctx, obfuscatedHeaderFirst)
 628         if err != nil {
 629                 return
 630         }
 631         if c != nil {
 632                 torrent.Add("initiated conn with preferred header obfuscation", 1)
 633                 return
 634         }
 635         if cl.config.ForceEncryption {
 636                 // We should have just tried with an obfuscated header. A plaintext
 637                 // header can't result in an encrypted connection, so we're done.
 638                 if !obfuscatedHeaderFirst {
 639                         panic(cl.config.EncryptionPolicy)
 640                 }
 641                 return
 642         }
 643         // Try again with encryption if we didn't earlier, or without if we did.
 644         c, err = cl.establishOutgoingConnEx(t, addr, ctx, !obfuscatedHeaderFirst)
 645         if c != nil {
 646                 torrent.Add("initiated conn with fallback header obfuscation", 1)
 647         }
 648         return
 649 }
 650
 651 // Called to dial out and run a connection. The addr we're given is already
 652 // considered half-open.
 653 func (cl *Client) outgoingConnection(t *Torrent, addr IpPort, ps peerSource) {
 654         cl.dialRateLimiter.Wait(context.Background())
 655         c, err := cl.establishOutgoingConn(t, addr)
 656         cl.lock()
 657         defer cl.unlock()
 658         // Don't release lock between here and addConnection, unless it's for
 659         // failure.
 660         cl.noLongerHalfOpen(t, addr.String())
 661         if err != nil {
 662                 if cl.config.Debug {
 663                         log.Printf("error establishing outgoing connection: %s", err)
 664                 }
 665                 return
 666         }
 667         if c == nil {
 668                 return
 669         }
 670         defer c.Close()
 671         c.Discovery = ps
 672         cl.runHandshookConn(c, t)
 673 }
 674
 675 // The port number for incoming peer connections. 0 if the client isn't
 676 // listening.
 677 func (cl *Client) incomingPeerPort() int {
 678         return cl.LocalPort()
 679 }
 680
 681 func (cl *Client) initiateHandshakes(c *connection, t *Torrent) (ok bool, err error) {
 682         if c.headerEncrypted {
 683                 var rw io.ReadWriter
 684                 rw, c.cryptoMethod, err = mse.InitiateHandshake(
 685                         struct {
 686                                 io.Reader
 687                                 io.Writer
 688                         }{c.r, c.w},
 689                         t.infoHash[:],
 690                         nil,
 691                         func() mse.CryptoMethod {
 692                                 switch {
 693                                 case cl.config.ForceEncryption:
 694                                         return mse.CryptoMethodRC4
 695                                 case cl.config.DisableEncryption:
 696                                         return mse.CryptoMethodPlaintext
 697                                 default:
 698                                         return mse.AllSupportedCrypto
 699                                 }
 700                         }(),
 701                 )
 702                 c.setRW(rw)
 703                 if err != nil {
 704                         return
 705                 }
 706         }
 707         ih, ok, err := cl.connBTHandshake(c, &t.infoHash)
 708         if ih != t.infoHash {
 709                 ok = false
 710         }
 711         return
 712 }
 713
 714 // Calls f with any secret keys.
 715 func (cl *Client) forSkeys(f func([]byte) bool) {
 716         cl.lock()
 717         defer cl.unlock()
 718         for ih := range cl.torrents {
 719                 if !f(ih[:]) {
 720                         break
 721                 }
 722         }
 723 }
 724
 725 // Do encryption and bittorrent handshakes as receiver.
 726 func (cl *Client) receiveHandshakes(c *connection) (t *Torrent, err error) {
 727         defer perf.ScopeTimerErr(&err)()
 728         var rw io.ReadWriter
 729         rw, c.headerEncrypted, c.cryptoMethod, err = handleEncryption(c.rw(), cl.forSkeys, cl.config.EncryptionPolicy)
 730         c.setRW(rw)
 731         if err == nil || err == mse.ErrNoSecretKeyMatch {
 732                 if c.headerEncrypted {
 733                         torrent.Add("handshakes received encrypted", 1)
 734                 } else {
 735                         torrent.Add("handshakes received unencrypted", 1)
 736                 }
 737         } else {
 738                 torrent.Add("handshakes received with error while handling encryption", 1)
 739         }
 740         if err != nil {
 741                 if err == mse.ErrNoSecretKeyMatch {
 742                         err = nil
 743                 }
 744                 return
 745         }
 746         if cl.config.ForceEncryption && !c.headerEncrypted {
 747                 err = errors.New("connection not encrypted")
 748                 return
 749         }
 750         ih, ok, err := cl.connBTHandshake(c, nil)
 751         if err != nil {
 752                 err = fmt.Errorf("error during bt handshake: %s", err)
 753                 return
 754         }
 755         if !ok {
 756                 return
 757         }
 758         cl.lock()
 759         t = cl.torrents[ih]
 760         cl.unlock()
 761         return
 762 }
 763
 764 // Returns !ok if handshake failed for valid reasons.
 765 func (cl *Client) connBTHandshake(c *connection, ih *metainfo.Hash) (ret metainfo.Hash, ok bool, err error) {
 766         res, ok, err := pp.Handshake(c.rw(), ih, cl.peerID, cl.extensionBytes)
 767         if err != nil || !ok {
 768                 return
 769         }
 770         ret = res.Hash
 771         c.PeerExtensionBytes = res.PeerExtensionBits
 772         c.PeerID = res.PeerID
 773         c.completedHandshake = time.Now()
 774         return
 775 }
 776
 777 func (cl *Client) runReceivedConn(c *connection) {
 778         err := c.conn.SetDeadline(time.Now().Add(cl.config.HandshakesTimeout))
 779         if err != nil {
 780                 panic(err)
 781         }
 782         t, err := cl.receiveHandshakes(c)
 783         if err != nil {
 784                 log.Fmsg(
 785                         "error receiving handshakes: %s", err,
 786                 ).AddValue(
 787                         debugLogValue,
 788                 ).Add(
 789                         "network", c.network,
 790                 ).Log(cl.logger)
 791                 torrent.Add("error receiving handshake", 1)
 792                 cl.lock()
 793                 cl.onBadAccept(c.remoteAddr)
 794                 cl.unlock()
 795                 return
 796         }
 797         if t == nil {
 798                 torrent.Add("received handshake for unloaded torrent", 1)
 799                 cl.lock()
 800                 cl.onBadAccept(c.remoteAddr)
 801                 cl.unlock()
 802                 return
 803         }
 804         torrent.Add("received handshake for loaded torrent", 1)
 805         cl.lock()
 806         defer cl.unlock()
 807         cl.runHandshookConn(c, t)
 808 }
 809
 810 func (cl *Client) runHandshookConn(c *connection, t *Torrent) {
 811         c.setTorrent(t)
 812         if c.PeerID == cl.peerID {
 813                 if c.outgoing {
 814                         connsToSelf.Add(1)
 815                         addr := c.conn.RemoteAddr().String()
 816                         cl.dopplegangerAddrs[addr] = struct{}{}
 817                 } else {
 818                         // Because the remote address is not necessarily the same as its
 819                         // client's torrent listen address, we won't record the remote address
 820                         // as a doppleganger. Instead, the initiator can record *us* as the
 821                         // doppleganger.
 822                 }
 823                 return
 824         }
 825         c.conn.SetWriteDeadline(time.Time{})
 826         c.r = deadlineReader{c.conn, c.r}
 827         completedHandshakeConnectionFlags.Add(c.connectionFlags(), 1)
 828         if connIsIpv6(c.conn) {
 829                 torrent.Add("completed handshake over ipv6", 1)
 830         }
 831         if err := t.addConnection(c); err != nil {
 832                 log.Fmsg("error adding connection: %s", err).AddValues(c, debugLogValue).Log(t.logger)
 833                 return
 834         }
 835         defer t.dropConnection(c)
 836         go c.writer(time.Minute)
 837         cl.sendInitialMessages(c, t)
 838         err := c.mainReadLoop()
 839         if err != nil && cl.config.Debug {
 840                 log.Printf("error during connection main read loop: %s", err)
 841         }
 842 }
 843
 844 // See the order given in Transmission's tr_peerMsgsNew.
 845 func (cl *Client) sendInitialMessages(conn *connection, torrent *Torrent) {
 846         if conn.PeerExtensionBytes.SupportsExtended() && cl.extensionBytes.SupportsExtended() {
 847                 conn.Post(pp.Message{
 848                         Type:       pp.Extended,
 849                         ExtendedID: pp.HandshakeExtendedID,
 850                         ExtendedPayload: func() []byte {
 851                                 msg := pp.ExtendedHandshakeMessage{
 852                                         M: map[pp.ExtensionName]pp.ExtensionNumber{
 853                                                 pp.ExtensionNameMetadata: metadataExtendedId,
 854                                         },
 855                                         V:            cl.config.ExtendedHandshakeClientVersion,
 856                                         Reqq:         64, // TODO: Really?
 857                                         YourIp:       pp.CompactIp(conn.remoteAddr.IP),
 858                                         Encryption:   !cl.config.DisableEncryption,
 859                                         Port:         cl.incomingPeerPort(),
 860                                         MetadataSize: torrent.metadataSize(),
 861                                         // TODO: We can figured these out specific to the socket
 862                                         // used.
 863                                         Ipv4: pp.CompactIp(cl.config.PublicIp4.To4()),
 864                                         Ipv6: cl.config.PublicIp6.To16(),
 865                                 }
 866                                 if !cl.config.DisablePEX {
 867                                         msg.M[pp.ExtensionNamePex] = pexExtendedId
 868                                 }
 869                                 return bencode.MustMarshal(msg)
 870                         }(),
 871                 })
 872         }
 873         func() {
 874                 if conn.fastEnabled() {
 875                         if torrent.haveAllPieces() {
 876                                 conn.Post(pp.Message{Type: pp.HaveAll})
 877                                 conn.sentHaves.AddRange(0, bitmap.BitIndex(conn.t.NumPieces()))
 878                                 return
 879                         } else if !torrent.haveAnyPieces() {
 880                                 conn.Post(pp.Message{Type: pp.HaveNone})
 881                                 conn.sentHaves.Clear()
 882                                 return
 883                         }
 884                 }
 885                 conn.PostBitfield()
 886         }()
 887         if conn.PeerExtensionBytes.SupportsDHT() && cl.extensionBytes.SupportsDHT() && cl.haveDhtServer() {
 888                 conn.Post(pp.Message{
 889                         Type: pp.Port,
 890                         Port: cl.dhtPort(),
 891                 })
 892         }
 893 }
 894
 895 func (cl *Client) dhtPort() (ret uint16) {
 896         cl.eachDhtServer(func(s *dht.Server) {
 897                 ret = uint16(missinggo.AddrPort(s.Addr()))
 898         })
 899         return
 900 }
 901
 902 func (cl *Client) haveDhtServer() (ret bool) {
 903         cl.eachDhtServer(func(_ *dht.Server) {
 904                 ret = true
 905         })
 906         return
 907 }
 908
 909 // Process incoming ut_metadata message.
 910 func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *connection) error {
 911         var d map[string]int
 912         err := bencode.Unmarshal(payload, &d)
 913         if _, ok := err.(bencode.ErrUnusedTrailingBytes); ok {
 914         } else if err != nil {
 915                 return fmt.Errorf("error unmarshalling bencode: %s", err)
 916         }
 917         msgType, ok := d["msg_type"]
 918         if !ok {
 919                 return errors.New("missing msg_type field")
 920         }
 921         piece := d["piece"]
 922         switch msgType {
 923         case pp.DataMetadataExtensionMsgType:
 924                 c.allStats(add(1, func(cs *ConnStats) *Count { return &cs.MetadataChunksRead }))
 925                 if !c.requestedMetadataPiece(piece) {
 926                         return fmt.Errorf("got unexpected piece %d", piece)
 927                 }
 928                 c.metadataRequests[piece] = false
 929                 begin := len(payload) - metadataPieceSize(d["total_size"], piece)
 930                 if begin < 0 || begin >= len(payload) {
 931                         return fmt.Errorf("data has bad offset in payload: %d", begin)
 932                 }
 933                 t.saveMetadataPiece(piece, payload[begin:])
 934                 c.lastUsefulChunkReceived = time.Now()
 935                 return t.maybeCompleteMetadata()
 936         case pp.RequestMetadataExtensionMsgType:
 937                 if !t.haveMetadataPiece(piece) {
 938                         c.Post(t.newMetadataExtensionMessage(c, pp.RejectMetadataExtensionMsgType, d["piece"], nil))
 939                         return nil
 940                 }
 941                 start := (1 << 14) * piece
 942                 c.Post(t.newMetadataExtensionMessage(c, pp.DataMetadataExtensionMsgType, piece, t.metadataBytes[start:start+t.metadataPieceSize(piece)]))
 943                 return nil
 944         case pp.RejectMetadataExtensionMsgType:
 945                 return nil
 946         default:
 947                 return errors.New("unknown msg_type value")
 948         }
 949 }
 950
 951 func (cl *Client) badPeerIPPort(ip net.IP, port int) bool {
 952         if port == 0 {
 953                 return true
 954         }
 955         if cl.dopplegangerAddr(net.JoinHostPort(ip.String(), strconv.FormatInt(int64(port), 10))) {
 956                 return true
 957         }
 958         if _, ok := cl.ipBlockRange(ip); ok {
 959                 return true
 960         }
 961         if _, ok := cl.badPeerIPs[ip.String()]; ok {
 962                 return true
 963         }
 964         return false
 965 }
 966
 967 // Return a Torrent ready for insertion into a Client.
 968 func (cl *Client) newTorrent(ih metainfo.Hash, specStorage storage.ClientImpl) (t *Torrent) {
 969         // use provided storage, if provided
 970         storageClient := cl.defaultStorage
 971         if specStorage != nil {
 972                 storageClient = storage.NewClient(specStorage)
 973         }
 974
 975         t = &Torrent{
 976                 cl:       cl,
 977                 infoHash: ih,
 978                 peers: prioritizedPeers{
 979                         om: btree.New(32),
 980                         getPrio: func(p Peer) peerPriority {
 981                                 return bep40PriorityIgnoreError(cl.publicAddr(p.IP), p.addr())
 982                         },
 983                 },
 984                 conns: make(map[*connection]struct{}, 2*cl.config.EstablishedConnsPerTorrent),
 985
 986                 halfOpen:          make(map[string]Peer),
 987                 pieceStateChanges: pubsub.NewPubSub(),
 988
 989                 storageOpener:       storageClient,
 990                 maxEstablishedConns: cl.config.EstablishedConnsPerTorrent,
 991
 992                 networkingEnabled: true,
 993                 requestStrategy:   2,
 994                 metadataChanged: sync.Cond{
 995                         L: cl.locker(),
 996                 },
 997                 duplicateRequestTimeout: 1 * time.Second,
 998         }
 999         t.logger = cl.logger.Clone().AddValue(t)
1000         t.setChunkSize(defaultChunkSize)
1001         return
1002 }
1003
1004 // A file-like handle to some torrent data resource.
1005 type Handle interface {
1006         io.Reader
1007         io.Seeker
1008         io.Closer
1009         io.ReaderAt
1010 }
1011
1012 func (cl *Client) AddTorrentInfoHash(infoHash metainfo.Hash) (t *Torrent, new bool) {
1013         return cl.AddTorrentInfoHashWithStorage(infoHash, nil)
1014 }
1015
1016 // Adds a torrent by InfoHash with a custom Storage implementation.
1017 // If the torrent already exists then this Storage is ignored and the
1018 // existing torrent returned with `new` set to `false`
1019 func (cl *Client) AddTorrentInfoHashWithStorage(infoHash metainfo.Hash, specStorage storage.ClientImpl) (t *Torrent, new bool) {
1020         cl.lock()
1021         defer cl.unlock()
1022         t, ok := cl.torrents[infoHash]
1023         if ok {
1024                 return
1025         }
1026         new = true
1027
1028         t = cl.newTorrent(infoHash, specStorage)
1029         cl.eachDhtServer(func(s *dht.Server) {
1030                 go t.dhtAnnouncer(s)
1031         })
1032         cl.torrents[infoHash] = t
1033         cl.clearAcceptLimits()
1034         t.updateWantPeersEvent()
1035         // Tickle Client.waitAccept, new torrent may want conns.
1036         cl.event.Broadcast()
1037         return
1038 }
1039
1040 // Add or merge a torrent spec. If the torrent is already present, the
1041 // trackers will be merged with the existing ones. If the Info isn't yet
1042 // known, it will be set. The display name is replaced if the new spec
1043 // provides one. Returns new if the torrent wasn't already in the client.
1044 // Note that any `Storage` defined on the spec will be ignored if the
1045 // torrent is already present (i.e. `new` return value is `true`)
1046 func (cl *Client) AddTorrentSpec(spec *TorrentSpec) (t *Torrent, new bool, err error) {
1047         t, new = cl.AddTorrentInfoHashWithStorage(spec.InfoHash, spec.Storage)
1048         if spec.DisplayName != "" {
1049                 t.SetDisplayName(spec.DisplayName)
1050         }
1051         if spec.InfoBytes != nil {
1052                 err = t.SetInfoBytes(spec.InfoBytes)
1053                 if err != nil {
1054                         return
1055                 }
1056         }
1057         cl.lock()
1058         defer cl.unlock()
1059         if spec.ChunkSize != 0 {
1060                 t.setChunkSize(pp.Integer(spec.ChunkSize))
1061         }
1062         t.addTrackers(spec.Trackers)
1063         t.maybeNewConns()
1064         return
1065 }
1066
1067 func (cl *Client) dropTorrent(infoHash metainfo.Hash) (err error) {
1068         t, ok := cl.torrents[infoHash]
1069         if !ok {
1070                 err = fmt.Errorf("no such torrent")
1071                 return
1072         }
1073         err = t.close()
1074         if err != nil {
1075                 panic(err)
1076         }
1077         delete(cl.torrents, infoHash)
1078         return
1079 }
1080
1081 func (cl *Client) allTorrentsCompleted() bool {
1082         for _, t := range cl.torrents {
1083                 if !t.haveInfo() {
1084                         return false
1085                 }
1086                 if !t.haveAllPieces() {
1087                         return false
1088                 }
1089         }
1090         return true
1091 }
1092
1093 // Returns true when all torrents are completely downloaded and false if the
1094 // client is stopped before that.
1095 func (cl *Client) WaitAll() bool {
1096         cl.lock()
1097         defer cl.unlock()
1098         for !cl.allTorrentsCompleted() {
1099                 if cl.closed.IsSet() {
1100                         return false
1101                 }
1102                 cl.event.Wait()
1103         }
1104         return true
1105 }
1106
1107 // Returns handles to all the torrents loaded in the Client.
1108 func (cl *Client) Torrents() []*Torrent {
1109         cl.lock()
1110         defer cl.unlock()
1111         return cl.torrentsAsSlice()
1112 }
1113
1114 func (cl *Client) torrentsAsSlice() (ret []*Torrent) {
1115         for _, t := range cl.torrents {
1116                 ret = append(ret, t)
1117         }
1118         return
1119 }
1120
1121 func (cl *Client) AddMagnet(uri string) (T *Torrent, err error) {
1122         spec, err := TorrentSpecFromMagnetURI(uri)
1123         if err != nil {
1124                 return
1125         }
1126         T, _, err = cl.AddTorrentSpec(spec)
1127         return
1128 }
1129
1130 func (cl *Client) AddTorrent(mi *metainfo.MetaInfo) (T *Torrent, err error) {
1131         T, _, err = cl.AddTorrentSpec(TorrentSpecFromMetaInfo(mi))
1132         var ss []string
1133         slices.MakeInto(&ss, mi.Nodes)
1134         cl.AddDHTNodes(ss)
1135         return
1136 }
1137
1138 func (cl *Client) AddTorrentFromFile(filename string) (T *Torrent, err error) {
1139         mi, err := metainfo.LoadFromFile(filename)
1140         if err != nil {
1141                 return
1142         }
1143         return cl.AddTorrent(mi)
1144 }
1145
1146 func (cl *Client) DhtServers() []*dht.Server {
1147         return cl.dhtServers
1148 }
1149
1150 func (cl *Client) AddDHTNodes(nodes []string) {
1151         for _, n := range nodes {
1152                 hmp := missinggo.SplitHostMaybePort(n)
1153                 ip := net.ParseIP(hmp.Host)
1154                 if ip == nil {
1155                         log.Printf("won't add DHT node with bad IP: %q", hmp.Host)
1156                         continue
1157                 }
1158                 ni := krpc.NodeInfo{
1159                         Addr: krpc.NodeAddr{
1160                                 IP:   ip,
1161                                 Port: hmp.Port,
1162                         },
1163                 }
1164                 cl.eachDhtServer(func(s *dht.Server) {
1165                         s.AddNode(ni)
1166                 })
1167         }
1168 }
1169
1170 func (cl *Client) banPeerIP(ip net.IP) {
1171         if cl.badPeerIPs == nil {
1172                 cl.badPeerIPs = make(map[string]struct{})
1173         }
1174         cl.badPeerIPs[ip.String()] = struct{}{}
1175 }
1176
1177 func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr IpPort, network string) (c *connection) {
1178         c = &connection{
1179                 conn:            nc,
1180                 outgoing:        outgoing,
1181                 Choked:          true,
1182                 PeerChoked:      true,
1183                 PeerMaxRequests: 250,
1184                 writeBuffer:     new(bytes.Buffer),
1185                 remoteAddr:      remoteAddr,
1186                 network:         network,
1187         }
1188         c.writerCond.L = cl.locker()
1189         c.setRW(connStatsReadWriter{nc, c})
1190         c.r = &rateLimitedReader{
1191                 l: cl.config.DownloadRateLimiter,
1192                 r: c.r,
1193         }
1194         return
1195 }
1196
1197 func (cl *Client) onDHTAnnouncePeer(ih metainfo.Hash, p dht.Peer) {
1198         cl.lock()
1199         defer cl.unlock()
1200         t := cl.torrent(ih)
1201         if t == nil {
1202                 return
1203         }
1204         t.addPeers([]Peer{{
1205                 IP:     p.IP,
1206                 Port:   p.Port,
1207                 Source: peerSourceDHTAnnouncePeer,
1208         }})
1209 }
1210
1211 func firstNotNil(ips ...net.IP) net.IP {
1212         for _, ip := range ips {
1213                 if ip != nil {
1214                         return ip
1215                 }
1216         }
1217         return nil
1218 }
1219
1220 func (cl *Client) eachListener(f func(socket) bool) {
1221         for _, s := range cl.conns {
1222                 if !f(s) {
1223                         break
1224                 }
1225         }
1226 }
1227
1228 func (cl *Client) findListener(f func(net.Listener) bool) (ret net.Listener) {
1229         cl.eachListener(func(l socket) bool {
1230                 ret = l
1231                 return !f(l)
1232         })
1233         return
1234 }
1235
1236 func (cl *Client) publicIp(peer net.IP) net.IP {
1237         // TODO: Use BEP 10 to determine how peers are seeing us.
1238         if peer.To4() != nil {
1239                 return firstNotNil(
1240                         cl.config.PublicIp4,
1241                         cl.findListenerIp(func(ip net.IP) bool { return ip.To4() != nil }),
1242                 )
1243         } else {
1244                 return firstNotNil(
1245                         cl.config.PublicIp6,
1246                         cl.findListenerIp(func(ip net.IP) bool { return ip.To4() == nil }),
1247                 )
1248         }
1249 }
1250
1251 func (cl *Client) findListenerIp(f func(net.IP) bool) net.IP {
1252         return missinggo.AddrIP(cl.findListener(func(l net.Listener) bool {
1253                 return f(missinggo.AddrIP(l.Addr()))
1254         }).Addr())
1255 }
1256
1257 // Our IP as a peer should see it.
1258 func (cl *Client) publicAddr(peer net.IP) IpPort {
1259         return IpPort{cl.publicIp(peer), uint16(cl.incomingPeerPort())}
1260 }
1261
1262 func (cl *Client) ListenAddrs() (ret []net.Addr) {
1263         cl.lock()
1264         defer cl.unlock()
1265         cl.eachListener(func(l socket) bool {
1266                 ret = append(ret, l.Addr())
1267                 return true
1268         })
1269         return
1270 }
1271
1272 func (cl *Client) onBadAccept(addr IpPort) {
1273         ip := maskIpForAcceptLimiting(addr.IP)
1274         if cl.acceptLimiter == nil {
1275                 cl.acceptLimiter = make(map[ipStr]int)
1276         }
1277         cl.acceptLimiter[ipStr(ip.String())]++
1278 }
1279
1280 func maskIpForAcceptLimiting(ip net.IP) net.IP {
1281         if ip4 := ip.To4(); ip4 != nil {
1282                 return ip4.Mask(net.CIDRMask(24, 32))
1283         }
1284         return ip
1285 }
1286
1287 func (cl *Client) clearAcceptLimits() {
1288         cl.acceptLimiter = nil
1289 }
1290
1291 func (cl *Client) acceptLimitClearer() {
1292         for {
1293                 select {
1294                 case <-cl.closed.LockedChan(cl.locker()):
1295                         return
1296                 case <-time.After(15 * time.Minute):
1297                         cl.lock()
1298                         cl.clearAcceptLimits()
1299                         cl.unlock()
1300                 }
1301         }
1302 }
1303
1304 func (cl *Client) rateLimitAccept(ip net.IP) bool {
1305         if cl.config.DisableAcceptRateLimiting {
1306                 return false
1307         }
1308         return cl.acceptLimiter[ipStr(maskIpForAcceptLimiting(ip).String())] > 0
1309 }
1310
1311 func (cl *Client) rLock() {
1312         cl._mu.RLock()
1313 }
1314
1315 func (cl *Client) rUnlock() {
1316         cl._mu.RUnlock()
1317 }
1318
1319 func (cl *Client) lock() {
1320         cl._mu.Lock()
1321 }
1322
1323 func (cl *Client) unlock() {
1324         cl._mu.Unlock()
1325 }
1326
1327 func (cl *Client) locker() sync.Locker {
1328         return clientLocker{cl}
1329 }
1330
1331 type clientLocker struct {
1332         *Client
1333 }
1334
1335 func (cl clientLocker) Lock() {
1336         cl.lock()
1337 }
1338
1339 func (cl clientLocker) Unlock() {
1340         cl.unlock()
1341 }