mirror
/
transmission
mirror of https://github.com/transmission/transmission
1
0
Fork 0
Code Issues Releases Wiki Activity

refactor: tr_peerIo (#4372)

This commit is contained in:
Charles Kerr 2022-12-16 01:23:12 -06:00 committed by GitHub
parent 078dc8bd08
commit e6d75a4b77
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
14 changed files with 781 additions and 871 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
libtransmission/announcer.cc
View File

@ -245,7 +245,7 @@ private:
{
for (auto& stop : stops_)
{
announce(stop, [](tr_announce_response const&) {});
announce(stop, [](tr_announce_response const& /*response*/) {});
}
stops_.clear();

10
libtransmission/bandwidth.cc
View File

@ -150,10 +150,10 @@ void tr_bandwidth::allocateBandwidth(
bandwidth.bytes_left_ = next_pulse_speed * period_msec / 1000U;
}
/* add this bandwidth's peer, if any, to the peer pool */
// add this bandwidth's peer, if any, to the peer pool
if (auto shared = this->peer_.lock(); shared)
{
shared->priority = priority;
shared->set_priority(priority);
peer_pool.push_back(std::move(shared));
}
@ -213,9 +213,9 @@ void tr_bandwidth::allocate(tr_direction dir, unsigned int period_msec)
for (auto const& io : refs)
{
io->flushOutgoingProtocolMsgs();
io->flush_outgoing_protocol_msgs();
switch (io->priority)
switch (io->priority())
{
case TR_PRI_HIGH:
high.push_back(io.get());
@ -244,7 +244,7 @@ void tr_bandwidth::allocate(tr_direction dir, unsigned int period_msec)
* or (2) the next tr_bandwidth::allocate () call, when we start over again. */
for (auto const& io : refs)
{
io->setEnabled(dir, io->hasBandwidthLeft(dir));
io->set_enabled(dir, io->has_bandwidth_left(dir));
}
}

169
libtransmission/handshake.cc
View File

@ -32,7 +32,7 @@ using DH = tr_message_stream_encryption::DH;
bool tr_handshake::build_handshake_message(tr_peerIo* io, uint8_t* buf) const
{
auto const& info_hash = io->torrentHash();
auto const& info_hash = io->torrent_hash();
TR_ASSERT_MSG(info_hash != tr_sha1_digest_t{}, "build_handshake_message requires an info_hash");
auto const info = mediator_->torrent(info_hash);
@ -62,16 +62,16 @@ bool tr_handshake::build_handshake_message(tr_peerIo* io, uint8_t* buf) const
tr_handshake::ParseResult tr_handshake::parse_handshake(tr_peerIo* peer_io)
{
tr_logAddTraceHand(this, fmt::format("payload: need {}, got {}", HandshakeSize, peer_io->readBufferSize()));
tr_logAddTraceHand(this, fmt::format("payload: need {}, got {}", HandshakeSize, peer_io->read_buffer_size()));
if (peer_io->readBufferSize() < HandshakeSize)
if (peer_io->read_buffer_size() < HandshakeSize)
{
return ParseResult::EncryptionWrong;
}
/* confirm the protocol */
auto name = decltype(HandshakeName){};
peer_io->readBytes(std::data(name), std::size(name));
peer_io->read_bytes(std::data(name), std::size(name));
if (name != HandshakeName)
{
return ParseResult::EncryptionWrong;
@ -80,16 +80,16 @@ tr_handshake::ParseResult tr_handshake::parse_handshake(tr_peerIo* peer_io)
/* read the reserved bytes */
auto flags = tr_bitfield{ HandshakeFlagsBits };
auto reserved = std::array<uint8_t, HandshakeFlagsBytes>{};
peer_io->readBytes(std::data(reserved), std::size(reserved));
peer_io->read_bytes(std::data(reserved), std::size(reserved));
flags.setRaw(std::data(reserved), std::size(reserved));
peer_io->enableDHT(flags.test(DhtFlag));
peer_io->enableLTEP(flags.test(LtepFlag));
peer_io->enableFEXT(flags.test(FextFlag));
peer_io->set_supports_dht(flags.test(DhtFlag));
peer_io->set_supports_ltep(flags.test(LtepFlag));
peer_io->set_supports_fext(flags.test(FextFlag));
// torrent hash
auto info_hash = tr_sha1_digest_t{};
peer_io->readBytes(std::data(info_hash), std::size(info_hash));
if (info_hash == tr_sha1_digest_t{} || info_hash != peer_io->torrentHash())
peer_io->read_bytes(std::data(info_hash), std::size(info_hash));
if (info_hash == tr_sha1_digest_t{} || info_hash != peer_io->torrent_hash())
{
tr_logAddTraceHand(this, "peer returned the wrong hash. wtf?");
return ParseResult::BadTorrent;
@ -97,7 +97,7 @@ tr_handshake::ParseResult tr_handshake::parse_handshake(tr_peerIo* peer_io)
// peer_id
auto peer_id = tr_peer_id_t{};
peer_io->readBytes(std::data(peer_id), std::size(peer_id));
peer_io->read_bytes(std::data(peer_id), std::size(peer_id));
set_peer_id(peer_id);
/* peer id */
@ -161,14 +161,14 @@ void tr_handshake::send_ya(tr_peerIo* io)
ReadState tr_handshake::read_yb(tr_peerIo* peer_io)
{
if (peer_io->readBufferSize() < std::size(HandshakeName))
if (peer_io->read_buffer_size() < std::size(HandshakeName))
{
return READ_LATER;
}
bool const is_encrypted = !peer_io->readBufferStartsWith(HandshakeName);
bool const is_encrypted = !peer_io->read_buffer_starts_with(HandshakeName);
auto peer_public_key = DH::key_bigend_t{};
if (is_encrypted && (peer_io->readBufferSize() < std::size(peer_public_key)))
if (is_encrypted && (peer_io->read_buffer_size() < std::size(peer_public_key)))
{
return READ_LATER;
}
@ -184,7 +184,7 @@ ReadState tr_handshake::read_yb(tr_peerIo* peer_io)
set_have_read_anything_from_peer(true);
// get the peer's public key
peer_io->readBytes(std::data(peer_public_key), std::size(peer_public_key));
peer_io->read_bytes(std::data(peer_public_key), std::size(peer_public_key));
dh_.setPeerPublicKey(peer_public_key);
/* now send these: HASH('req1', S), HASH('req2', SKEY) xor HASH('req3', S),
@ -194,7 +194,7 @@ ReadState tr_handshake::read_yb(tr_peerIo* peer_io)
/* HASH('req1', S) */
outbuf.add(tr_sha1::digest("req1"sv, dh_.secret()));
auto const& info_hash = peer_io->torrentHash();
auto const& info_hash = peer_io->torrent_hash();
TR_ASSERT_MSG(info_hash != tr_sha1_digest_t{}, "readYb requires an info_hash");
/* HASH('req2', SKEY) xor HASH('req3', S) */
@ -214,7 +214,7 @@ ReadState tr_handshake::read_yb(tr_peerIo* peer_io)
* PadC is reserved for future extensions to the handshake...
* standard practice at this time is for it to be zero-length */
peer_io->write(outbuf, false);
peer_io->encryptInit(peer_io->isIncoming(), dh_, info_hash);
peer_io->encrypt_init(peer_io->is_incoming(), dh_, info_hash);
outbuf.add(VC);
outbuf.addUint32(crypto_provide());
outbuf.addUint16(0);
@ -241,7 +241,7 @@ ReadState tr_handshake::read_yb(tr_peerIo* peer_io)
// A will be able to resynchronize on ENCRYPT(VC)"
ReadState tr_handshake::read_vc(tr_peerIo* peer_io)
{
auto const info_hash = peer_io->torrentHash();
auto const info_hash = peer_io->torrent_hash();
TR_ASSERT_MSG(info_hash != tr_sha1_digest_t{}, "readVC requires an info_hash");
// find the end of PadB by looking for `ENCRYPT(VC)`
@ -252,24 +252,24 @@ ReadState tr_handshake::read_vc(tr_peerIo* peer_io)
for (size_t i = 0; i < PadbMaxlen; ++i)
{
if (peer_io->readBufferSize() < std::size(needle))
if (peer_io->read_buffer_size() < std::size(needle))
{
tr_logAddTraceHand(this, "not enough bytes... returning read_more");
return READ_LATER;
}
if (peer_io->readBufferStartsWith(needle))
if (peer_io->read_buffer_starts_with(needle))
{
tr_logAddTraceHand(this, "got it!");
// We already know it's a match; now we just need to
// consume it from the read buffer.
peer_io->decryptInit(peer_io->isIncoming(), dh_, info_hash);
peer_io->readBytes(std::data(needle), std::size(needle));
peer_io->decrypt_init(peer_io->is_incoming(), dh_, info_hash);
peer_io->read_bytes(std::data(needle), std::size(needle));
set_state(tr_handshake::State::AwaitingCryptoSelect);
return READ_NOW;
}
peer_io->readBufferDrain(1);
peer_io->read_buffer_drain(1);
}
tr_logAddTraceHand(this, "couldn't find ENCRYPT(VC)");
@ -278,13 +278,13 @@ ReadState tr_handshake::read_vc(tr_peerIo* peer_io)
ReadState tr_handshake::read_crypto_select(tr_peerIo* peer_io)
{
if (static size_t constexpr NeedLen = sizeof(uint32_t) + sizeof(uint16_t); peer_io->readBufferSize() < NeedLen)
if (static size_t constexpr NeedLen = sizeof(uint32_t) + sizeof(uint16_t); peer_io->read_buffer_size() < NeedLen)
{
return READ_LATER;
}
auto crypto_select = uint32_t{};
peer_io->readUint32(&crypto_select);
peer_io->read_uint32(&crypto_select);
crypto_select_ = crypto_select;
tr_logAddTraceHand(this, fmt::format("crypto select is {}", crypto_select));
@ -295,7 +295,7 @@ ReadState tr_handshake::read_crypto_select(tr_peerIo* peer_io)
}
uint16_t pad_d_len = 0;
peer_io->readUint16(&pad_d_len);
peer_io->read_uint16(&pad_d_len);
tr_logAddTraceHand(this, fmt::format("pad_d_len is {}", pad_d_len));
if (pad_d_len > 512)
@ -313,13 +313,13 @@ ReadState tr_handshake::read_crypto_select(tr_peerIo* peer_io)
ReadState tr_handshake::read_pad_d(tr_peerIo* peer_io)
{
size_t const needlen = pad_d_len_;
tr_logAddTraceHand(this, fmt::format("pad d: need {}, got {}", needlen, peer_io->readBufferSize()));
if (peer_io->readBufferSize() < needlen)
tr_logAddTraceHand(this, fmt::format("pad d: need {}, got {}", needlen, peer_io->read_buffer_size()));
if (peer_io->read_buffer_size() < needlen)
{
return READ_LATER;
}
peer_io->readBufferDrain(needlen);
peer_io->read_buffer_drain(needlen);
set_state(tr_handshake::State::AwaitingHandshake);
return READ_NOW;
@ -334,15 +334,15 @@ ReadState tr_handshake::read_pad_d(tr_peerIo* peer_io)
ReadState tr_handshake::read_handshake(tr_peerIo* peer_io)
{
static auto constexpr Needlen = IncomingHandshakeLen;
tr_logAddTraceHand(this, fmt::format("payload: need {}, got {}", Needlen, peer_io->readBufferSize()));
if (peer_io->readBufferSize() < Needlen)
tr_logAddTraceHand(this, fmt::format("payload: need {}, got {}", Needlen, peer_io->read_buffer_size()));
if (peer_io->read_buffer_size() < Needlen)
{
return READ_LATER;
}
set_have_read_anything_from_peer(true);
if (peer_io->readBufferStartsWith(HandshakeName)) // unencrypted
if (peer_io->read_buffer_starts_with(HandshakeName)) // unencrypted
{
if (encryption_mode_ == TR_ENCRYPTION_REQUIRED)
{
@ -361,7 +361,7 @@ ReadState tr_handshake::read_handshake(tr_peerIo* peer_io)
}
auto name = decltype(HandshakeName){};
peer_io->readBytes(std::data(name), std::size(name));
peer_io->read_bytes(std::data(name), std::size(name));
if (name != HandshakeName)
{
return done(false);
@ -370,15 +370,15 @@ ReadState tr_handshake::read_handshake(tr_peerIo* peer_io)
// reserved bytes / flags
auto reserved = std::array<uint8_t, HandshakeFlagsBytes>{};
auto flags = tr_bitfield{ HandshakeFlagsBits };
peer_io->readBytes(std::data(reserved), std::size(reserved));
peer_io->read_bytes(std::data(reserved), std::size(reserved));
flags.setRaw(std::data(reserved), std::size(reserved));
peer_io->enableDHT(flags.test(DhtFlag));
peer_io->enableLTEP(flags.test(LtepFlag));
peer_io->enableFEXT(flags.test(FextFlag));
peer_io->set_supports_dht(flags.test(DhtFlag));
peer_io->set_supports_ltep(flags.test(LtepFlag));
peer_io->set_supports_fext(flags.test(FextFlag));
/* torrent hash */
auto hash = tr_sha1_digest_t{};
peer_io->readBytes(std::data(hash), std::size(hash));
peer_io->read_bytes(std::data(hash), std::size(hash));
if (is_incoming())
{
@ -388,11 +388,11 @@ ReadState tr_handshake::read_handshake(tr_peerIo* peer_io)
return done(false);
}
peer_io->setTorrentHash(hash);
peer_io->set_torrent_hash(hash);
}
else // outgoing
{
if (peer_io->torrentHash() != hash)
if (peer_io->torrent_hash() != hash)
{
tr_logAddTraceHand(this, "peer returned the wrong hash. wtf?");
return done(false);
@ -412,7 +412,7 @@ ReadState tr_handshake::read_handshake(tr_peerIo* peer_io)
return done(false);
}
peer_io->writeBytes(std::data(msg), std::size(msg), false);
peer_io->write_bytes(std::data(msg), std::size(msg), false);
have_sent_bittorrent_handshake_ = true;
}
@ -424,11 +424,11 @@ ReadState tr_handshake::read_peer_id(tr_peerIo* peer_io)
{
// read the peer_id
auto peer_id = tr_peer_id_t{};
if (peer_io->readBufferSize() < std::size(peer_id))
if (peer_io->read_buffer_size() < std::size(peer_id))
{
return READ_LATER;
}
peer_io->readBytes(std::data(peer_id), std::size(peer_id));
peer_io->read_bytes(std::data(peer_id), std::size(peer_id));
set_peer_id(peer_id);
auto client = std::array<char, 128>{};
@ -436,7 +436,7 @@ ReadState tr_handshake::read_peer_id(tr_peerIo* peer_io)
tr_logAddTraceHand(this, fmt::format("peer-id is '{}' ... isIncoming is {}", std::data(client), is_incoming()));
// if we've somehow connected to ourselves, don't keep the connection
auto const info_hash = peer_io_->torrentHash();
auto const info_hash = peer_io_->torrent_hash();
auto const info = mediator_->torrent(info_hash);
auto const connected_to_self = info && info->client_peer_id == peer_id;
@ -448,15 +448,15 @@ ReadState tr_handshake::read_ya(tr_peerIo* peer_io)
auto peer_public_key = DH::key_bigend_t{};
tr_logAddTraceHand(
this,
fmt::format("in readYa... need {}, have {}", std::size(peer_public_key), peer_io->readBufferSize()));
fmt::format("in readYa... need {}, have {}", std::size(peer_public_key), peer_io->read_buffer_size()));
if (peer_io->readBufferSize() < std::size(peer_public_key))
if (peer_io->read_buffer_size() < std::size(peer_public_key))
{
return READ_LATER;
}
/* read the incoming peer's public key */
peer_io->readBytes(std::data(peer_public_key), std::size(peer_public_key));
peer_io->read_bytes(std::data(peer_public_key), std::size(peer_public_key));
dh_.setPeerPublicKey(peer_public_key);
// send our public key to the peer
@ -474,21 +474,21 @@ ReadState tr_handshake::read_pad_a(tr_peerIo* peer_io)
for (size_t i = 0; i < PadaMaxlen; ++i)
{
if (peer_io->readBufferSize() < std::size(needle))
if (peer_io->read_buffer_size() < std::size(needle))
{
tr_logAddTraceHand(this, "not enough bytes... returning read_more");
return READ_LATER;
}
if (peer_io->readBufferStartsWith(needle))
if (peer_io->read_buffer_starts_with(needle))
{
tr_logAddTraceHand(this, "found it... looking setting to awaiting_crypto_provide");
peer_io->readBufferDrain(std::size(needle));
peer_io->read_buffer_drain(std::size(needle));
set_state(State::AwaitingCryptoProvide);
return READ_NOW;
}
peer_io->readBufferDrain(1U);
peer_io->read_buffer_drain(1U);
}
tr_logAddTraceHand(this, "couldn't find HASH('req', S)");
@ -505,7 +505,7 @@ ReadState tr_handshake::read_crypto_provide(tr_peerIo* peer_io)
size_t const needlen = sizeof(obfuscated_hash) + /* HASH('req2', SKEY) xor HASH('req3', S) */
std::size(VC) + sizeof(crypto_provide) + sizeof(padc_len);
if (peer_io->readBufferSize() < needlen)
if (peer_io->read_buffer_size() < needlen)
{
return READ_LATER;
}
@ -515,7 +515,7 @@ ReadState tr_handshake::read_crypto_provide(tr_peerIo* peer_io)
* by building the latter and xor'ing it with what the peer sent us */
tr_logAddTraceHand(this, "reading obfuscated torrent hash...");
auto req2 = tr_sha1_digest_t{};
peer_io->readBytes(std::data(req2), std::size(req2));
peer_io->read_bytes(std::data(req2), std::size(req2));
auto const req3 = tr_sha1::digest("req3"sv, dh_.secret());
for (size_t i = 0; i < std::size(obfuscated_hash); ++i)
@ -528,7 +528,7 @@ ReadState tr_handshake::read_crypto_provide(tr_peerIo* peer_io)
bool const client_is_seed = info->is_done;
bool const peer_is_seed = mediator_->is_peer_known_seed(info->id, peer_io->address());
tr_logAddTraceHand(this, fmt::format("got INCOMING connection's encrypted handshake for torrent [{}]", info->id));
peer_io->setTorrentHash(info->info_hash);
peer_io->set_torrent_hash(info->info_hash);
if (client_is_seed && peer_is_seed)
{
@ -544,18 +544,18 @@ ReadState tr_handshake::read_crypto_provide(tr_peerIo* peer_io)
/* next part: ENCRYPT(VC, crypto_provide, len(PadC), */
auto const& info_hash = peer_io->torrentHash();
auto const& info_hash = peer_io->torrent_hash();
TR_ASSERT_MSG(info_hash != tr_sha1_digest_t{}, "readCryptoProvide requires an info_hash");
peer_io->decryptInit(peer_io->isIncoming(), dh_, info_hash);
peer_io->decrypt_init(peer_io->is_incoming(), dh_, info_hash);
auto vc_in = vc_t{};
peer_io->readBytes(std::data(vc_in), std::size(vc_in));
peer_io->read_bytes(std::data(vc_in), std::size(vc_in));
peer_io->readUint32(&crypto_provide);
peer_io->read_uint32(&crypto_provide);
crypto_provide_ = crypto_provide;
tr_logAddTraceHand(this, fmt::format("crypto_provide is {}", crypto_provide));
peer_io->readUint16(&padc_len);
peer_io->read_uint16(&padc_len);
tr_logAddTraceHand(this, fmt::format("padc is {}", padc_len));
if (padc_len > PadcMaxlen)
{
@ -570,18 +570,18 @@ ReadState tr_handshake::read_crypto_provide(tr_peerIo* peer_io)
ReadState tr_handshake::read_pad_c(tr_peerIo* peer_io)
{
if (auto const needlen = pad_c_len_ + sizeof(uint16_t); peer_io->readBufferSize() < needlen)
if (auto const needlen = pad_c_len_ + sizeof(uint16_t); peer_io->read_buffer_size() < needlen)
{
return READ_LATER;
}
// read the throwaway padc
auto pad_c = std::array<char, PadcMaxlen>{};
peer_io->readBytes(std::data(pad_c), pad_c_len_);
peer_io->read_bytes(std::data(pad_c), pad_c_len_);
/* read ia_len */
uint16_t ia_len = 0;
peer_io->readUint16(&ia_len);
peer_io->read_uint16(&ia_len);
tr_logAddTraceHand(this, fmt::format("ia_len is {}", ia_len));
ia_len_ = ia_len;
set_state(State::AwaitingIa);
@ -592,9 +592,9 @@ ReadState tr_handshake::read_ia(tr_peerIo* peer_io)
{
size_t const needlen = ia_len_;
tr_logAddTraceHand(this, fmt::format("reading IA... have {}, need {}", peer_io->readBufferSize(), needlen));
tr_logAddTraceHand(this, fmt::format("reading IA... have {}, need {}", peer_io->read_buffer_size(), needlen));
if (peer_io->readBufferSize() < needlen)
if (peer_io->read_buffer_size() < needlen)
{
return READ_LATER;
}
@ -603,9 +603,9 @@ ReadState tr_handshake::read_ia(tr_peerIo* peer_io)
*** B->A: ENCRYPT(VC, crypto_select, len(padD), padD), ENCRYPT2(Payload Stream)
**/
auto const& info_hash = peer_io->torrentHash();
auto const& info_hash = peer_io->torrent_hash();
TR_ASSERT_MSG(info_hash != tr_sha1_digest_t{}, "readIA requires an info_hash");
peer_io->encryptInit(peer_io->isIncoming(), dh_, info_hash);
peer_io->encrypt_init(peer_io->is_incoming(), dh_, info_hash);
auto outbuf = libtransmission::Buffer{};
// send VC
@ -664,8 +664,8 @@ ReadState tr_handshake::read_ia(tr_peerIo* peer_io)
ReadState tr_handshake::read_payload_stream(tr_peerIo* peer_io)
{
static auto constexpr Needlen = HandshakeSize;
tr_logAddTraceHand(this, fmt::format("reading payload stream... have {}, need {}", peer_io->readBufferSize(), Needlen));
if (peer_io->readBufferSize() < Needlen)
tr_logAddTraceHand(this, fmt::format("reading payload stream... have {}, need {}", peer_io->read_buffer_size(), Needlen));
if (peer_io->read_buffer_size() < Needlen)
{
return READ_LATER;
}
@ -690,8 +690,6 @@ ReadState tr_handshake::read_payload_stream(tr_peerIo* peer_io)
ReadState tr_handshake::can_read(tr_peerIo* peer_io, void* vhandshake, size_t* piece)
{
TR_ASSERT(tr_isPeerIo(peer_io));
auto* handshake = static_cast<tr_handshake*>(vhandshake);
bool ready_for_more = true;
@ -771,46 +769,45 @@ ReadState tr_handshake::can_read(tr_peerIo* peer_io, void* vhandshake, size_t* p
}
else if (handshake->is_state(State::AwaitingPadC))
{
ready_for_more = peer_io->readBufferSize() >= handshake->pad_c_len_;
ready_for_more = peer_io->read_buffer_size() >= handshake->pad_c_len_;
}
else if (handshake->is_state(State::AwaitingPadD))
{
ready_for_more = peer_io->readBufferSize() >= handshake->pad_d_len_;
ready_for_more = peer_io->read_buffer_size() >= handshake->pad_d_len_;
}
else if (handshake->is_state(State::AwaitingIa))
{
ready_for_more = peer_io->readBufferSize() >= handshake->ia_len_;
ready_for_more = peer_io->read_buffer_size() >= handshake->ia_len_;
}
}
return ret;
}
void tr_handshake::on_error(tr_peerIo* io, short what, void* vhandshake)
void tr_handshake::on_error(tr_peerIo* io, tr_error const& error, void* vhandshake)
{
auto const errcode = errno;
auto* handshake = static_cast<tr_handshake*>(vhandshake);
if (io->socket.is_utp() && !io->isIncoming() && handshake->is_state(State::AwaitingYb))
if (io->is_utp() && !io->is_incoming() && handshake->is_state(State::AwaitingYb))
{
// the peer probably doesn't speak µTP.
auto const info_hash = io->torrentHash();
auto const info_hash = io->torrent_hash();
auto const info = handshake->mediator_->torrent(info_hash);
/* Don't mark a peer as non-µTP unless it's really a connect failure. */
if ((errcode == ETIMEDOUT || errcode == ECONNREFUSED) && info)
if ((error.code == ETIMEDOUT || error.code == ECONNREFUSED) && info)
{
handshake->mediator_->set_utp_failed(info_hash, io->address());
}
if (handshake->mediator_->allows_tcp() && io->reconnect() == 0)
if (handshake->mediator_->allows_tcp() && io->reconnect())
{
auto msg = std::array<uint8_t, HandshakeSize>{};
handshake->build_handshake_message(io, std::data(msg));
handshake->have_sent_bittorrent_handshake_ = true;
handshake->set_state(State::AwaitingHandshake);
io->writeBytes(std::data(msg), std::size(msg), false);
io->write_bytes(std::data(msg), std::size(msg), false);
}
}
@ -818,20 +815,18 @@ void tr_handshake::on_error(tr_peerIo* io, short what, void* vhandshake)
* have encountered a peer that doesn't do encryption... reconnect and
* try a plaintext handshake */
if ((handshake->is_state(State::AwaitingYb) || handshake->is_state(State::AwaitingVc)) &&
handshake->encryption_mode_ != TR_ENCRYPTION_REQUIRED && handshake->mediator_->allows_tcp() && io->reconnect() == 0)
handshake->encryption_mode_ != TR_ENCRYPTION_REQUIRED && handshake->mediator_->allows_tcp() && io->reconnect())
{
auto msg = std::array<uint8_t, HandshakeSize>{};
tr_logAddTraceHand(handshake, "handshake failed, trying plaintext...");
handshake->build_handshake_message(io, std::data(msg));
handshake->have_sent_bittorrent_handshake_ = true;
handshake->set_state(State::AwaitingHandshake);
io->writeBytes(std::data(msg), std::size(msg), false);
io->write_bytes(std::data(msg), std::size(msg), false);
}
else
{
tr_logAddTraceHand(
handshake,
fmt::format("libevent got an error: what={:d}, errno={:d} ({:s})", what, errcode, tr_strerror(errcode)));
tr_logAddTraceHand(handshake, fmt::format("handshake socket err: {:s} ({:d})", error.message, error.code));
handshake->done(false);
}
}
@ -924,7 +919,7 @@ tr_handshake::tr_handshake(Mediator* mediator, std::shared_ptr<tr_peerIo> peer_i
{
timeout_timer_->startSingleShot(HandshakeTimeoutSec);
peer_io_->setCallbacks(&tr_handshake::can_read, nullptr, &tr_handshake::on_error, this);
peer_io_->set_callbacks(&tr_handshake::can_read, nullptr, &tr_handshake::on_error, this);
if (is_incoming())
{
@ -941,6 +936,6 @@ tr_handshake::tr_handshake(Mediator* mediator, std::shared_ptr<tr_peerIo> peer_i
have_sent_bittorrent_handshake_ = true;
set_state(State::AwaitingHandshake);
peer_io_->writeBytes(std::data(msg), std::size(msg), false);
peer_io_->write_bytes(std::data(msg), std::size(msg), false);
}
}

8
libtransmission/handshake.h
View File

@ -108,7 +108,7 @@ private:
static ReadState can_read(tr_peerIo* peer_io, void* vhandshake, size_t* piece);
static void on_error(tr_peerIo* io, short what, void* vhandshake);
static void on_error(tr_peerIo* io, tr_error const&, void* vhandshake);
bool build_handshake_message(tr_peerIo* io, uint8_t* buf) const;
@ -141,13 +141,13 @@ private:
ReadState done(bool is_connected)
{
peer_io_->clearCallbacks();
peer_io_->clear_callbacks();
return fire_done(is_connected) ? READ_LATER : READ_ERR;
}
[[nodiscard]] auto is_incoming() const noexcept
{
return peer_io_->isIncoming();
return peer_io_->is_incoming();
}
[[nodiscard]] constexpr auto state() const noexcept
@ -181,7 +181,7 @@ private:
auto walk = data;
walk = std::copy(std::begin(public_key), std::end(public_key), walk);
walk += mediator_->pad(walk, PadMax);
io->writeBytes(data, walk - data, false);
io->write_bytes(data, walk - data, false);
}
bool fire_done(bool is_connected);

24
libtransmission/net.cc
View File

@ -268,30 +268,6 @@ tr_peer_socket tr_netOpenPeerSocket(tr_session* session, tr_address const& addr,
return ret;
}
tr_peer_socket tr_netOpenPeerUTPSocket(tr_session* session, tr_address const& addr, tr_port port, bool /*client_is_seed*/)
{
auto ret = tr_peer_socket{};
if (session->utp_context != nullptr && addr.is_valid_for_peers(port))
{
auto const [ss, sslen] = addr.to_sockaddr(port);
if (auto* const sock = utp_create_socket(session->utp_context); sock != nullptr)
{
if (utp_connect(sock, reinterpret_cast<sockaddr const*>(&ss), sslen) != -1)
{
ret = tr_peer_socket{ addr, port, sock };
}
else
{
utp_close(sock);
}
}
}
return ret;
}
static tr_socket_t tr_netBindTCPImpl(tr_address const& addr, tr_port port, bool suppress_msgs, int* err_out)
{
TR_ASSERT(addr.is_valid());

832
libtransmission/peer-io.cc
View File

File diff suppressed because it is too large Load Diff

239
libtransmission/peer-io.h
View File

@ -31,10 +31,14 @@
#include "tr-buffer.h"
#include "utils-ev.h"
class tr_peerIo;
struct tr_bandwidth;
struct struct_utp_context;
namespace libtransmission::test
{
class HandshakeTest;
} // namespace libtransmission::test
/**
* @addtogroup networked_io Networked IO
* @{
@ -47,137 +51,144 @@ enum ReadState
READ_ERR
};
auto inline constexpr PEER_IO_MAGIC_NUMBER = 206745;
namespace libtransmission::test
{
class HandshakeTest;
} // namespace libtransmission::test
class tr_peerIo final : public std::enable_shared_from_this<tr_peerIo>
{
using DH = tr_message_stream_encryption::DH;
using Filter = tr_message_stream_encryption::Filter;
public:
using CanRead = ReadState (*)(tr_peerIo* io, void* user_data, size_t* setme_piece_byte_count);
using DidWrite = void (*)(tr_peerIo* io, size_t bytesWritten, bool wasPieceData, void* userData);
using GotError = void (*)(tr_peerIo* io, tr_error const& error, void* userData);
tr_peerIo(
tr_session* session_in,
tr_sha1_digest_t const* info_hash,
bool is_incoming,
bool is_seed,
tr_bandwidth* parent_bandwidth);
~tr_peerIo();
static std::shared_ptr<tr_peerIo> newOutgoing(
static std::shared_ptr<tr_peerIo> new_outgoing(
tr_session* session,
tr_bandwidth* parent,
tr_address const& addr,
tr_port port,
tr_sha1_digest_t const& torrent_hash,
tr_sha1_digest_t const& info_hash,
bool is_seed,
bool utp);
static std::shared_ptr<tr_peerIo> newIncoming(tr_session* session, tr_bandwidth* parent, tr_peer_socket socket);
static std::shared_ptr<tr_peerIo> new_incoming(tr_session* session, tr_bandwidth* parent, tr_peer_socket socket);
void set_socket(tr_peer_socket);
[[nodiscard]] constexpr auto is_utp() const noexcept
{
return socket_.is_utp();
}
void clear();
void readBytes(void* bytes, size_t byte_count);
void read_bytes(void* bytes, size_t byte_count);
void readUint8(uint8_t* setme)
void read_uint8(uint8_t* setme)
{
readBytes(setme, sizeof(uint8_t));
read_bytes(setme, sizeof(uint8_t));
}
void readUint16(uint16_t* setme);
void readUint32(uint32_t* setme);
void read_uint16(uint16_t* setme);
void read_uint32(uint32_t* setme);
int reconnect();
[[nodiscard]] bool reconnect();
void setEnabled(tr_direction dir, bool is_enabled);
void set_enabled(tr_direction dir, bool is_enabled);
[[nodiscard]] constexpr auto const& address() const noexcept
{
return socket.address();
return socket_.address();
}
[[nodiscard]] constexpr auto socketAddress() const noexcept
[[nodiscard]] constexpr auto socket_address() const noexcept
{
return socket.socketAddress();
return socket_.socketAddress();
}
[[nodiscard]] auto display_name() const
{
return socket.display_name();
return socket_.display_name();
}
void readBufferDrain(size_t byte_count);
void read_buffer_drain(size_t byte_count);
[[nodiscard]] auto readBufferSize() const noexcept
[[nodiscard]] auto read_buffer_size() const noexcept
{
return std::size(inbuf);
return std::size(inbuf_);
}
template<typename T>
[[nodiscard]] auto readBufferStartsWith(T const& t) const noexcept
[[nodiscard]] auto read_buffer_starts_with(T const& t) const noexcept
{
return inbuf.startsWith(t);
return inbuf_.startsWith(t);
}
void readBufferAdd(void const* data, size_t n_bytes);
size_t flush_outgoing_protocol_msgs();
size_t flush(tr_direction dir, size_t byte_limit);
size_t flushOutgoingProtocolMsgs(tr_error** error = nullptr);
size_t flush(tr_direction dir, size_t byte_limit, tr_error** error = nullptr);
void writeBytes(void const* bytes, size_t n_bytes, bool is_piece_data);
void write_bytes(void const* bytes, size_t n_bytes, bool is_piece_data);
// Write all the data from `buf`.
// This is a destructive add: `buf` is empty after this call.
void write(libtransmission::Buffer& buf, bool is_piece_data);
[[nodiscard]] size_t getWriteBufferSpace(uint64_t now) const noexcept;
[[nodiscard]] size_t get_write_buffer_space(uint64_t now) const noexcept;
[[nodiscard]] auto hasBandwidthLeft(tr_direction dir) noexcept
[[nodiscard]] auto has_bandwidth_left(tr_direction dir) noexcept
{
return bandwidth_.clamp(dir, 1024) > 0;
}
[[nodiscard]] auto getPieceSpeedBytesPerSecond(uint64_t now, tr_direction dir) noexcept
[[nodiscard]] auto get_piece_speed_bytes_per_second(uint64_t now, tr_direction dir) noexcept
{
return bandwidth_.getPieceSpeedBytesPerSecond(now, dir);
}
constexpr void enableFEXT(bool flag) noexcept
constexpr void set_supports_fext(bool flag) noexcept
{
fast_extension_supported_ = flag;
}
[[nodiscard]] constexpr auto supportsFEXT() const noexcept
[[nodiscard]] constexpr auto supports_fext() const noexcept
{
return fast_extension_supported_;
}
constexpr void enableLTEP(bool flag) noexcept
constexpr void set_supports_ltep(bool flag) noexcept
{
extended_protocol_supported_ = flag;
}
[[nodiscard]] constexpr auto supportsLTEP() const noexcept
[[nodiscard]] constexpr auto supports_ltep() const noexcept
{
return extended_protocol_supported_;
}
constexpr void enableDHT(bool flag) noexcept
constexpr void set_supports_dht(bool flag) noexcept
{
dht_supported_ = flag;
}
[[nodiscard]] constexpr auto supportsDHT() const noexcept
[[nodiscard]] constexpr auto supports_dht() const noexcept
{
return dht_supported_;
}
[[nodiscard]] constexpr auto supportsUTP() const noexcept
[[nodiscard]] constexpr auto supports_utp() const noexcept
{
return utp_supported_;
}
[[nodiscard]] constexpr auto isSeed() const noexcept
[[nodiscard]] constexpr auto is_seed() const noexcept
{
return is_seed_;
}
@ -192,68 +203,49 @@ public:
return bandwidth_;
}
void setParent(tr_bandwidth* parent)
void set_parent(tr_bandwidth* parent)
{
bandwidth_.setParent(parent);
}
[[nodiscard]] constexpr auto isIncoming() const noexcept
[[nodiscard]] constexpr auto is_incoming() const noexcept
{
return is_incoming_;
}
void setTorrentHash(tr_sha1_digest_t hash) noexcept
void set_torrent_hash(tr_sha1_digest_t hash) noexcept
{
torrent_hash_ = hash;
info_hash_ = hash;
}
[[nodiscard]] constexpr auto const& torrentHash() const noexcept
[[nodiscard]] constexpr auto const& torrent_hash() const noexcept
{
return torrent_hash_;
return info_hash_;
}
using tr_can_read_cb = ReadState (*)(tr_peerIo* io, void* user_data, size_t* setme_piece_byte_count);
using tr_did_write_cb = void (*)(tr_peerIo* io, size_t bytesWritten, bool wasPieceData, void* userData);
using tr_net_error_cb = void (*)(tr_peerIo* io, short what, void* userData);
void setCallbacks(tr_can_read_cb readcb, tr_did_write_cb writecb, tr_net_error_cb errcb, void* user_data);
void set_callbacks(CanRead can_read, DidWrite did_write, GotError got_error, void* user_data);
void clearCallbacks()
void clear_callbacks()
{
setCallbacks(nullptr, nullptr, nullptr, nullptr);
set_callbacks(nullptr, nullptr, nullptr, nullptr);
}
tr_peer_socket socket = {};
tr_session* const session;
tr_can_read_cb canRead = nullptr;
tr_did_write_cb didWrite = nullptr;
tr_net_error_cb gotError = nullptr;
void* userData = nullptr;
void call_error_callback(short what)
[[nodiscard]] constexpr auto priority() const noexcept
{
if (gotError != nullptr)
{
gotError(this, what, userData);
}
return priority_;
}
libtransmission::Buffer inbuf;
libtransmission::Buffer outbuf;
[[nodiscard]] constexpr auto is_encrypted() const noexcept
{
return filter_.is_active();
}
std::deque<std::pair<size_t /*n_bytes*/, bool /*is_piece_data*/>> outbuf_info;
constexpr void set_priority(tr_priority_t priority)
{
priority_ = priority;
}
libtransmission::evhelpers::event_unique_ptr event_read;
libtransmission::evhelpers::event_unique_ptr event_write;
short int pendingEvents = 0;
tr_priority_t priority = TR_PRI_NORMAL;
bool utp_supported_ = false;
void decryptInit(bool is_incoming, DH const& dh, tr_sha1_digest_t const& info_hash)
void decrypt_init(bool is_incoming, DH const& dh, tr_sha1_digest_t const& info_hash)
{
filter_.decryptInit(is_incoming, dh, info_hash);
}
@ -263,53 +255,88 @@ public:
filter_.decrypt(buflen, buf);
}
void encryptInit(bool is_incoming, DH const& dh, tr_sha1_digest_t const& info_hash)
void encrypt_init(bool is_incoming, DH const& dh, tr_sha1_digest_t const& info_hash)
{
filter_.encryptInit(is_incoming, dh, info_hash);
}
static void utp_init(struct_utp_context* ctx);
private:
static constexpr auto RcvBuf = size_t{ 256 * 1024 };
friend class libtransmission::test::HandshakeTest;
void call_error_callback(tr_error const& error)
{
if (got_error_ != nullptr)
{
got_error_(this, error, user_data_);
}
}
void encrypt(size_t buflen, void* buf)
{
filter_.encrypt(buflen, buf);
}
[[nodiscard]] bool isEncrypted() const noexcept
{
return filter_.is_active();
}
void on_utp_state_change(int new_state);
void on_utp_error(int errcode);
static void utpInit(struct_utp_context* ctx);
void close();
tr_peerIo(
tr_session* session_in,
tr_sha1_digest_t const* torrent_hash,
bool is_incoming,
bool is_seed,
tr_bandwidth* parent_bandwidth,
tr_peer_socket sock);
static void event_read_cb(evutil_socket_t fd, short /*event*/, void* vio);
static void event_write_cb(evutil_socket_t fd, short /*event*/, void* vio);
private:
friend class libtransmission::test::HandshakeTest;
void event_enable(short event);
void event_disable(short event);
void can_read_wrapper();
void did_write_wrapper(size_t bytes_transferred);
size_t try_read(size_t max);
size_t try_write(size_t max);
// this is only public for testing purposes.
// production code should use newOutgoing() or newIncoming()
// production code should use new_outgoing() or new_incoming()
static std::shared_ptr<tr_peerIo> create(
tr_session* session,
tr_bandwidth* parent,
tr_sha1_digest_t const* torrent_hash,
tr_sha1_digest_t const* info_hash,
bool is_incoming,
bool is_seed,
tr_peer_socket socket);
tr_bandwidth bandwidth_;
bool is_seed);
Filter filter_;
tr_sha1_digest_t torrent_hash_;
std::deque<std::pair<size_t /*n_bytes*/, bool /*is_piece_data*/>> outbuf_info_;
tr_peer_socket socket_ = {};
tr_bandwidth bandwidth_;
tr_sha1_digest_t info_hash_;
libtransmission::Buffer inbuf_;
libtransmission::Buffer outbuf_;
tr_session* const session_;
CanRead can_read_ = nullptr;
DidWrite did_write_ = nullptr;
GotError got_error_ = nullptr;
void* user_data_ = nullptr;
libtransmission::evhelpers::event_unique_ptr event_read_;
libtransmission::evhelpers::event_unique_ptr event_write_;
short int pending_events_ = 0;
tr_priority_t priority_ = TR_PRI_NORMAL;
bool const is_seed_;
bool const is_incoming_;
bool utp_supported_ = false;
bool dht_supported_ = false;
bool extended_protocol_supported_ = false;
bool fast_extension_supported_ = false;

18
libtransmission/peer-mgr.cc
View File

@ -1089,11 +1089,11 @@ static bool on_handshake_done(tr_peerMgr* manager, tr_handshake::Result const& r
bool const ok = result.is_connected;
bool success = false;
tr_swarm* const s = getExistingSwarm(manager, result.io->torrentHash());
tr_swarm* const s = getExistingSwarm(manager, result.io->torrent_hash());
auto const [addr, port] = result.io->socketAddress();
auto const [addr, port] = result.io->socket_address();
if (result.io->isIncoming())
if (result.io->is_incoming())
{
manager->incoming_handshakes.erase(addr);
}
@ -1135,7 +1135,7 @@ static bool on_handshake_done(tr_peerMgr* manager, tr_handshake::Result const& r
atom->piece_data_time = 0;
atom->lastConnectionAt = tr_time();
if (!result.io->isIncoming())
if (!result.io->is_incoming())
{
atom->flags |= ADDED_F_CONNECTABLE;
atom->flags2 &= ~MyflagUnreachable;
@ -1143,7 +1143,7 @@ static bool on_handshake_done(tr_peerMgr* manager, tr_handshake::Result const& r
/* In principle, this flag specifies whether the peer groks µTP,
not whether it's currently connected over µTP. */
if (result.io->socket.is_utp())
if (result.io->is_utp())
{
atom->flags |= ADDED_F_UTP_FLAGS;
}
@ -1152,7 +1152,7 @@ static bool on_handshake_done(tr_peerMgr* manager, tr_handshake::Result const& r
{
tr_logAddTraceSwarm(s, fmt::format("banned peer {} tried to reconnect", atom->display_name()));
}
else if (result.io->isIncoming() && s->peerCount() >= s->tor->peerLimit())
else if (result.io->is_incoming() && s->peerCount() >= s->tor->peerLimit())
{
/* too many peers already */
}
@ -1170,7 +1170,7 @@ static bool on_handshake_done(tr_peerMgr* manager, tr_handshake::Result const& r
client = tr_quark_new(std::data(buf));
}
result.io->setParent(&s->tor->bandwidth_);
result.io->set_parent(&s->tor->bandwidth_);
createBitTorrentPeer(s->tor, result.io, atom, client);
success = true;
@ -1202,7 +1202,7 @@ void tr_peerMgrAddIncoming(tr_peerMgr* manager, tr_peer_socket&& socket)
manager->incoming_handshakes.try_emplace(
address,
&manager->handshake_mediator_,
tr_peerIo::newIncoming(session, &session->top_bandwidth_, std::move(socket)),
tr_peerIo::new_incoming(session, &session->top_bandwidth_, std::move(socket)),
session->encryptionMode(),
[manager](tr_handshake::Result const& result) { return on_handshake_done(manager, result); });
}
@ -2759,7 +2759,7 @@ void initiateConnection(tr_peerMgr* mgr, tr_swarm* s, peer_atom& atom)
s,
fmt::format("Starting an OUTGOING {} connection with {}", utp ? " µTP" : "TCP", atom.display_name()));
auto peer_io = tr_peerIo::newOutgoing(
auto peer_io = tr_peerIo::new_outgoing(
mgr->session,
&mgr->session->top_bandwidth_,
atom.addr,

142
libtransmission/peer-msgs.cc
View File

@ -203,7 +203,7 @@ class tr_peerMsgsImpl;
static ReadState canRead(tr_peerIo* io, void* vmsgs, size_t* piece);
static void cancelAllRequestsToClient(tr_peerMsgsImpl* msgs);
static void didWrite(tr_peerIo* io, size_t bytes_written, bool was_piece_data, void* vmsgs);
static void gotError(tr_peerIo* io, short what, void* vmsgs);
static void gotError(tr_peerIo* io, tr_error const& err, void* vmsgs);
static void peerPulse(void* vmsgs);
static void protocolSendCancel(tr_peerMsgsImpl* msgs, struct peer_request const& req);
static void protocolSendChoke(tr_peerMsgsImpl* msgs, bool choke);
@ -268,20 +268,20 @@ public:
pex_timer_->startRepeating(SendPexInterval);
}
if (io->supportsUTP())
if (io->supports_utp())
{
tr_peerMgrSetUtpSupported(torrent, io->address());
tr_peerMgrSetUtpFailed(torrent, io->address(), false);
}
if (io->supportsLTEP())
if (io->supports_ltep())
{
sendLtepHandshake(this);
}
tellPeerWhatWeHave(this);
if (session->allowsDHT() && io->supportsDHT())
if (session->allowsDHT() && io->supports_dht())
{
// only send PORT over IPv6 iff IPv6 DHT is running (BEP-32).
if (io->address().is_ipv4() || tr_globalIPv6(nullptr).has_value())
@ -290,7 +290,7 @@ public:
}
}
io->setCallbacks(canRead, didWrite, gotError, this);
io->set_callbacks(canRead, didWrite, gotError, this);
updateDesiredRequestCount(this);
}
@ -326,7 +326,7 @@ public:
bool isTransferringPieces(uint64_t now, tr_direction dir, tr_bytes_per_second_t* setme_bytes_per_second) const override
{
auto const bytes_per_second = io->getPieceSpeedBytesPerSecond(now, dir);
auto const bytes_per_second = io->get_piece_speed_bytes_per_second(now, dir);
if (setme_bytes_per_second != nullptr)
{
@ -374,17 +374,17 @@ public:
[[nodiscard]] bool is_utp_connection() const noexcept override
{
return io->socket.is_utp();
return io->is_utp();
}
[[nodiscard]] bool is_encrypted() const override
{
return io->isEncrypted();
return io->is_encrypted();
}
[[nodiscard]] bool is_incoming_connection() const override
{
return io->isIncoming();
return io->is_incoming();
}
[[nodiscard]] tr_bandwidth& bandwidth() noexcept override
@ -409,7 +409,7 @@ public:
[[nodiscard]] std::pair<tr_address, tr_port> socketAddress() const override
{
return io->socketAddress();
return io->socket_address();
}
[[nodiscard]] std::string display_name() const override
@ -729,7 +729,7 @@ tr_peerMsgs* tr_peerMsgsNew(
static void protocolSendReject(tr_peerMsgsImpl* msgs, struct peer_request const* req)
{
TR_ASSERT(msgs->io->supportsFEXT());
TR_ASSERT(msgs->io->supports_fext());
auto& out = msgs->outMessages;
@ -795,7 +795,7 @@ static void protocolSendChoke(tr_peerMsgsImpl* msgs, bool choke)
static void protocolSendHaveAll(tr_peerMsgsImpl* msgs)
{
TR_ASSERT(msgs->io->supportsFEXT());
TR_ASSERT(msgs->io->supports_fext());
auto& out = msgs->outMessages;
@ -809,7 +809,7 @@ static void protocolSendHaveAll(tr_peerMsgsImpl* msgs)
static void protocolSendHaveNone(tr_peerMsgsImpl* msgs)
{
TR_ASSERT(msgs->io->supportsFEXT());
TR_ASSERT(msgs->io->supports_fext());
auto& out = msgs->outMessages;
@ -854,7 +854,7 @@ static bool popNextMetadataRequest(tr_peerMsgsImpl* msgs, int* setme)
static void cancelAllRequestsToClient(tr_peerMsgsImpl* msgs)
{
if (auto const must_send_rej = msgs->io->supportsFEXT(); must_send_rej)
if (auto const must_send_rej = msgs->io->supports_fext(); must_send_rej)
{
for (auto const& req : msgs->peer_requested_)
{
@ -872,7 +872,7 @@ static void cancelAllRequestsToClient(tr_peerMsgsImpl* msgs)
static void sendLtepHandshake(tr_peerMsgsImpl* msgs)
{
auto& out = msgs->outMessages;
auto const ipv6 = tr_globalIPv6(msgs->io->session);
auto const ipv6 = tr_globalIPv6(msgs->session);
static tr_quark version_quark = 0;
if (msgs->clientSentLtepHandshake)
@ -987,7 +987,7 @@ static void parseLtepHandshake(tr_peerMsgsImpl* msgs, uint32_t len)
// so try using a strbuf to handle it on the stack
auto tmp = tr_strbuf<char, 512>{};
tmp.resize(len);
msgs->io->readBytes(std::data(tmp), std::size(tmp));
msgs->io->read_bytes(std::data(tmp), std::size(tmp));
auto const handshake_sv = tmp.sv();
auto val = tr_variant{};
@ -1062,14 +1062,14 @@ static void parseLtepHandshake(tr_peerMsgsImpl* msgs, uint32_t len)
uint8_t const* addr = nullptr;
auto addr_len = size_t{};
if (msgs->io->isIncoming() && tr_variantDictFindRaw(&val, TR_KEY_ipv4, &addr, &addr_len) && addr_len == 4)
if (msgs->io->is_incoming() && tr_variantDictFindRaw(&val, TR_KEY_ipv4, &addr, &addr_len) && addr_len == 4)
{
pex.addr.type = TR_AF_INET;
memcpy(&pex.addr.addr.addr4, addr, 4);
tr_peerMgrAddPex(msgs->torrent, TR_PEER_FROM_LTEP, &pex, 1);
}
if (msgs->io->isIncoming() && tr_variantDictFindRaw(&val, TR_KEY_ipv6, &addr, &addr_len) && addr_len == 16)
if (msgs->io->is_incoming() && tr_variantDictFindRaw(&val, TR_KEY_ipv6, &addr, &addr_len) && addr_len == 16)
{
pex.addr.type = TR_AF_INET6;
memcpy(&pex.addr.addr.addr6, addr, 16);
@ -1093,7 +1093,7 @@ static void parseUtMetadata(tr_peerMsgsImpl* msgs, uint32_t msglen)
auto tmp = std::vector<char>{};
tmp.resize(msglen);
msgs->io->readBytes(std::data(tmp), std::size(tmp));
msgs->io->read_bytes(std::data(tmp), std::size(tmp));
char const* const msg_end = std::data(tmp) + std::size(tmp);
auto dict = tr_variant{};
@ -1164,7 +1164,7 @@ static void parseUtPex(tr_peerMsgsImpl* msgs, uint32_t msglen)
auto tmp = std::vector<char>{};
tmp.resize(msglen);
msgs->io->readBytes(std::data(tmp), std::size(tmp));
msgs->io->read_bytes(std::data(tmp), std::size(tmp));
if (tr_variant val; tr_variantFromBuf(&val, TR_VARIANT_PARSE_BENC | TR_VARIANT_PARSE_INPLACE, tmp))
{
@ -1209,7 +1209,7 @@ static void parseLtep(tr_peerMsgsImpl* msgs, uint32_t msglen)
TR_ASSERT(msglen > 0);
auto ltep_msgid = uint8_t{};
msgs->io->readUint8(&ltep_msgid);
msgs->io->read_uint8(&ltep_msgid);
msglen--;
if (ltep_msgid == LtepMessages::Handshake)
@ -1217,7 +1217,7 @@ static void parseLtep(tr_peerMsgsImpl* msgs, uint32_t msglen)
logtrace(msgs, "got ltep handshake");
parseLtepHandshake(msgs, msglen);
if (msgs->io->supportsLTEP())
if (msgs->io->supports_ltep())
{
sendLtepHandshake(msgs);
msgs->sendPex();
@ -1238,7 +1238,7 @@ static void parseLtep(tr_peerMsgsImpl* msgs, uint32_t msglen)
else
{
logtrace(msgs, fmt::format(FMT_STRING("skipping unknown ltep message ({:d})"), static_cast<int>(ltep_msgid)));
msgs->io->readBufferDrain(msglen);
msgs->io->read_buffer_drain(msglen);
}
}
@ -1250,7 +1250,7 @@ static ReadState readBtLength(tr_peerMsgsImpl* msgs, size_t inlen)
return READ_LATER;
}
msgs->io->readUint32(&len);
msgs->io->read_uint32(&len);
if (len == 0) /* peer sent us a keepalive message */
{
logtrace(msgs, "got KeepAlive");
@ -1274,7 +1274,7 @@ static ReadState readBtId(tr_peerMsgsImpl* msgs, size_t inlen)
}
auto id = uint8_t{};
msgs->io->readUint8(&id);
msgs->io->read_uint8(&id);
msgs->incoming.id = id;
logtrace(
msgs,
@ -1350,7 +1350,7 @@ static void peerMadeRequest(tr_peerMsgsImpl* msgs, struct peer_request const* re
msgs->peer_requested_.emplace_back(*req);
prefetchPieces(msgs);
}
else if (msgs->io->supportsFEXT())
else if (msgs->io->supports_fext())
{
protocolSendReject(msgs, req);
}
@ -1411,7 +1411,7 @@ static int clientGotBlock(tr_peerMsgsImpl* msgs, std::unique_ptr<std::vector<uin
static ReadState readBtPiece(tr_peerMsgsImpl* msgs, size_t inlen, size_t* setme_piece_bytes_read)
{
TR_ASSERT(msgs->io->readBufferSize() >= inlen);
TR_ASSERT(msgs->io->read_buffer_size() >= inlen);
logtrace(msgs, "In readBtPiece");
@ -1424,8 +1424,8 @@ static ReadState readBtPiece(tr_peerMsgsImpl* msgs, size_t inlen, size_t* setme_
}
auto req = peer_request{};
msgs->io->readUint32(&req.index);
msgs->io->readUint32(&req.offset);
msgs->io->read_uint32(&req.index);
msgs->io->read_uint32(&req.offset);
req.length = msgs->incoming.length - 9;
logtrace(msgs, fmt::format(FMT_STRING("got incoming block header {:d}:{:d}->{:d}"), req.index, req.offset, req.length));
msgs->incoming.block_req = req;
@ -1449,7 +1449,7 @@ static ReadState readBtPiece(tr_peerMsgsImpl* msgs, size_t inlen, size_t* setme_
auto const n_to_read = std::min({ n_left_in_block, n_left_in_req, inlen });
auto const old_length = std::size(*block_buf);
block_buf->resize(old_length + n_to_read);
msgs->io->readBytes(&((*block_buf)[old_length]), n_to_read);
msgs->io->read_bytes(&((*block_buf)[old_length]), n_to_read);
msgs->publish(tr_peer_event::GotPieceData(n_to_read));
*setme_piece_bytes_read += n_to_read;
@ -1493,9 +1493,9 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
{
uint8_t const id = msgs->incoming.id;
#ifdef TR_ENABLE_ASSERTS
auto const start_buflen = msgs->io->readBufferSize();
auto const start_buflen = msgs->io->read_buffer_size();
#endif
bool const fext = msgs->io->supportsFEXT();
bool const fext = msgs->io->supports_fext();
auto ui32 = uint32_t{};
auto msglen = uint32_t{ msgs->incoming.length };
@ -1556,7 +1556,7 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
break;
case BtPeerMsgs::Have:
msgs->io->readUint32(&ui32);
msgs->io->read_uint32(&ui32);
logtrace(msgs, fmt::format(FMT_STRING("got Have: {:d}"), ui32));
if (msgs->torrent->hasMetainfo() && ui32 >= msgs->torrent->pieceCount())
@ -1579,7 +1579,7 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
{
logtrace(msgs, "got a bitfield");
auto tmp = std::vector<uint8_t>(msglen);
msgs->io->readBytes(std::data(tmp), std::size(tmp));
msgs->io->read_bytes(std::data(tmp), std::size(tmp));
msgs->have_ = tr_bitfield{ msgs->torrent->hasMetainfo() ? msgs->torrent->pieceCount() : std::size(tmp) * 8 };
msgs->have_.setRaw(std::data(tmp), std::size(tmp));
msgs->publish(tr_peer_event::GotBitfield(&msgs->have_));
@ -1590,9 +1590,9 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
case BtPeerMsgs::Request:
{
struct peer_request r;
msgs->io->readUint32(&r.index);
msgs->io->readUint32(&r.offset);
msgs->io->readUint32(&r.length);
msgs->io->read_uint32(&r.index);
msgs->io->read_uint32(&r.offset);
msgs->io->read_uint32(&r.length);
logtrace(msgs, fmt::format(FMT_STRING("got Request: {:d}:{:d}->{:d}"), r.index, r.offset, r.length));
peerMadeRequest(msgs, &r);
break;
@ -1601,9 +1601,9 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
case BtPeerMsgs::Cancel:
{
struct peer_request r;
msgs->io->readUint32(&r.index);
msgs->io->readUint32(&r.offset);
msgs->io->readUint32(&r.length);
msgs->io->read_uint32(&r.index);
msgs->io->read_uint32(&r.offset);
msgs->io->read_uint32(&r.length);
msgs->cancels_sent_to_client.add(tr_time(), 1);
logtrace(msgs, fmt::format(FMT_STRING("got a Cancel {:d}:{:d}->{:d}"), r.index, r.offset, r.length));
@ -1638,7 +1638,7 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
logtrace(msgs, "Got a BtPeerMsgs::Port");
auto hport = uint16_t{};
msgs->io->readUint16(&hport); // readUint16 performs ntoh
msgs->io->read_uint16(&hport); // read_uint16 performs ntoh
if (auto const dht_port = tr_port::fromHost(hport); !std::empty(dht_port))
{
msgs->dht_port = dht_port;
@ -1649,7 +1649,7 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
case BtPeerMsgs::FextSuggest:
logtrace(msgs, "Got a BtPeerMsgs::FextSuggest");
msgs->io->readUint32(&ui32);
msgs->io->read_uint32(&ui32);
if (fext)
{
@ -1665,7 +1665,7 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
case BtPeerMsgs::FextAllowedFast:
logtrace(msgs, "Got a BtPeerMsgs::FextAllowedFast");
msgs->io->readUint32(&ui32);
msgs->io->read_uint32(&ui32);
if (fext)
{
@ -1717,9 +1717,9 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
{
struct peer_request r;
logtrace(msgs, "Got a BtPeerMsgs::FextReject");
msgs->io->readUint32(&r.index);
msgs->io->readUint32(&r.offset);
msgs->io->readUint32(&r.length);
msgs->io->read_uint32(&r.index);
msgs->io->read_uint32(&r.offset);
msgs->io->read_uint32(&r.length);
if (fext)
{
@ -1742,12 +1742,12 @@ static ReadState readBtMessage(tr_peerMsgsImpl* msgs, size_t inlen)
default:
logtrace(msgs, fmt::format(FMT_STRING("peer sent us an UNKNOWN: {:d}"), static_cast<int>(id)));
msgs->io->readBufferDrain(msglen);
msgs->io->read_buffer_drain(msglen);
break;
}
TR_ASSERT(msglen + 1 == msgs->incoming.length);
TR_ASSERT(msgs->io->readBufferSize() == start_buflen - msglen);
TR_ASSERT(msgs->io->read_buffer_size() == start_buflen - msglen);
msgs->state = AwaitingBt::Length;
return READ_NOW;
@ -1808,25 +1808,22 @@ static int clientGotBlock(
return 0;
}
static void didWrite(tr_peerIo* io, size_t bytes_written, bool was_piece_data, void* vmsgs)
static void didWrite(tr_peerIo* /*io*/, size_t bytes_written, bool was_piece_data, void* vmsgs)
{
auto* msgs = static_cast<tr_peerMsgsImpl*>(vmsgs);
auto* const msgs = static_cast<tr_peerMsgsImpl*>(vmsgs);
if (was_piece_data)
{
msgs->publish(tr_peer_event::SentPieceData(bytes_written));
}
if (tr_isPeerIo(io) && io->userData != nullptr)
{
peerPulse(msgs);
}
peerPulse(msgs);
}
static ReadState canRead(tr_peerIo* io, void* vmsgs, size_t* piece)
{
auto* msgs = static_cast<tr_peerMsgsImpl*>(vmsgs);
size_t const inlen = io->readBufferSize();
size_t const inlen = io->read_buffer_size();
logtrace(
msgs,
@ -1947,7 +1944,7 @@ static size_t fillOutputBuffer(tr_peerMsgsImpl* msgs, time_t now)
size_t bytes_written = 0;
struct peer_request req;
bool const have_messages = !std::empty(msgs->outMessages);
bool const fext = msgs->io->supportsFEXT();
bool const fext = msgs->io->supports_fext();
/**
*** Protocol messages
@ -1974,7 +1971,8 @@ static size_t fillOutputBuffer(tr_peerMsgsImpl* msgs, time_t now)
*** Metadata Pieces
**/
if (auto piece = int{}; msgs->io->getWriteBufferSpace(now) >= METADATA_PIECE_SIZE && popNextMetadataRequest(msgs, &piece))
if (auto piece = int{};
msgs->io->get_write_buffer_space(now) >= METADATA_PIECE_SIZE && popNextMetadataRequest(msgs, &piece))
{
auto ok = bool{ false };
@ -2031,7 +2029,7 @@ static size_t fillOutputBuffer(tr_peerMsgsImpl* msgs, time_t now)
*** Data Blocks
**/
if (msgs->io->getWriteBufferSpace(now) >= tr_block_info::BlockSize && !std::empty(msgs->peer_requested_))
if (msgs->io->get_write_buffer_space(now) >= tr_block_info::BlockSize && !std::empty(msgs->peer_requested_))
{
req = msgs->peer_requested_.front();
msgs->peer_requested_.erase(std::begin(msgs->peer_requested_));
@ -2133,31 +2131,9 @@ static void peerPulse(void* vmsgs)
}
}
static void gotError(tr_peerIo* io, short what, void* vmsgs)
static void gotError(tr_peerIo* /*io*/, tr_error const& /*error*/, void* vmsgs)
{
auto* msgs = static_cast<tr_peerMsgsImpl*>(vmsgs);
if ((what & BEV_EVENT_TIMEOUT) != 0)
{
logdbg(msgs, fmt::format(FMT_STRING("libevent got a timeout, what={:d}"), what));
}
if ((what & BEV_EVENT_EOF) != 0)
{
logdbg(msgs, fmt::format("peer closed connection. {:s}", io->display_name()));
}
else if (what == BEV_EVENT_ERROR)
{
// exact BEV_EVENT_ERROR are high frequency errors from utp_on_error which were already logged appropriately
logtrace(msgs, fmt::format("libevent got an error! what={:d}, errno={:d} ({:s})", what, errno, tr_strerror(errno)));
}
else if ((what & BEV_EVENT_ERROR) != 0)
{
// read or write error
logdbg(msgs, fmt::format("libevent got an error! what={:d}, errno={:d} ({:s})", what, errno, tr_strerror(errno)));
}
msgs->publish(tr_peer_event::GotError(ENOTCONN));
static_cast<tr_peerMsgsImpl*>(vmsgs)->publish(tr_peer_event::GotError(ENOTCONN));
}
static void sendBitfield(tr_peerMsgsImpl* msgs)
@ -2176,7 +2152,7 @@ static void sendBitfield(tr_peerMsgsImpl* msgs)
static void tellPeerWhatWeHave(tr_peerMsgsImpl* msgs)
{
bool const fext = msgs->io->supportsFEXT();
bool const fext = msgs->io->supports_fext();
if (fext && msgs->torrent->hasAll())
{

98
libtransmission/peer-socket.cc
View File

@ -49,7 +49,7 @@ tr_peer_socket::tr_peer_socket(tr_address const& address, tr_port port, struct U
void tr_peer_socket::close(tr_session* session)
{
if (is_tcp())
if (is_tcp() && (handle.tcp != TR_BAD_SOCKET))
{
tr_netClose(session, handle.tcp);
}
@ -64,3 +64,99 @@ void tr_peer_socket::close(tr_session* session)
type_ = Type::None;
handle = {};
}
size_t tr_peer_socket::try_write(Buffer& buf, size_t max, tr_error** error) const
{
if (max == size_t{})
{
return {};
}
if (is_tcp())
{
return buf.toSocket(handle.tcp, max, error);
}
#ifdef WITH_UTP
if (is_utp())
{
auto iov = buf.vecs(max);
errno = 0;
auto const n = utp_writev(handle.utp, reinterpret_cast<struct utp_iovec*>(std::data(iov)), std::size(iov));
auto const error_code = errno;
if (n > 0)
{
buf.drain(n);
return static_cast<size_t>(n);
}
if (n < 0 && error_code != 0)
{
tr_error_set(error, error_code, tr_strerror(error_code));
}
}
#endif
return {};
}
size_t tr_peer_socket::try_read(Buffer& buf, size_t max, tr_error** error) const
{
if (max == size_t{})
{
return {};
}
if (is_tcp())
{
return buf.addSocket(handle.tcp, max, error);
}
#ifdef WITH_UTP
if (is_utp())
{
// utp_read_drained() notifies libutp that this read buffer is
// empty. It opens up the congestion window by sending an ACK
// (soonish) if one was not going to be sent.
if (std::empty(buf))
{
utp_read_drained(handle.utp);
}
}
#endif
return {};
}
tr_peer_socket tr_netOpenPeerUTPSocket(
tr_session* session,
tr_address const& addr,
tr_port port,
bool /*client_is_seed*/,
void* userdata)
{
auto ret = tr_peer_socket{};
if (session->utp_context != nullptr && addr.is_valid_for_peers(port))
{
auto const [ss, sslen] = addr.to_sockaddr(port);
if (auto* const sock = utp_create_socket(session->utp_context); sock != nullptr)
{
utp_set_userdata(sock, userdata);
if (utp_connect(sock, reinterpret_cast<sockaddr const*>(&ss), sslen) != -1)
{
ret = tr_peer_socket{ addr, port, sock };
}
else
{
utp_close(sock);
}
}
}
return ret;
}

40
libtransmission/peer-socket.h
View File

@ -15,8 +15,10 @@
#include "transmission.h"
#include "error.h"
#include "net.h"
#include "tr-assert.h"
#include "tr-buffer.h"
struct UTPSocket;
struct tr_session;
@ -24,6 +26,8 @@ struct tr_session;
class tr_peer_socket
{
public:
using Buffer = libtransmission::Buffer;
tr_peer_socket() = default;
tr_peer_socket(tr_session* session, tr_address const& address, tr_port port, tr_socket_t sock);
tr_peer_socket(tr_address const& address, tr_port port, struct UTPSocket* const sock);
@ -35,6 +39,9 @@ public:
void close(tr_session* session);
size_t try_write(Buffer& buf, size_t max, tr_error** error) const;
size_t try_read(Buffer& buf, size_t max, tr_error** error) const;
[[nodiscard]] constexpr std::pair<tr_address, tr_port> socketAddress() const noexcept
{
return std::make_pair(address_, port_);
@ -85,6 +92,32 @@ public:
#endif
}
[[nodiscard]] constexpr size_t guess_packet_overhead(size_t n_bytes) const noexcept
{
if (is_tcp())
{
// https://web.archive.org/web/20140912230020/http://sd.wareonearth.com:80/~phil/net/overhead/
// TCP over Ethernet:
// Assuming no header compression (e.g. not PPP)
// Add 20 IPv4 header or 40 IPv6 header (no options)
// Add 20 TCP header
// Add 12 bytes optional TCP timestamps
// Max TCP Payload data rates over ethernet are thus:
// (1500-40)/ (38+1500) = 94.9285 % IPv4, minimal headers
// (1500-52)/ (38+1500) = 94.1482 % IPv4, TCP timestamps
// (1500-52)/ (42+1500) = 93.9040 % 802.1q, IPv4, TCP timestamps
// (1500-60)/ (38+1500) = 93.6281 % IPv6, minimal headers
// (1500-72)/ (38+1500) = 92.8479 % IPv6, TCP timestamps
// (1500-72)/ (42+1500) = 92.6070 % 802.1q, IPv6, TCP timestamps
// So, let's guess around 7% overhead
return n_bytes / 14U;
}
// We only guess for TCP; uTP tracks its overhead via UTP_ON_OVERHEAD_STATISTICS
return {};
}
union
{
tr_socket_t tcp;
@ -106,4 +139,9 @@ private:
};
tr_peer_socket tr_netOpenPeerSocket(tr_session* session, tr_address const& addr, tr_port port, bool client_is_seed);
tr_peer_socket tr_netOpenPeerUTPSocket(tr_session* session, tr_address const& addr, tr_port port, bool client_is_seed);
tr_peer_socket tr_netOpenPeerUTPSocket(
tr_session* session,
tr_address const& addr,
tr_port port,
bool client_is_seed,
void* userdata);

18
libtransmission/tr-buffer.h
View File

@ -281,17 +281,27 @@ public:
evbuffer_expand(buf_.get(), n_bytes - size());
}
// -1 on error, 0 on eof, >0 for num bytes read
auto addSocket(tr_socket_t sockfd, size_t n_bytes, tr_error** error = nullptr)
size_t addSocket(tr_socket_t sockfd, size_t n_bytes, tr_error** error = nullptr)
{
EVUTIL_SET_SOCKET_ERROR(0);
auto const res = evbuffer_read(buf_.get(), sockfd, static_cast<int>(n_bytes));
auto const err = EVUTIL_SOCKET_ERROR();
if (res == -1)
if (res > 0)
{
return static_cast<size_t>(res);
}
if (res == 0)
{
tr_error_set(error, ENOTCONN, tr_strerror(ENOTCONN));
}
else
{
tr_error_set(error, err, tr_net_strerror(err));
}
return res;
return {};
}
// Move all data from one buffer into another.

20
libtransmission/tr-utp.cc
View File

@ -112,15 +112,6 @@ static void utp_send_to(
ss->udp_core_->sendto(buf, buflen, to, tolen);
}
#ifdef TR_UTP_TRACE
static void utp_log(tr_session* const /*session*/, char const* const msg)
{
fmt::print(stderr, FMT_STRING("[µTP] {}\n"), msg);
}
#endif
static uint64 utp_callback(utp_callback_arguments* args)
{
auto* const session = static_cast<tr_session*>(utp_context_get_userdata(args->context));
@ -131,11 +122,9 @@ static uint64 utp_callback(utp_callback_arguments* args)
switch (args->callback_type)
{
#ifdef TR_UTP_TRACE
case UTP_LOG:
utp_log(session, args->buf);
fmt::print(stderr, FMT_STRING("[µTP] {}\n"), args->buf);
break;
#endif
case UTP_ON_ACCEPT:
@ -204,20 +193,15 @@ void tr_utpInit(tr_session* session)
}
utp_context_set_userdata(ctx, session);
utp_set_callback(ctx, UTP_ON_ACCEPT, &utp_callback);
utp_set_callback(ctx, UTP_SENDTO, &utp_callback);
tr_peerIo::utpInit(ctx);
tr_peerIo::utp_init(ctx);
#ifdef TR_UTP_TRACE
utp_set_callback(ctx, UTP_LOG, &utp_callback);
utp_context_set_option(ctx, UTP_LOG_NORMAL, 1);
utp_context_set_option(ctx, UTP_LOG_MTU, 1);
utp_context_set_option(ctx, UTP_LOG_DEBUG, 1);
#endif
session->utp_context = ctx;

32
tests/libtransmission/handshake-test.cc
View File

@ -159,7 +159,7 @@ public:
auto sockpair = std::array<evutil_socket_t, 2>{ -1, -1 };
EXPECT_EQ(0, evutil_socketpair(LOCAL_SOCKETPAIR_AF, SOCK_STREAM, 0, std::data(sockpair))) << tr_strerror(errno);
return std::make_pair(
tr_peerIo::newIncoming(
tr_peerIo::new_incoming(
session,
&session->top_bandwidth_,
tr_peer_socket(session, DefaultPeerAddr, DefaultPeerPort, sockpair[0])),
@ -170,15 +170,9 @@ public:
{
auto sockpair = std::array<evutil_socket_t, 2>{ -1, -1 };
EXPECT_EQ(0, evutil_socketpair(LOCAL_SOCKETPAIR_AF, SOCK_STREAM, 0, std::data(sockpair))) << tr_strerror(errno);
return std::make_pair(
tr_peerIo::create(
session,
&session->top_bandwidth_,
&info_hash,
false /*is_incoming*/,
false /*is_seed*/,
tr_peer_socket(session, DefaultPeerAddr, DefaultPeerPort, sockpair[0])),
sockpair[1]);
auto peer_io = tr_peerIo::create(session, &session->top_bandwidth_, &info_hash, false /*incoming*/, false /*seed*/);
peer_io->set_socket(tr_peer_socket(session, DefaultPeerAddr, DefaultPeerPort, sockpair[0]));
return std::make_pair(std::move(peer_io), sockpair[1]);
}
static constexpr auto makePeerId(std::string_view sv)
@ -250,7 +244,7 @@ TEST_F(HandshakeTest, incomingPlaintext)
EXPECT_EQ(io, res->io);
EXPECT_TRUE(res->peer_id);
EXPECT_EQ(peer_id, res->peer_id);
EXPECT_EQ(TorrentWeAreSeeding.info_hash, io->torrentHash());
EXPECT_EQ(TorrentWeAreSeeding.info_hash, io->torrent_hash());
evutil_closesocket(sock);
}
@ -276,7 +270,7 @@ TEST_F(HandshakeTest, incomingPlaintextUnknownInfoHash)
EXPECT_TRUE(res->read_anything_from_peer);
EXPECT_EQ(io, res->io);
EXPECT_FALSE(res->peer_id);
EXPECT_EQ(tr_sha1_digest_t{}, io->torrentHash());
EXPECT_EQ(tr_sha1_digest_t{}, io->torrent_hash());
evutil_closesocket(sock);
}
@ -302,8 +296,8 @@ TEST_F(HandshakeTest, outgoingPlaintext)
EXPECT_EQ(io, res->io);
EXPECT_TRUE(res->peer_id);
EXPECT_EQ(peer_id, res->peer_id);
EXPECT_EQ(UbuntuTorrent.info_hash, io->torrentHash());
EXPECT_EQ(tr_sha1_to_string(UbuntuTorrent.info_hash), tr_sha1_to_string(io->torrentHash()));
EXPECT_EQ(UbuntuTorrent.info_hash, io->torrent_hash());
EXPECT_EQ(tr_sha1_to_string(UbuntuTorrent.info_hash), tr_sha1_to_string(io->torrent_hash()));
evutil_closesocket(sock);
}
@ -340,8 +334,8 @@ TEST_F(HandshakeTest, incomingEncrypted)
EXPECT_EQ(io, res->io);
EXPECT_TRUE(res->peer_id);
EXPECT_EQ(ExpectedPeerId, res->peer_id);
EXPECT_EQ(UbuntuTorrent.info_hash, io->torrentHash());
EXPECT_EQ(tr_sha1_to_string(UbuntuTorrent.info_hash), tr_sha1_to_string(io->torrentHash()));
EXPECT_EQ(UbuntuTorrent.info_hash, io->torrent_hash());
EXPECT_EQ(tr_sha1_to_string(UbuntuTorrent.info_hash), tr_sha1_to_string(io->torrent_hash()));
evutil_closesocket(sock);
}
@ -374,7 +368,7 @@ TEST_F(HandshakeTest, incomingEncryptedUnknownInfoHash)
EXPECT_TRUE(res);
EXPECT_FALSE(res->is_connected);
EXPECT_TRUE(res->read_anything_from_peer);
EXPECT_EQ(tr_sha1_digest_t{}, io->torrentHash());
EXPECT_EQ(tr_sha1_digest_t{}, io->torrent_hash());
evutil_closesocket(sock);
}
@ -416,8 +410,8 @@ TEST_F(HandshakeTest, outgoingEncrypted)
EXPECT_EQ(io, res->io);
EXPECT_TRUE(res->peer_id);
EXPECT_EQ(ExpectedPeerId, res->peer_id);
EXPECT_EQ(UbuntuTorrent.info_hash, io->torrentHash());
EXPECT_EQ(tr_sha1_to_string(UbuntuTorrent.info_hash), tr_sha1_to_string(io->torrentHash()));
EXPECT_EQ(UbuntuTorrent.info_hash, io->torrent_hash());
EXPECT_EQ(tr_sha1_to_string(UbuntuTorrent.info_hash), tr_sha1_to_string(io->torrent_hash()));
evutil_closesocket(sock);
}