oxen-io/lokinet
1
1
Fork 0
mirror of https://github.com/oxen-io/lokinet synced 2023-12-14 06:53:00 +01:00
Code Issues Projects Releases Wiki Activity

there do you see tom??

Browse source
This commit is contained in:
dan 2023-03-02 12:00:13 -08:00
parent 0304172378
commit d783beae24
2 changed files with 66 additions and 63 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

99
llarp/quic/connection.cpp
View file

@ -637,6 +637,7 @@ namespace llarp::quic
{
log::debug(logcat, "{} called", __PRETTY_FUNCTION__);
flush_streams();
schedule_retransmit();
log::debug(logcat, "{} finished", __PRETTY_FUNCTION__);
}
@ -766,42 +767,6 @@ namespace llarp::quic
stream.id(),
nwrite,
ndatalen);
if (nwrite == 0) // we are probably done, but maybe congested
{
log::debug(logcat,
"Done stream writing to {} (either stream is congested or we have nothing else to "
"send right now)",
stream.id());
ngtcp2_conn_stat cstat;
ngtcp2_conn_get_conn_stat(conn.get(), &cstat);
log::debug(logcat, "Current unacked bytes in flight: {}, Congestion window: {}",
cstat.bytes_in_flight, cstat.cwnd);
//log::debug(logcat, "Updating pkt tx time at {}" ,__LINE__);
//ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts);
it = strs.erase(it);
continue;
}
if (nwrite > 0)
{
if (ndatalen >= 0)
{
log::debug(logcat, "consumed {} bytes from stream {}", ndatalen, stream.id());
stream.wrote(ndatalen);
}
log::debug(logcat, "Sending stream data packet");
if (!send_packet(nwrite))
return;
log::debug(logcat, "Updating pkt tx time at {}" ,__LINE__);
ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts); // so far always useful
++stream_packets;
std::advance(it, 1);
continue;
}
if (nwrite < 0)
{
@ -844,7 +809,39 @@ namespace llarp::quic
log::warning(logcat, "Error writing non-stream data: {}", ngtcp2_strerror(nwrite));
break;
}
if (ndatalen >= 0)
{
log::debug(logcat, "consumed {} bytes from stream {}", ndatalen, stream.id());
stream.wrote(ndatalen);
}
if (nwrite == 0) // we are probably done, but maybe congested
{
log::debug(logcat,
"Done stream writing to {} (either stream is congested or we have nothing else to "
"send right now)",
stream.id());
ngtcp2_conn_stat cstat;
ngtcp2_conn_get_conn_stat(conn.get(), &cstat);
log::debug(logcat, "Current unacked bytes in flight: {}, Congestion window: {}",
cstat.bytes_in_flight, cstat.cwnd);
//log::debug(logcat, "Updating pkt tx time at {}" ,__LINE__);
//ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts);
it = strs.erase(it);
continue;
}
log::debug(logcat, "Sending stream data packet");
if (!send_packet(nwrite))
return;
log::debug(logcat, "Updating pkt tx time at {}" ,__LINE__);
ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts); // so far always useful
++stream_packets;
std::advance(it, 1);
//it = strs.erase(it);
if (++stream_packets == max_stream_packets)
{
log::debug(logcat, "Max stream packets ({}) reached", max_stream_packets);
@ -852,9 +849,6 @@ namespace llarp::quic
ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts);
return;
}
log::debug(logcat, "Ding!");
it = strs.erase(it);
}
}
@ -927,9 +921,9 @@ namespace llarp::quic
ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts);
}
log::debug(logcat, "Updating pkt tx time at {}" ,__LINE__);
ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts);
schedule_retransmit();
log::debug(logcat, "Exiting flush_streams()");
//ngtcp2_conn_update_pkt_tx_time(conn.get(), *ts);
//schedule_retransmit();
}
void
@ -1163,6 +1157,25 @@ namespace llarp::quic
return str;
}
/*
const std::shared_ptr<Stream>&
Connection::open_stream(Stream::data_callback_t data_cb, Stream::close_callback_t close_cb)
{
std::shared_ptr<Stream> stream{new Stream{
*this, std::move(data_cb), std::move(close_cb), endpoint.default_stream_buffer_size}};
if (int rv = ngtcp2_conn_open_bidi_stream(conn.get(),
&stream->stream_id.id,
std::get<0>(stream->user_data).get());
rv != 0)
throw std::runtime_error{"Stream creation failed: "s + ngtcp2_strerror(rv)};
auto& str = streams[stream->stream_id];
str = std::move(stream);
return str;
}
*/
const std::shared_ptr<Stream>&
Connection::get_stream(StreamID s) const
{

30
llarp/quic/endpoint.cpp
View file

@ -1,5 +1,6 @@
#include "endpoint.hpp"
#include "client.hpp"
#include "ngtcp2/ngtcp2.h"
#include "server.hpp"
#include "uvw/async.h"
#include <llarp/crypto/crypto.hpp>
@ -254,7 +255,7 @@ namespace llarp::quic
log::debug(logcat, "Closing connection {}", conn.base_cid);
ngtcp2_connection_close_error err;
ngtcp2_connection_close_error_set_transport_error(
ngtcp2_connection_close_error_set_transport_error_liberr(
&err,
code,
reinterpret_cast<uint8_t*>(const_cast<char*>(close_reason.data())),
@ -265,10 +266,9 @@ namespace llarp::quic
Path path;
ngtcp2_pkt_info pi;
auto written = ngtcp2_conn_write_connection_close_versioned(
auto written = ngtcp2_conn_write_connection_close(
conn,
path,
NGTCP2_PKT_INFO_VERSION,
&pi,
u8data(conn.conn_buffer),
conn.conn_buffer.size(),
@ -344,27 +344,17 @@ namespace llarp::quic
if (cleanup)
clean_alias_conns();
for (auto it = conns.begin(); it != conns.end(); ++it)
for (auto & it : conns)
{
if (auto* conn_ptr = std::get_if<primary_conn_ptr>(&it->second))
if (auto* conn_ptr = std::get_if<primary_conn_ptr>(&it.second))
{
Connection& conn = **conn_ptr;
auto exp = ngtcp2_conn_get_expiry(conn);
auto expiry = std::chrono::nanoseconds{static_cast<std::chrono::nanoseconds::rep>(exp)};
auto ngtcp2_expiry_delta = std::chrono::duration_cast<std::chrono::milliseconds>(
expiry - get_time().time_since_epoch());
log::debug(logcat, "ngtcp2_conn_get_expiry returned {} with now_ts {} and expiry_delta {}",
exp, now_ts, ngtcp2_expiry_delta.count());
// a bit of buffer on the expiration time in case the last call to
// ngtcp2_conn_get_expiry() returned ~0ms from now and the connection
// hasn't had time to handle it yet. 5ms should do.
if (ngtcp2_expiry_delta.count() > -500)
continue;
log::debug(logcat, "Draining connection {}", it->first);
start_draining(conn);
if (auto rv = ngtcp2_conn_handle_expiry(conn, now_ts); rv != 0)
{
log::warning(logcat, "ngtcp2_conn_handle_expiry returned code {} at {}", rv, __LINE__);
close_connection(conn, ngtcp2_err_infer_quic_transport_error_code(rv));
}
}
}
}