java.lang.Object
org.klomp.snark.PeerConnectionOut
All Implemented Interfaces:
Runnable
class PeerConnectionOutextends Object implements Runnable
Manages outgoing BitTorrent protocol messages to a peer.
This class maintains a queue of outgoing messages and handles bandwidth limiting, message prioritization, and reliable delivery to the peer.

  • Field Summary

    Fields
    Modifier and Type
    Field
    Description
    private static final AtomicLong
    __id
    Sequence number of the created connections.
    private final long
    _id
    This connection's position in that sequence.
    private static final AtomicLong
    _loadCount
    Prefetches measured, cheap instrumentation for the cold read stall share.
    private static final AtomicLong
    _loadTotalMs
    Total prefetch time measured, in ms.
    private final Log
    _log
    Logger for the outbound peer connection.
    private static long
    _maxLoadMs
    Slowest prefetch measured, in ms.
    private static final ThreadPoolExecutor
    _prefetchExec
    Pool running the piece prefetches.
    private final AtomicInteger
    _prefetchInFlight
    Uncompleted prefetch tasks for this connection; bounds memory to depth x block size.
    private static final AtomicInteger
    _prefetchThreadId
    Sequence number of the prefetch threads.
    private final DataOutputStream
    dout
    Stream the messages are written to.
    (package private) long
    lastSent
    Last send time.
    (package private) static int
    MAX_PREFETCH_INFLIGHT
    Max uncompleted prefetches per connection; bounds pinned buffers to depth x 16KB.
    private static final int
    MAX_PREFETCH_THREADS
    Pool running the piece prefetches, one thread per concurrent burst up to the thread limit, with idle threads dying so the cost at rest is zero.
    private static final int
    MAX_PRIORITY_SCAN
    Max consecutive PIECE messages the priority scan walks before giving up.
    private static final int
    MAX_QUEUE_SIZE
    The messages waiting to be written, a bounded queue so a slow network cannot grow the memory without limit.
    private final Peer
    peer
    The peer this connection serves.
    private static final long
    PREFETCH_THREAD_IDLE_MS
    Idle prefetch threads exit after this long.
    private boolean
    quit
    True once the connection should close.
    private static final int
    REQ_TIMEOUT
    Retransmit requests not received in 7m.
    private static final int
    SEND_TIMEOUT
    Remove messages not sent in 3m.
    private final BlockingQueue<Message>
    sendQueue
    The messages waiting to be written.
    private static final long
    SLOW_LOAD_WARN_MS
    Loads slower than this are logged; sustained counts confirm cold-read stalls.
    private Thread
    thread
    The thread draining the send queue.

  • Constructor Summary

    Constructors
    Constructor
    Description
    PeerConnectionOut(Peer peer, DataOutputStream dout)
    Creates a new outgoing connection handler.

  • Method Summary

    Modifier and Type
    Method
    Description
    private void
    addMessage(Message m)
    Adds a message to the sendQueue and notifies the method waiting on the sendQueue to change.
    (package private) void
    cancelRequest(int piece, int begin, int length)
    Called by the PeerState when the other side doesn't want this request to be handled anymore.
    (package private) void
    cancelRequestMessages()
    Remove all Request messages from the queue.
    (package private) void
    cancelRequestMessages(int piece)
    Remove Request messages for a piece from the queue.
    void
    disconnect()
    Disconnect and close the output stream.
    (package private) boolean
    hasPendingPiece()
    Whether a PIECE message is still queued, i.e.
    (package private) Message
    headPiece()
    The first queued PIECE message in send order.
    private void
    kickPrefetch(Message m)
    Submit an off-thread load for a freshly queued PIECE so its data is ready before the sender thread reaches it, decoupling disk latency from the write path.
    (package private) int
    prefetchInFlight()
    Current number of uncompleted prefetch loads for this connection.
    (package private) int
    queuedBytes()
    Returns the total bytes of piece messages currently queued.
    (package private) static void
    recordLoad(long ms)
    Record a deferred-load duration.
    private boolean
    removeMessage(int type)
    Removes a particular message type from the queue.
    (package private) void
    retransmitRequests(List<Request> requests)
    Retransmit requests that haven't been received within REQ_TIMEOUT.
    void
    run()
    Pull messages from the send queue and deliver to the peer.
    private void
    runPrefetch(Message m)
    Prefetch task: claim exclusive load rights, skip if purged meanwhile, read outside all locks, then publish.
    (package private) void
    sendAlive()
    Queue a keepalive message if the queue is empty.
    (package private) void
    sendAllowedFast(int piece)
    Queue an allowed fast message (BEP 6).
    (package private) void
    sendBitfield(BitField bitfield)
    Queue a bitfield, have_all, or have_none message as appropriate.
    (package private) void
    sendCancel(Request req)
    Send a cancel message and remove any matching request from the queue.
    (package private) void
    sendChoke(boolean choke)
    Queue a choke or unchoke message, cancelling any pending inverse.
    (package private) void
    sendExtension(int id, byte[] bytes)
    Queue an extension message.
    (package private) void
    sendHave(int piece)
    Queue a have message for the given piece.
    private void
    sendHaveAll()
    Queue a have-all message.
    private void
    sendHaveNone()
    Queue a have-none message.
    (package private) void
    sendInterest(boolean interest)
    Queue an interested or not-interested message, cancelling any pending inverse.
    (package private) void
    sendPiece(int piece, int begin, int length, DataLoader loader)
    Queue a piece message with a callback to load the data from disk when required.
    (package private) void
    sendPort(int port)
    Queue a port message.
    (package private) void
    sendReject(int piece, int begin, int length)
    Queue a reject message.
    (package private) void
    sendRequest(Request req)
    Queue a request message, checking for duplicates.
    (package private) void
    sendRequests(List<Request> requests)
    Send all requests in the list.
    void
    startup()
    Start the sender thread.

    Methods inherited from class Object

    clone, equals, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

  • Field Details

    • _log

      private final Log _log
      Logger for the outbound peer connection.

    • peer

      private final Peer peer
      The peer this connection serves.

    • dout

      private final DataOutputStream dout
      Stream the messages are written to.

    • thread

      private Thread thread
      The thread draining the send queue.

    • quit

      private boolean quit
      True once the connection should close.

    • MAX_QUEUE_SIZE

      private static final int MAX_QUEUE_SIZE
      The messages waiting to be written, a bounded queue so a slow network cannot grow the memory without limit.
      See Also:

    • MAX_PRIORITY_SCAN

      private static final int MAX_PRIORITY_SCAN
      Max consecutive PIECE messages the priority scan walks before giving up.
      See Also:

    • sendQueue

      private final BlockingQueue<Message> sendQueue
      The messages waiting to be written.

    • MAX_PREFETCH_THREADS

      private static final int MAX_PREFETCH_THREADS
      Pool running the piece prefetches, one thread per concurrent burst up to the thread limit, with idle threads dying so the cost at rest is zero. When all threads are busy or the per connection depth cap is hit, new pieces stay lazy and fall back to the inline load.
      See Also:

    • PREFETCH_THREAD_IDLE_MS

      private static final long PREFETCH_THREAD_IDLE_MS
      Idle prefetch threads exit after this long.
      See Also:

    • MAX_PREFETCH_INFLIGHT

      static volatile int MAX_PREFETCH_INFLIGHT
      Max uncompleted prefetches per connection; bounds pinned buffers to depth x 16KB.

    • SLOW_LOAD_WARN_MS

      private static final long SLOW_LOAD_WARN_MS
      Loads slower than this are logged; sustained counts confirm cold-read stalls.
      See Also:

    • _prefetchThreadId

      private static final AtomicInteger _prefetchThreadId
      Sequence number of the prefetch threads.

    • _prefetchExec

      private static final ThreadPoolExecutor _prefetchExec
      Pool running the piece prefetches.

    • _loadCount

      private static final AtomicLong _loadCount
      Prefetches measured, cheap instrumentation for the cold read stall share.

    • _loadTotalMs

      private static final AtomicLong _loadTotalMs
      Total prefetch time measured, in ms.

    • _maxLoadMs

      private static volatile long _maxLoadMs
      Slowest prefetch measured, in ms.

    • __id

      private static final AtomicLong __id
      Sequence number of the created connections.

    • _id

      private final long _id
      This connection's position in that sequence.

    • lastSent

      long lastSent
      Last send time.

    • _prefetchInFlight

      private final AtomicInteger _prefetchInFlight
      Uncompleted prefetch tasks for this connection; bounds memory to depth x block size.

    • SEND_TIMEOUT

      private static final int SEND_TIMEOUT
      Remove messages not sent in 3m.
      See Also:

    • REQ_TIMEOUT

      private static final int REQ_TIMEOUT
      Retransmit requests not received in 7m.
      See Also:

  • Constructor Details

    • PeerConnectionOut

      public PeerConnectionOut(Peer peer, DataOutputStream dout)
      Creates a new outgoing connection handler.
      Parameters:
      peer - the remote peer this connection sends to, whose state and preferences drive the message queue
      dout - the output stream to send messages on

  • Method Details

    • recordLoad

      static void recordLoad(long ms)
      Record a deferred-load duration. Called from both the prefetch task and the inline fallback in Message.sendMessage(DataOutputStream).
      Parameters:
      ms - load duration in milliseconds

    • startup

      public void startup()
      Start the sender thread.

    • hasPendingPiece

      boolean hasPendingPiece()
      Whether a PIECE message is still queued, i.e. we are actively uploading to the peer.
      Returns:
      true if any PIECE message is pending

    • headPiece

      Message headPiece()
      The first queued PIECE message in send order.
      Returns:
      the oldest pending PIECE message, or null if none queued

    • prefetchInFlight

      int prefetchInFlight()
      Current number of uncompleted prefetch loads for this connection.
      Returns:
      in-flight prefetch count, capped by MAX_PREFETCH_INFLIGHT

    • run

      public void run()
      Pull messages from the send queue and deliver to the peer.
      Specified by:
      run in interface Runnable

    • disconnect

      public void disconnect()
      Disconnect and close the output stream.

    • addMessage

      private void addMessage(Message m)
      Adds a message to the sendQueue and notifies the method waiting on the sendQueue to change. Control messages are never silently dropped: if the queue is full, a queued piece message is dropped to make room for it. Piece messages are dropped instead, the requester will retry.
      Parameters:
      m - the message to queue for sending to the peer

    • removeMessage

      private boolean removeMessage(int type)
      Removes a particular message type from the queue.
      Parameters:
      type - the Message type to remove.
      Returns:
      true when a message of the given type was removed, false otherwise.

    • sendAlive

      void sendAlive()
      Queue a keepalive message if the queue is empty.

    • sendChoke

      void sendChoke(boolean choke)
      Queue a choke or unchoke message, cancelling any pending inverse.
      Parameters:
      choke - true to choke, false to unchoke

    • sendInterest

      void sendInterest(boolean interest)
      Queue an interested or not-interested message, cancelling any pending inverse.
      Parameters:
      interest - true for interested, false for not-interested

    • sendHave

      void sendHave(int piece)
      Queue a have message for the given piece.
      Parameters:
      piece - the piece number

    • sendBitfield

      void sendBitfield(BitField bitfield)
      Queue a bitfield, have_all, or have_none message as appropriate.
      Parameters:
      bitfield - the bitfield to send

    • retransmitRequests

      void retransmitRequests(List<Request> requests)
      Retransmit requests that haven't been received within REQ_TIMEOUT. Only call if !supportsFast()
      Parameters:
      requests - the list of outstanding requests

    • sendRequests

      void sendRequests(List<Request> requests)
      Send all requests in the list.
      Parameters:
      requests - the list of requests to send

    • sendRequest

      void sendRequest(Request req)
      Queue a request message, checking for duplicates.
      Parameters:
      req - the request to send

    • queuedBytes

      int queuedBytes()
      Returns the total bytes of piece messages currently queued. Used by PeerState to limit pipelined requests.
      Returns:
      total queued piece bytes

    • sendPiece

      void sendPiece(int piece, int begin, int length, DataLoader loader)
      Queue a piece message with a callback to load the data from disk when required.
      Parameters:
      piece - the zero-based index of the piece requested
      begin - the offset within that piece where the wanted bytes start
      length - the number of bytes wanted from that offset
      loader - the callback that reads the wanted bytes from storage when the message is actually sent

    • kickPrefetch

      private void kickPrefetch(Message m)
      Submit an off-thread load for a freshly queued PIECE so its data is ready before the sender thread reaches it, decoupling disk latency from the write path. Best-effort: when the depth cap or thread pool is saturated the message stays lazy and loads inline as before. The cap check is racy in principle, but enqueues are single-threaded per connection (PeerState's reader), so the bound holds exactly.
      Parameters:
      m - the queued PIECE message

    • runPrefetch

      private void runPrefetch(Message m)
      Prefetch task: claim exclusive load rights, skip if purged meanwhile, read outside all locks, then publish. Storage errors already surface as a null result (with REJECT) via PeerCoordinator; an unexpected Throwable here drops only this block instead of killing the connection as the old inline path would have.
      Parameters:
      m - the PIECE message to load

    • sendCancel

      void sendCancel(Request req)
      Send a cancel message and remove any matching request from the queue.
      Parameters:
      req - the request to cancel

    • cancelRequestMessages

      void cancelRequestMessages()
      Remove all Request messages from the queue. Does not send a cancel message.

    • cancelRequestMessages

      void cancelRequestMessages(int piece)
      Remove Request messages for a piece from the queue. Does not send a cancel message.
      Parameters:
      piece - the piece index

    • cancelRequest

      void cancelRequest(int piece, int begin, int length)
      Called by the PeerState when the other side doesn't want this request to be handled anymore. Removes any pending Piece Message from our send queue. Does not send a cancel message.
      Parameters:
      piece - the piece number
      begin - the offset within the piece
      length - the length of the request

    • sendExtension

      void sendExtension(int id, byte[] bytes)
      Queue an extension message.
      Parameters:
      id - the extension message id from the BEP 10 handshake
      bytes - the extension payload, queued by reference and not copied

    • sendPort

      void sendPort(int port)
      Queue a port message.
      Parameters:
      port - the port the peer is listening on for incoming connections

    • sendHaveAll

      private void sendHaveAll()
      Queue a have-all message.

    • sendHaveNone

      private void sendHaveNone()
      Queue a have-none message.

    • sendReject

      void sendReject(int piece, int begin, int length)
      Queue a reject message.
      Parameters:
      piece - the zero-based index of the rejected piece
      begin - the offset within that piece that was rejected
      length - the number of bytes rejected from that offset

    • sendAllowedFast

      void sendAllowedFast(int piece)
      Queue an allowed fast message (BEP 6).
      Parameters:
      piece - the zero-based index of the piece the peer may request without asking