Green-Sky/solanaceae_ngc_ft1
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

after 2 weeks of porting over the ngc_ft1 code to solanaceae and rewriting the highlevel logic

Browse Source 
(29 commits predate this)
This commit is contained in:
Green Sky
2023-08-19 22:37:55 +02:00
commit 89d8d728ff
24 changed files with 4233 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

281
solanaceae/ngc_ext/ngcext.cpp Normal file
View File

@ -0,0 +1,281 @@
#include "./ngcext.hpp"
#include <iostream>
NGCEXTEventProvider::NGCEXTEventProvider(ToxEventProviderI& tep) : _tep(tep) {
_tep.subscribe(this, Tox_Event::TOX_EVENT_GROUP_CUSTOM_PACKET);
_tep.subscribe(this, Tox_Event::TOX_EVENT_GROUP_CUSTOM_PRIVATE_PACKET);
}
#define _DATA_HAVE(x, error) if ((data_size - curser) < (x)) { error; }
bool NGCEXTEventProvider::parse_hs1_request_last_ids(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
return false;
}
bool NGCEXTEventProvider::parse_hs1_response_last_ids(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
return false;
}
bool NGCEXTEventProvider::parse_ft1_request(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool // dont care private
) {
Events::NGCEXT_ft1_request e;
e.group_number = group_number;
e.peer_number = peer_number;
size_t curser = 0;
// - 4 byte (file_kind)
e.file_kind = 0u;
_DATA_HAVE(sizeof(e.file_kind), std::cerr << "NGCEXT: packet too small, missing file_kind\n"; return false)
for (size_t i = 0; i < sizeof(e.file_kind); i++, curser++) {
e.file_kind |= uint32_t(data[curser]) << (i*8);
}
// - X bytes (file_kind dependent id, differnt sizes)
e.file_id = {data+curser, data+curser+(data_size-curser)};
return dispatch(
NGCEXT_Event::FT1_REQUEST,
e
);
}
bool NGCEXTEventProvider::parse_ft1_init(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
if (!_private) {
std::cerr << "NGCEXT: ft1_init cant be public\n";
return false;
}
Events::NGCEXT_ft1_init e;
e.group_number = group_number;
e.peer_number = peer_number;
size_t curser = 0;
// - 4 byte (file_kind)
e.file_kind = 0u;
_DATA_HAVE(sizeof(e.file_kind), std::cerr << "NGCEXT: packet too small, missing file_kind\n"; return false)
for (size_t i = 0; i < sizeof(e.file_kind); i++, curser++) {
e.file_kind |= uint32_t(data[curser]) << (i*8);
}
// - 8 bytes (data size)
e.file_size = 0u;
_DATA_HAVE(sizeof(e.file_size), std::cerr << "NGCEXT: packet too small, missing file_size\n"; return false)
for (size_t i = 0; i < sizeof(e.file_size); i++, curser++) {
e.file_size |= size_t(data[curser]) << (i*8);
}
// - 1 byte (temporary_file_tf_id)
_DATA_HAVE(sizeof(e.transfer_id), std::cerr << "NGCEXT: packet too small, missing transfer_id\n"; return false)
e.transfer_id = data[curser++];
// - X bytes (file_kind dependent id, differnt sizes)
e.file_id = {data+curser, data+curser+(data_size-curser)};
return dispatch(
NGCEXT_Event::FT1_INIT,
e
);
}
bool NGCEXTEventProvider::parse_ft1_init_ack(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
if (!_private) {
std::cerr << "NGCEXT: ft1_init_ack cant be public\n";
return false;
}
Events::NGCEXT_ft1_init_ack e;
e.group_number = group_number;
e.peer_number = peer_number;
size_t curser = 0;
// - 1 byte (temporary_file_tf_id)
_DATA_HAVE(sizeof(e.transfer_id), std::cerr << "NGCEXT: packet too small, missing transfer_id\n"; return false)
e.transfer_id = data[curser++];
return dispatch(
NGCEXT_Event::FT1_INIT_ACK,
e
);
}
bool NGCEXTEventProvider::parse_ft1_data(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
if (!_private) {
std::cerr << "NGCEXT: ft1_data cant be public\n";
return false;
}
Events::NGCEXT_ft1_data e;
e.group_number = group_number;
e.peer_number = peer_number;
size_t curser = 0;
// - 1 byte (temporary_file_tf_id)
_DATA_HAVE(sizeof(e.transfer_id), std::cerr << "NGCEXT: packet too small, missing transfer_id\n"; return false)
e.transfer_id = data[curser++];
// - 2 bytes (sequence_id)
e.sequence_id = 0u;
_DATA_HAVE(sizeof(e.sequence_id), std::cerr << "NGCEXT: packet too small, missing sequence_id\n"; return false)
for (size_t i = 0; i < sizeof(e.sequence_id); i++, curser++) {
e.sequence_id |= uint32_t(data[curser]) << (i*8);
}
// - X bytes (the data fragment)
// (size is implicit)
e.data = {data+curser, data+curser+(data_size-curser)};
return dispatch(
NGCEXT_Event::FT1_DATA,
e
);
}
bool NGCEXTEventProvider::parse_ft1_data_ack(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
if (!_private) {
std::cerr << "NGCEXT: ft1_data_ack cant be public\n";
return false;
}
Events::NGCEXT_ft1_data_ack e;
e.group_number = group_number;
e.peer_number = peer_number;
size_t curser = 0;
// - 1 byte (temporary_file_tf_id)
_DATA_HAVE(sizeof(e.transfer_id), std::cerr << "NGCEXT: packet too small, missing transfer_id\n"; return false)
e.transfer_id = data[curser++];
while (curser < data_size) {
_DATA_HAVE(sizeof(uint16_t), std::cerr << "NGCEXT: packet too small, missing seq_id\n"; return false)
uint16_t seq_id = data[curser++];
seq_id |= data[curser++] << (1*8);
e.sequence_ids.push_back(seq_id);
}
return dispatch(
NGCEXT_Event::FT1_DATA_ACK,
e
);
}
bool NGCEXTEventProvider::parse_ft1_message(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
) {
if (_private) {
std::cerr << "NGCEXT: ft1_message cant be private (yet)\n";
return false;
}
Events::NGCEXT_ft1_message e;
e.group_number = group_number;
e.peer_number = peer_number;
size_t curser = 0;
// - 4 byte (message_id)
e.message_id = 0u;
_DATA_HAVE(sizeof(e.message_id), std::cerr << "NGCEXT: packet too small, missing message_id\n"; return false)
for (size_t i = 0; i < sizeof(e.message_id); i++, curser++) {
e.message_id |= uint32_t(data[curser]) << (i*8);
}
// - 4 byte (file_kind)
e.file_kind = 0u;
_DATA_HAVE(sizeof(e.file_kind), std::cerr << "NGCEXT: packet too small, missing file_kind\n"; return false)
for (size_t i = 0; i < sizeof(e.file_kind); i++, curser++) {
e.file_kind |= uint32_t(data[curser]) << (i*8);
}
// - X bytes (file_kind dependent id, differnt sizes)
e.file_id = {data+curser, data+curser+(data_size-curser)};
return dispatch(
NGCEXT_Event::FT1_MESSAGE,
e
);
}
bool NGCEXTEventProvider::handlePacket(
const uint32_t group_number,
const uint32_t peer_number,
const uint8_t* data,
const size_t data_size,
const bool _private
) {
if (data_size < 1) {
return false; // waht
}
NGCEXT_Event pkg_type = static_cast<NGCEXT_Event>(data[0]);
switch (pkg_type) {
case NGCEXT_Event::HS1_REQUEST_LAST_IDS:
return false;
case NGCEXT_Event::HS1_RESPONSE_LAST_IDS:
return false;
case NGCEXT_Event::FT1_REQUEST:
return parse_ft1_request(group_number, peer_number, data+1, data_size-1, _private);
case NGCEXT_Event::FT1_INIT:
return parse_ft1_init(group_number, peer_number, data+1, data_size-1, _private);
case NGCEXT_Event::FT1_INIT_ACK:
return parse_ft1_init_ack(group_number, peer_number, data+1, data_size-1, _private);
case NGCEXT_Event::FT1_DATA:
return parse_ft1_data(group_number, peer_number, data+1, data_size-1, _private);
case NGCEXT_Event::FT1_DATA_ACK:
return parse_ft1_data_ack(group_number, peer_number, data+1, data_size-1, _private);
case NGCEXT_Event::FT1_MESSAGE:
return parse_ft1_message(group_number, peer_number, data+1, data_size-1, _private);
default:
return false;
}
return false;
}
bool NGCEXTEventProvider::onToxEvent(const Tox_Event_Group_Custom_Packet* e) {
const auto group_number = tox_event_group_custom_packet_get_group_number(e);
const auto peer_number = tox_event_group_custom_packet_get_peer_id(e);
const uint8_t* data = tox_event_group_custom_packet_get_data(e);
const auto data_length = tox_event_group_custom_packet_get_data_length(e);
return handlePacket(group_number, peer_number, data, data_length, false);
}
bool NGCEXTEventProvider::onToxEvent(const Tox_Event_Group_Custom_Private_Packet* e) {
const auto group_number = tox_event_group_custom_private_packet_get_group_number(e);
const auto peer_number = tox_event_group_custom_private_packet_get_peer_id(e);
const uint8_t* data = tox_event_group_custom_private_packet_get_data(e);
const auto data_length = tox_event_group_custom_private_packet_get_data_length(e);
return handlePacket(group_number, peer_number, data, data_length, true);
}

278
solanaceae/ngc_ext/ngcext.hpp Normal file
View File

@ -0,0 +1,278 @@
#pragma once
// solanaceae port of tox_ngc_ext
#include <solanaceae/toxcore/tox_event_interface.hpp>
#include <solanaceae/util/event_provider.hpp>
#include <solanaceae/toxcore/tox_key.hpp>
#include <array>
#include <vector>
namespace Events {
// TODO: implement events as non-owning
struct NGCEXT_hs1_request_last_ids {
uint32_t group_number;
uint32_t peer_number;
// - peer_key bytes (peer key we want to know ids for)
ToxKey peer_key;
// - 1 byte (uint8_t count ids, atleast 1)
uint8_t count_ids;
};
struct NGCEXT_hs1_response_last_ids {
uint32_t group_number;
uint32_t peer_number;
// respond to a request with 0 or more message ids, sorted by newest first
// - peer_key bytes (the msg_ids are from)
ToxKey peer_key;
// - 1 byte (uint8_t count ids, can be 0)
uint8_t count_ids;
// - array [
// - msg_id bytes (the message id)
// - ]
std::vector<uint32_t> msg_ids;
};
struct NGCEXT_ft1_request {
uint32_t group_number;
uint32_t peer_number;
// request the other side to initiate a FT
// - 4 byte (file_kind)
uint32_t file_kind;
// - X bytes (file_kind dependent id, differnt sizes)
std::vector<uint8_t> file_id;
};
struct NGCEXT_ft1_init {
uint32_t group_number;
uint32_t peer_number;
// tell the other side you want to start a FT
// - 4 byte (file_kind)
uint32_t file_kind;
// - 8 bytes (data size)
uint64_t file_size;
// - 1 byte (temporary_file_tf_id, for this peer only, technically just a prefix to distinguish between simultainious fts)
uint8_t transfer_id;
// - X bytes (file_kind dependent id, differnt sizes)
std::vector<uint8_t> file_id;
// TODO: max supported lossy packet size
};
struct NGCEXT_ft1_init_ack {
uint32_t group_number;
uint32_t peer_number;
// - 1 byte (transfer_id)
uint8_t transfer_id;
// TODO: max supported lossy packet size
};
struct NGCEXT_ft1_data {
uint32_t group_number;
uint32_t peer_number;
// data fragment
// - 1 byte (temporary_file_tf_id)
uint8_t transfer_id;
// - 2 bytes (sequece id)
uint16_t sequence_id;
// - X bytes (the data fragment)
// (size is implicit)
std::vector<uint8_t> data;
};
struct NGCEXT_ft1_data_ack {
uint32_t group_number;
uint32_t peer_number;
// - 1 byte (temporary_file_tf_id)
uint8_t transfer_id;
// - array [ (of sequece ids)
// - 2 bytes (sequece id)
// - ]
std::vector<uint16_t> sequence_ids;
};
struct NGCEXT_ft1_message {
uint32_t group_number;
uint32_t peer_number;
// - 4 byte (message_id)
uint32_t message_id;
// request the other side to initiate a FT
// - 4 byte (file_kind)
uint32_t file_kind;
// - X bytes (file_kind dependent id, differnt sizes)
std::vector<uint8_t> file_id;
};
} // Events
enum class NGCEXT_Event : uint8_t {
//TODO: make it possible to go further back
// request last (few) message_ids for a peer
// - peer_key bytes (peer key we want to know ids for)
// - 1 byte (uint8_t count ids, atleast 1)
HS1_REQUEST_LAST_IDS = 0x80 | 1u,
// respond to a request with 0 or more message ids, sorted by newest first
// - peer_key bytes (the msg_ids are from)
// - 1 byte (uint8_t count ids, can be 0)
// - array [
// - msg_id bytes (the message id)
// - ]
HS1_RESPONSE_LAST_IDS,
// request the other side to initiate a FT
// - 4 byte (file_kind)
// - X bytes (file_kind dependent id, differnt sizes)
FT1_REQUEST = 0x80 | 8u,
// TODO: request result negative, speed up not found
// tell the other side you want to start a FT
// TODO: might use id layer instead. with it, it would look similar to friends_ft
// - 4 byte (file_kind)
// - 8 bytes (data size, can be 0 if unknown, BUT files have to be atleast 1 byte)
// - 1 byte (temporary_file_tf_id, for this peer only, technically just a prefix to distinguish between simultainious fts)
// - X bytes (file_kind dependent id, differnt sizes)
FT1_INIT,
// acknowlage init (like an accept)
// like tox ft control continue
// - 1 byte (transfer_id)
FT1_INIT_ACK,
// TODO: init deny, speed up non acceptance
// data fragment
// - 1 byte (temporary_file_tf_id)
// - 2 bytes (sequece id)
// - X bytes (the data fragment)
// (size is implicit)
FT1_DATA,
// acknowlage data fragments
// TODO: last 3 should be sufficient, 5 should be generous
// - 1 byte (temporary_file_tf_id)
// // this is implicit (pkg size)- 1 byte (number of sequence ids to ack, this kind of depends on window size)
// - array [ (of sequece ids)
// - 2 bytes (sequece id)
// - ]
FT1_DATA_ACK,
// send file as message
// basically the opposite of request
// contains file_kind and file_id (and timestamp?)
// - 4 byte (message_id)
// - 4 byte (file_kind)
// - X bytes (file_kind dependent id, differnt sizes)
FT1_MESSAGE,
MAX
};
struct NGCEXTEventI {
using enumType = NGCEXT_Event;
virtual bool onEvent(const Events::NGCEXT_hs1_request_last_ids&) { return false; }
virtual bool onEvent(const Events::NGCEXT_hs1_response_last_ids&) { return false; }
virtual bool onEvent(const Events::NGCEXT_ft1_request&) { return false; }
virtual bool onEvent(const Events::NGCEXT_ft1_init&) { return false; }
virtual bool onEvent(const Events::NGCEXT_ft1_init_ack&) { return false; }
virtual bool onEvent(const Events::NGCEXT_ft1_data&) { return false; }
virtual bool onEvent(const Events::NGCEXT_ft1_data_ack&) { return false; }
virtual bool onEvent(const Events::NGCEXT_ft1_message&) { return false; }
};
using NGCEXTEventProviderI = EventProviderI<NGCEXTEventI>;
class NGCEXTEventProvider : public ToxEventI, public NGCEXTEventProviderI {
ToxEventProviderI& _tep;
public:
NGCEXTEventProvider(ToxEventProviderI& tep/*, ToxI& t*/);
protected:
bool parse_hs1_request_last_ids(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_hs1_response_last_ids(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_ft1_request(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_ft1_init(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_ft1_init_ack(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_ft1_data(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_ft1_data_ack(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool parse_ft1_message(
uint32_t group_number, uint32_t peer_number,
const uint8_t* data, size_t data_size,
bool _private
);
bool handlePacket(
const uint32_t group_number,
const uint32_t peer_number,
const uint8_t* data,
const size_t data_size,
const bool _private
);
protected:
bool onToxEvent(const Tox_Event_Group_Custom_Packet* e) override;
bool onToxEvent(const Tox_Event_Group_Custom_Private_Packet* e) override;
};

250
solanaceae/ngc_ft1/ledbat.cpp Normal file
View File

@ -0,0 +1,250 @@
#include "./ledbat.hpp"
#include <algorithm>
#include <chrono>
#include <cmath>
#include <deque>
#include <cstdint>
#include <cassert>
#include <iomanip>
#include <iostream>
#include <limits>
// https://youtu.be/0HRwNSA-JYM
inline constexpr bool PLOTTING = false;
LEDBAT::LEDBAT(size_t maximum_segment_data_size) : MAXIMUM_SEGMENT_DATA_SIZE(maximum_segment_data_size) {
_time_start_offset = clock::now();
}
size_t LEDBAT::canSend(void) const {
if (_in_flight.empty()) {
return MAXIMUM_SEGMENT_DATA_SIZE;
}
const int64_t cspace = _cwnd - _in_flight_bytes;
if (cspace < MAXIMUM_SEGMENT_DATA_SIZE) {
return 0u;
}
const int64_t fspace = _fwnd - _in_flight_bytes;
if (fspace < MAXIMUM_SEGMENT_DATA_SIZE) {
return 0u;
}
size_t space = std::ceil(std::min<float>(cspace, fspace) / MAXIMUM_SEGMENT_DATA_SIZE) * MAXIMUM_SEGMENT_DATA_SIZE;
return space;
}
std::vector<LEDBAT::SeqIDType> LEDBAT::getTimeouts(void) const {
std::vector<LEDBAT::SeqIDType> list;
// after 2 delays we trigger timeout
const auto now_adjusted = getTimeNow() - getCurrentDelay()*2.f;
for (const auto& [seq, time_stamp, size] : _in_flight) {
if (now_adjusted > time_stamp) {
list.push_back(seq);
}
}
return list;
}
void LEDBAT::onSent(SeqIDType seq, size_t data_size) {
if (true) {
for (const auto& it : _in_flight) {
assert(std::get<0>(it) != seq);
}
}
_in_flight.push_back({seq, getTimeNow(), data_size + SEGMENT_OVERHEAD});
_in_flight_bytes += data_size + SEGMENT_OVERHEAD;
_recently_sent_bytes += data_size + SEGMENT_OVERHEAD;
}
void LEDBAT::onAck(std::vector<SeqIDType> seqs) {
// only take the smallest value
float most_recent {-std::numeric_limits<float>::infinity()};
int64_t acked_data {0};
const auto now {getTimeNow()};
for (const auto& seq : seqs) {
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq](const auto& v) -> bool {
return std::get<0>(v) == seq;
});
if (it == _in_flight.end()) {
continue; // not found, ignore
} else {
addRTT(now - std::get<1>(*it));
// TODO: remove
most_recent = std::max(most_recent, std::get<1>(*it));
_in_flight_bytes -= std::get<2>(*it);
_recently_acked_data += std::get<2>(*it);
assert(_in_flight_bytes >= 0); // TODO: this triggers
_in_flight.erase(it);
}
}
if (most_recent == -std::numeric_limits<float>::infinity()) {
return; // not found, ignore
}
updateWindows();
}
void LEDBAT::onLoss(SeqIDType seq, bool discard) {
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq](const auto& v) -> bool {
assert(!std::isnan(std::get<1>(v)));
return std::get<0>(v) == seq;
});
if (it == _in_flight.end()) {
// error
return; // not found, ignore ??
}
_recently_lost_data = true;
// at most once per rtt?
if (PLOTTING) {
std::cerr << "CCA: onLoss: TIME: " << getTimeNow() << "\n";
}
// TODO: "if data lost is not to be retransmitted"
if (discard) {
_in_flight_bytes -= std::get<2>(*it);
assert(_in_flight_bytes >= 0);
_in_flight.erase(it);
}
// TODO: reset timestamp?
updateWindows();
}
float LEDBAT::getCurrentDelay(void) const {
float sum {0.f};
size_t count {0};
for (size_t i = 0; i < _tmp_rtt_buffer.size(); i++) {
//sum += _tmp_rtt_buffer.at(_tmp_rtt_buffer.size()-(1+i));
sum += _tmp_rtt_buffer.at(i);
count++;
}
if (count) {
return sum / count;
} else {
return std::numeric_limits<float>::infinity();
}
}
void LEDBAT::addRTT(float new_delay) {
auto now = getTimeNow();
_base_delay = std::min(_base_delay, new_delay);
// TODO: use fixed size instead? allocations can ruin perf
_rtt_buffer.push_back({now, new_delay});
_tmp_rtt_buffer.push_front(new_delay);
// HACKY
if (_tmp_rtt_buffer.size() > current_delay_filter_window) {
_tmp_rtt_buffer.resize(current_delay_filter_window);
}
// is it 1 minute yet
if (now - _rtt_buffer.front().first >= 30.f) {
float new_section_minimum = new_delay;
for (const auto it : _rtt_buffer) {
new_section_minimum = std::min(it.second, new_section_minimum);
}
_rtt_buffer_minutes.push_back(new_section_minimum);
_rtt_buffer.clear();
if (_rtt_buffer_minutes.size() > 20) {
_rtt_buffer_minutes.pop_front();
}
_base_delay = std::numeric_limits<float>::infinity();
for (const float it : _rtt_buffer_minutes) {
_base_delay = std::min(_base_delay, it);
}
}
}
void LEDBAT::updateWindows(void) {
const auto now {getTimeNow()};
const float current_delay {getCurrentDelay()};
if (now - _last_cwnd >= current_delay) {
const float queuing_delay {current_delay - _base_delay};
_fwnd = max_byterate_allowed * current_delay;
_fwnd *= 1.3f; // try do balance conservative algo a bit, current_delay
float gain {1.f / std::min(16.f, std::ceil(2.f*target_delay/_base_delay))};
//gain *= 400.f; // from packets to bytes ~
gain *= _recently_acked_data/5.f; // from packets to bytes ~
//gain *= 0.1f;
if (_recently_lost_data) {
_cwnd = std::clamp(
_cwnd / 2.f,
//_cwnd / 1.6f,
2.f * MAXIMUM_SEGMENT_SIZE,
_cwnd
);
} else {
// LEDBAT++ (the Rethinking the LEDBAT Protocol paper)
// "Multiplicative decrease"
const float constant {2.f}; // spec recs 1
if (queuing_delay < target_delay) {
_cwnd = std::min(
_cwnd + gain,
_fwnd
);
} else if (queuing_delay > target_delay) {
_cwnd = std::clamp(
_cwnd + std::max(
gain - constant * _cwnd * (queuing_delay / target_delay - 1.f),
-_cwnd/2.f // at most halve
),
// never drop below 2 "packets" in flight
2.f * MAXIMUM_SEGMENT_SIZE,
// cap rate
_fwnd
);
} // no else, we on point. very unlikely with float
}
if (PLOTTING) { // plotting
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " cwnd: " << _cwnd << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " fwnd: " << _fwnd << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " current_delay: " << current_delay << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " base_delay: " << _base_delay << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " gain: " << gain << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " speed: " << (_recently_sent_bytes / (now - _last_cwnd)) / (1024*1024) << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " in_flight_bytes: " << _in_flight_bytes << "\n";
}
_last_cwnd = now;
_recently_acked_data = 0;
_recently_lost_data = false;
_recently_sent_bytes = 0;
}
}

122
solanaceae/ngc_ft1/ledbat.hpp Normal file
View File

@ -0,0 +1,122 @@
#pragma once
#include <chrono>
#include <deque>
#include <vector>
#include <cstdint>
// LEDBAT: https://www.rfc-editor.org/rfc/rfc6817
// LEDBAT++: https://www.ietf.org/archive/id/draft-irtf-iccrg-ledbat-plus-plus-01.txt
// LEDBAT++ implementation
struct LEDBAT {
public: // config
using SeqIDType = std::pair<uint8_t, uint16_t>; // tf_id, seq_id
static constexpr size_t IPV4_HEADER_SIZE {20};
static constexpr size_t IPV6_HEADER_SIZE {40}; // bru
static constexpr size_t UDP_HEADER_SIZE {8};
// TODO: tcp AND IPv6 will be different
static constexpr size_t SEGMENT_OVERHEAD {
4+ // ft overhead
46+ // tox?
UDP_HEADER_SIZE+
IPV4_HEADER_SIZE
};
// TODO: make configurable, set with tox ngc lossy packet size
//const size_t MAXIMUM_SEGMENT_DATA_SIZE {1000-4};
const size_t MAXIMUM_SEGMENT_DATA_SIZE {500-4};
//static constexpr size_t maximum_segment_size {496 + segment_overhead}; // tox 500 - 4 from ft
const size_t MAXIMUM_SEGMENT_SIZE {MAXIMUM_SEGMENT_DATA_SIZE + SEGMENT_OVERHEAD}; // tox 500 - 4 from ft
//static_assert(maximum_segment_size == 574); // mesured in wireshark
// ledbat++ says 60ms, we might need other values if relayed
//const float target_delay {0.060f};
const float target_delay {0.030f};
//const float target_delay {0.120f}; // 2x if relayed?
// TODO: use a factor for multiple of rtt
static constexpr size_t current_delay_filter_window {16*4};
//static constexpr size_t rtt_buffer_size_max {2000};
float max_byterate_allowed {10*1024*1024}; // 10MiB/s
public:
LEDBAT(size_t maximum_segment_data_size);
// return the current believed window in bytes of how much data can be inflight,
// without overstepping the delay requirement
float getCWnD(void) const {
return _cwnd;
}
// TODO: api for how much data we should send
// take time since last sent into account
// respect max_byterate_allowed
size_t canSend(void) const;
// get the list of timed out seq_ids
std::vector<SeqIDType> getTimeouts(void) const;
public: // callbacks
// data size is without overhead
void onSent(SeqIDType seq, size_t data_size);
void onAck(std::vector<SeqIDType> seqs);
// if discard, not resent, not inflight
void onLoss(SeqIDType seq, bool discard);
private:
using clock = std::chrono::steady_clock;
// make values relative to algo start for readability (and precision)
// get timestamp in seconds
float getTimeNow(void) const {
return std::chrono::duration<float>{clock::now() - _time_start_offset}.count();
}
// moving avg over the last few delay samples
// VERY sensitive to bundling acks
float getCurrentDelay(void) const;
void addRTT(float new_delay);
void updateWindows(void);
private: // state
//float _cto {2.f}; // congestion timeout value in seconds
float _cwnd {2.f * MAXIMUM_SEGMENT_SIZE}; // in bytes
float _base_delay {2.f}; // lowest mesured delay in _rtt_buffer in seconds
float _last_cwnd {0.f}; // timepoint of last cwnd correction
int64_t _recently_acked_data {0}; // reset on _last_cwnd
bool _recently_lost_data {false};
int64_t _recently_sent_bytes {0};
// initialize to low value, will get corrected very fast
float _fwnd {0.01f * max_byterate_allowed}; // in bytes
// ssthresh
// spec recomends 10min
// TODO: optimize and devide into spans of 1min (spec recom)
std::deque<float> _tmp_rtt_buffer;
std::deque<std::pair<float, float>> _rtt_buffer; // timepoint, delay
std::deque<float> _rtt_buffer_minutes;
// list of sequence ids and timestamps of when they where sent
std::deque<std::tuple<SeqIDType, float, size_t>> _in_flight;
int64_t _in_flight_bytes {0};
private: // helper
clock::time_point _time_start_offset;
};

719
solanaceae/ngc_ft1/ngcft1.cpp Normal file
View File

@ -0,0 +1,719 @@
#include "./ngcft1.hpp"
#include <solanaceae/toxcore/utils.hpp>
#include <sodium.h>
#include <iostream>
#include <set>
#include <algorithm>
#include <cassert>
#include <vector>
bool NGCFT1::sendPKG_FT1_REQUEST(
uint32_t group_number, uint32_t peer_number,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size
) {
// - 1 byte packet id
// - 4 byte file_kind
// - X bytes file_id
std::vector<uint8_t> pkg;
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::FT1_REQUEST));
for (size_t i = 0; i < sizeof(file_kind); i++) {
pkg.push_back((file_kind>>(i*8)) & 0xff);
}
for (size_t i = 0; i < file_id_size; i++) {
pkg.push_back(file_id[i]);
}
// lossless
return _t.toxGroupSendCustomPrivatePacket(group_number, peer_number, true, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PRIVATE_PACKET_OK;
}
bool NGCFT1::sendPKG_FT1_INIT(
uint32_t group_number, uint32_t peer_number,
uint32_t file_kind,
uint64_t file_size,
uint8_t transfer_id,
const uint8_t* file_id, size_t file_id_size
) {
// - 1 byte packet id
// - 4 byte (file_kind)
// - 8 bytes (data size)
// - 1 byte (temporary_file_tf_id, for this peer only, technically just a prefix to distinguish between simultainious fts)
// - X bytes (file_kind dependent id, differnt sizes)
std::vector<uint8_t> pkg;
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::FT1_INIT));
for (size_t i = 0; i < sizeof(file_kind); i++) {
pkg.push_back((file_kind>>(i*8)) & 0xff);
}
for (size_t i = 0; i < sizeof(file_size); i++) {
pkg.push_back((file_size>>(i*8)) & 0xff);
}
pkg.push_back(transfer_id);
for (size_t i = 0; i < file_id_size; i++) {
pkg.push_back(file_id[i]);
}
// lossless
return _t.toxGroupSendCustomPrivatePacket(group_number, peer_number, true, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PRIVATE_PACKET_OK;
}
bool NGCFT1::sendPKG_FT1_INIT_ACK(
uint32_t group_number, uint32_t peer_number,
uint8_t transfer_id
) {
// - 1 byte packet id
// - 1 byte transfer_id
std::vector<uint8_t> pkg;
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::FT1_INIT_ACK));
pkg.push_back(transfer_id);
// lossless
return _t.toxGroupSendCustomPrivatePacket(group_number, peer_number, true, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PRIVATE_PACKET_OK;
}
bool NGCFT1::sendPKG_FT1_DATA(
uint32_t group_number, uint32_t peer_number,
uint8_t transfer_id,
uint16_t sequence_id,
const uint8_t* data, size_t data_size
) {
assert(data_size > 0);
// TODO
// check header_size+data_size <= max pkg size
std::vector<uint8_t> pkg;
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::FT1_DATA));
pkg.push_back(transfer_id);
pkg.push_back(sequence_id & 0xff);
pkg.push_back((sequence_id >> (1*8)) & 0xff);
// TODO: optimize
for (size_t i = 0; i < data_size; i++) {
pkg.push_back(data[i]);
}
// lossy
return _t.toxGroupSendCustomPrivatePacket(group_number, peer_number, false, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PRIVATE_PACKET_OK;
}
bool NGCFT1::sendPKG_FT1_DATA_ACK(
uint32_t group_number, uint32_t peer_number,
uint8_t transfer_id,
const uint16_t* seq_ids, size_t seq_ids_size
) {
std::vector<uint8_t> pkg;
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::FT1_DATA_ACK));
pkg.push_back(transfer_id);
// TODO: optimize
for (size_t i = 0; i < seq_ids_size; i++) {
pkg.push_back(seq_ids[i] & 0xff);
pkg.push_back((seq_ids[i] >> (1*8)) & 0xff);
}
// lossy
return _t.toxGroupSendCustomPrivatePacket(group_number, peer_number, false, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PRIVATE_PACKET_OK;
}
bool NGCFT1::sendPKG_FT1_MESSAGE(
uint32_t group_number,
uint32_t message_id,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size
) {
std::vector<uint8_t> pkg;
pkg.push_back(static_cast<uint8_t>(NGCEXT_Event::FT1_MESSAGE));
for (size_t i = 0; i < sizeof(message_id); i++) {
pkg.push_back((message_id>>(i*8)) & 0xff);
}
for (size_t i = 0; i < sizeof(file_kind); i++) {
pkg.push_back((file_kind>>(i*8)) & 0xff);
}
for (size_t i = 0; i < file_id_size; i++) {
pkg.push_back(file_id[i]);
}
// lossless
return _t.toxGroupSendCustomPacket(group_number, true, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PACKET_OK;
}
void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<LEDBAT::SeqIDType>& timeouts_set) {
auto& tf_opt = peer.send_transfers.at(idx);
assert(tf_opt.has_value());
auto& tf = tf_opt.value();
tf.time_since_activity += time_delta;
switch (tf.state) {
using State = Group::Peer::SendTransfer::State;
case State::INIT_SENT:
if (tf.time_since_activity >= init_retry_timeout_after) {
if (tf.inits_sent >= 3) {
// delete, timed out 3 times
std::cerr << "NGCFT1 warning: ft init timed out, deleting\n";
dispatch(
NGCFT1_Event::send_done,
Events::NGCFT1_send_done{
group_number, peer_number,
static_cast<uint8_t>(idx),
}
);
tf_opt.reset();
} else {
// timed out, resend
std::cerr << "NGCFT1 warning: ft init timed out, resending\n";
sendPKG_FT1_INIT(group_number, peer_number, tf.file_kind, tf.file_size, idx, tf.file_id.data(), tf.file_id.size());
tf.inits_sent++;
tf.time_since_activity = 0.f;
}
}
//break;
return;
case State::SENDING: {
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
// no ack after 5 sec -> resend
//if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) {
if (timeouts_set.count({idx, id})) {
// TODO: can fail
sendPKG_FT1_DATA(group_number, peer_number, idx, id, data.data(), data.size());
peer.cca->onLoss({idx, id}, false);
time_since_activity = 0.f;
timeouts_set.erase({idx, id});
}
});
if (tf.time_since_activity >= sending_give_up_after) {
// no ack after 30sec, close ft
std::cerr << "NGCFT1 warning: sending ft in progress timed out, deleting\n";
dispatch(
NGCFT1_Event::send_done,
Events::NGCFT1_send_done{
group_number, peer_number,
static_cast<uint8_t>(idx),
}
);
// clean up cca
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
peer.cca->onLoss({idx, id}, true);
timeouts_set.erase({idx, id});
});
tf_opt.reset();
//continue; // dangerous control flow
return;
}
// if chunks in flight < window size (2)
//while (tf.ssb.size() < ngc_ft1_ctx->options.packet_window_size) {
int64_t can_packet_size {static_cast<int64_t>(peer.cca->canSend())};
//if (can_packet_size) {
//std::cerr << "FT: can_packet_size: " << can_packet_size;
//}
size_t count {0};
while (can_packet_size > 0 && tf.file_size > 0) {
std::vector<uint8_t> new_data;
// TODO: parameterize packet size? -> only if JF increases lossy packet size >:)
//size_t chunk_size = std::min<size_t>(496u, tf.file_size - tf.file_size_current);
//size_t chunk_size = std::min<size_t>(can_packet_size, tf.file_size - tf.file_size_current);
size_t chunk_size = std::min<size_t>({
//496u,
//996u,
peer.cca->MAXIMUM_SEGMENT_DATA_SIZE,
static_cast<size_t>(can_packet_size),
tf.file_size - tf.file_size_current
});
if (chunk_size == 0) {
tf.state = State::FINISHING;
break; // we done
}
new_data.resize(chunk_size);
//ngc_ft1_ctx->cb_send_data[tf.file_kind](
//tox,
//group_number, peer_number,
//idx,
//tf.file_size_current,
//new_data.data(), new_data.size(),
//ngc_ft1_ctx->ud_send_data.count(tf.file_kind) ? ngc_ft1_ctx->ud_send_data.at(tf.file_kind) : nullptr
//);
assert(idx <= 0xffu);
// TODO: check return value
dispatch(
NGCFT1_Event::send_data,
Events::NGCFT1_send_data{
group_number, peer_number,
static_cast<uint8_t>(idx),
tf.file_size_current,
new_data.data(), new_data.size(),
}
);
uint16_t seq_id = tf.ssb.add(std::move(new_data));
sendPKG_FT1_DATA(group_number, peer_number, idx, seq_id, tf.ssb.entries.at(seq_id).data.data(), tf.ssb.entries.at(seq_id).data.size());
peer.cca->onSent({idx, seq_id}, chunk_size);
#if defined(EXTRA_LOGGING) && EXTRA_LOGGING == 1
fprintf(stderr, "FT: sent data size: %ld (seq %d)\n", chunk_size, seq_id);
#endif
tf.file_size_current += chunk_size;
can_packet_size -= chunk_size;
count++;
}
//if (count) {
//std::cerr << " split over " << count << "\n";
//}
}
break;
case State::FINISHING: // we still have unacked packets
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
// no ack after 5 sec -> resend
//if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) {
if (timeouts_set.count({idx, id})) {
sendPKG_FT1_DATA(group_number, peer_number, idx, id, data.data(), data.size());
peer.cca->onLoss({idx, id}, false);
time_since_activity = 0.f;
timeouts_set.erase({idx, id});
}
});
if (tf.time_since_activity >= sending_give_up_after) {
// no ack after 30sec, close ft
// TODO: notify app
std::cerr << "NGCFT1 warning: sending ft finishing timed out, deleting\n";
// clean up cca
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
peer.cca->onLoss({idx, id}, true);
timeouts_set.erase({idx, id});
});
tf_opt.reset();
}
break;
default: // invalid state, delete
std::cerr << "NGCFT1 error: ft in invalid state, deleting\n";
tf_opt.reset();
//continue;
return;
}
}
void NGCFT1::iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer) {
auto timeouts = peer.cca->getTimeouts();
std::set<LEDBAT::SeqIDType> timeouts_set{timeouts.cbegin(), timeouts.cend()};
for (size_t idx = 0; idx < peer.send_transfers.size(); idx++) {
if (peer.send_transfers.at(idx).has_value()) {
updateSendTransfer(time_delta, group_number, peer_number, peer, idx, timeouts_set);
}
}
// TODO: receiving tranfers?
}
NGCFT1::NGCFT1(
ToxI& t,
ToxEventProviderI& tep,
NGCEXTEventProviderI& neep
) : _t(t), _tep(tep), _neep(neep)
{
_neep.subscribe(this, NGCEXT_Event::FT1_REQUEST);
_neep.subscribe(this, NGCEXT_Event::FT1_INIT);
_neep.subscribe(this, NGCEXT_Event::FT1_INIT_ACK);
_neep.subscribe(this, NGCEXT_Event::FT1_DATA);
_neep.subscribe(this, NGCEXT_Event::FT1_DATA_ACK);
_neep.subscribe(this, NGCEXT_Event::FT1_MESSAGE);
_tep.subscribe(this, Tox_Event::TOX_EVENT_GROUP_PEER_EXIT);
}
void NGCFT1::iterate(float time_delta) {
for (auto& [group_number, group] : groups) {
for (auto& [peer_number, peer] : group.peers) {
iteratePeer(time_delta, group_number, peer_number, peer);
}
}
}
void NGCFT1::NGC_FT1_send_request_private(
uint32_t group_number, uint32_t peer_number,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size
) {
// TODO: error check
sendPKG_FT1_REQUEST(group_number, peer_number, file_kind, file_id, file_id_size);
}
bool NGCFT1::NGC_FT1_send_init_private(
uint32_t group_number, uint32_t peer_number,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size,
size_t file_size,
uint8_t* transfer_id
) {
if (std::get<0>(_t.toxGroupPeerGetConnectionStatus(group_number, peer_number)).value_or(TOX_CONNECTION_NONE) == TOX_CONNECTION_NONE) {
std::cerr << "NGCFT1 error: cant init ft, peer offline\n";
return false;
}
auto& peer = groups[group_number].peers[peer_number];
// allocate transfer_id
size_t idx = peer.next_send_transfer_idx;
peer.next_send_transfer_idx = (peer.next_send_transfer_idx + 1) % 256;
{ // TODO: extract
size_t i = idx;
bool found = false;
do {
if (!peer.send_transfers[i].has_value()) {
// free slot
idx = i;
found = true;
break;
}
i = (i + 1) % 256;
} while (i != idx);
if (!found) {
std::cerr << "NGCFT1 error: cant init ft, no free transfer slot\n";
return false;
}
}
// TODO: check return value
sendPKG_FT1_INIT(group_number, peer_number, file_kind, file_size, idx, file_id, file_id_size);
peer.send_transfers[idx] = Group::Peer::SendTransfer{
file_kind,
std::vector(file_id, file_id+file_id_size),
Group::Peer::SendTransfer::State::INIT_SENT,
1,
0.f,
file_size,
0,
{}, // ssb
};
if (transfer_id != nullptr) {
*transfer_id = idx;
}
return true;
}
bool NGCFT1::NGC_FT1_send_message_public(
uint32_t group_number,
uint32_t& message_id,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size
) {
// create msg_id
message_id = randombytes_random();
// TODO: check return value
return sendPKG_FT1_MESSAGE(group_number, message_id, file_kind, file_id, file_id_size);
}
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_request& e) {
//#if !NDEBUG
std::cout << "NGCFT1: FT1_REQUEST fk:" << e.file_kind << " [" << bin2hex(e.file_id) << "]\n";
//#endif
// .... just rethrow??
// TODO: dont
return dispatch(
NGCFT1_Event::recv_request,
Events::NGCFT1_recv_request{
e.group_number, e.peer_number,
static_cast<NGCFT1_file_kind>(e.file_kind),
e.file_id.data(), e.file_id.size()
}
);
}
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_init& e) {
//#if !NDEBUG
std::cout << "NGCFT1: FT1_INIT fk:" << e.file_kind << " fs:" << e.file_size << " tid:" << int(e.transfer_id) << " [" << bin2hex(e.file_id) << "]\n";
//#endif
bool accept = false;
dispatch(
NGCFT1_Event::recv_init,
Events::NGCFT1_recv_init{
e.group_number, e.peer_number,
static_cast<NGCFT1_file_kind>(e.file_kind),
e.file_id.data(), e.file_id.size(),
e.transfer_id,
e.file_size,
accept
}
);
if (!accept) {
std::cout << "NGCFT1: rejected init\n";
return true; // return true?
}
sendPKG_FT1_INIT_ACK(e.group_number, e.peer_number, e.transfer_id);
std::cout << "NGCFT1: accepted init\n";
auto& peer = groups[e.group_number].peers[e.peer_number];
if (peer.recv_transfers[e.transfer_id].has_value()) {
std::cerr << "NGCFT1 warning: overwriting existing recv_transfer " << int(e.transfer_id) << "\n";
}
peer.recv_transfers[e.transfer_id] = Group::Peer::RecvTransfer{
e.file_kind,
e.file_id,
Group::Peer::RecvTransfer::State::INITED,
e.file_size,
0u,
{} // rsb
};
return true;
}
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_init_ack& e) {
//#if !NDEBUG
std::cout << "NGCFT1: FT1_INIT_ACK\n";
//#endif
// we now should start sending data
if (!groups.count(e.group_number)) {
std::cerr << "NGCFT1 warning: init_ack for unknown group\n";
return true;
}
Group::Peer& peer = groups[e.group_number].peers[e.peer_number];
if (!peer.send_transfers[e.transfer_id].has_value()) {
std::cerr << "NGCFT1 warning: init_ack for unknown transfer\n";
return true;
}
Group::Peer::SendTransfer& transfer = peer.send_transfers[e.transfer_id].value();
using State = Group::Peer::SendTransfer::State;
if (transfer.state != State::INIT_SENT) {
std::cerr << "NGCFT1 error: inti_ack but not in INIT_SENT state\n";
return true;
}
// iterate will now call NGC_FT1_send_data_cb
transfer.state = State::SENDING;
transfer.time_since_activity = 0.f;
return true;
}
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data& e) {
#if !NDEBUG
std::cout << "NGCFT1: FT1_DATA\n";
#endif
if (e.data.empty()) {
std::cerr << "NGCFT1 error: data of size 0!\n";
return true;
}
if (!groups.count(e.group_number)) {
std::cerr << "NGCFT1 warning: data for unknown group\n";
return true;
}
Group::Peer& peer = groups[e.group_number].peers[e.peer_number];
if (!peer.recv_transfers[e.transfer_id].has_value()) {
std::cerr << "NGCFT1 warning: data for unknown transfer\n";
return true;
}
auto& transfer = peer.recv_transfers[e.transfer_id].value();
// do reassembly, ignore dups
transfer.rsb.add(e.sequence_id, std::vector<uint8_t>(e.data)); // TODO: ugly explicit copy for what should just be a move
// loop for chunks without holes
while (transfer.rsb.canPop()) {
auto data = transfer.rsb.pop();
// TODO: check return value
dispatch(
NGCFT1_Event::recv_data,
Events::NGCFT1_recv_data{
e.group_number, e.peer_number,
e.transfer_id,
transfer.file_size_current,
data.data(), data.size()
}
);
transfer.file_size_current += data.size();
}
// send acks
std::vector<uint16_t> ack_seq_ids(transfer.rsb.ack_seq_ids.cbegin(), transfer.rsb.ack_seq_ids.cend());
// TODO: check if this caps at max acks
if (!ack_seq_ids.empty()) {
// TODO: check return value
sendPKG_FT1_DATA_ACK(e.group_number, e.peer_number, e.transfer_id, ack_seq_ids.data(), ack_seq_ids.size());
}
if (transfer.file_size_current == transfer.file_size) {
// TODO: set all data received, and clean up
//transfer.state = Group::Peer::RecvTransfer::State::RECV;
dispatch(
NGCFT1_Event::recv_done,
Events::NGCFT1_recv_done{
e.group_number, e.peer_number,
e.transfer_id
}
);
}
return true;
}
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data_ack& e) {
#if !NDEBUG
std::cout << "NGCFT1: FT1_DATA_ACK\n";
#endif
if (!groups.count(e.group_number)) {
std::cerr << "NGCFT1 warning: data_ack for unknown group\n";
return true;
}
Group::Peer& peer = groups[e.group_number].peers[e.peer_number];
if (!peer.send_transfers[e.transfer_id].has_value()) {
std::cerr << "NGCFT1 warning: data_ack for unknown transfer\n";
return true;
}
Group::Peer::SendTransfer& transfer = peer.send_transfers[e.transfer_id].value();
using State = Group::Peer::SendTransfer::State;
if (transfer.state != State::SENDING && transfer.state != State::FINISHING) {
std::cerr << "NGCFT1 error: data_ack but not in SENDING or FINISHING state (" << int(transfer.state) << ")\n";
return true;
}
//if ((length - curser) % sizeof(uint16_t) != 0) {
//fprintf(stderr, "FT: data_ack with misaligned data\n");
//return;
//}
transfer.time_since_activity = 0.f;
std::vector<LEDBAT::SeqIDType> seqs;
for (const auto it : e.sequence_ids) {
// TODO: improve this o.o
seqs.push_back({e.transfer_id, it});
transfer.ssb.erase(it);
}
peer.cca->onAck(seqs);
// delete if all packets acked
if (transfer.file_size == transfer.file_size_current && transfer.ssb.size() == 0) {
std::cout << "NGCFT1: " << int(e.transfer_id) << " done\n";
dispatch(
NGCFT1_Event::send_done,
Events::NGCFT1_send_done{
e.group_number, e.peer_number,
e.transfer_id,
}
);
peer.send_transfers[e.transfer_id].reset();
}
return true;
}
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_message& e) {
std::cout << "NGCFT1: FT1_MESSAGE mid:" << e.message_id << " fk:" << e.file_kind << " [" << bin2hex(e.file_id) << "]\n";
// .... just rethrow??
// TODO: dont
return dispatch(
NGCFT1_Event::recv_message,
Events::NGCFT1_recv_message{
e.group_number, e.peer_number,
e.message_id,
static_cast<NGCFT1_file_kind>(e.file_kind),
e.file_id.data(), e.file_id.size()
}
);
}
bool NGCFT1::onToxEvent(const Tox_Event_Group_Peer_Exit* e) {
const auto group_number = tox_event_group_peer_exit_get_group_number(e);
const auto peer_number = tox_event_group_peer_exit_get_peer_id(e);
// peer disconnected, end all transfers
if (!groups.count(group_number)) {
return false;
}
auto& group = groups.at(group_number);
if (!group.peers.count(peer_number)) {
return false;
}
auto& peer = group.peers.at(peer_number);
for (size_t i = 0; i < peer.send_transfers.size(); i++) {
auto& it_opt = peer.send_transfers.at(i);
if (!it_opt.has_value()) {
continue;
}
std::cout << "NGCFT1: sending " << int(i) << " canceled bc peer offline\n";
dispatch(
NGCFT1_Event::send_done,
Events::NGCFT1_send_done{
group_number, peer_number,
static_cast<uint8_t>(i),
}
);
it_opt.reset();
}
for (size_t i = 0; i < peer.recv_transfers.size(); i++) {
auto& it_opt = peer.recv_transfers.at(i);
if (!it_opt.has_value()) {
continue;
}
std::cout << "NGCFT1: receiving " << int(i) << " canceled bc peer offline\n";
dispatch(
NGCFT1_Event::recv_done,
Events::NGCFT1_recv_done{
group_number, peer_number,
static_cast<uint8_t>(i),
}
);
it_opt.reset();
}
// reset cca
peer.cca = std::make_unique<LEDBAT>(500-4); // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
return false;
}

252
solanaceae/ngc_ft1/ngcft1.hpp Normal file
View File

@ -0,0 +1,252 @@
#pragma once
// solanaceae port of tox_ngc_ft1
#include <solanaceae/toxcore/tox_interface.hpp>
#include <solanaceae/toxcore/tox_event_interface.hpp>
#include "./ngcext.hpp"
#include "./ledbat.hpp"
#include "./rcv_buf.hpp"
#include "./snd_buf.hpp"
#include "./ngcft1_file_kind.hpp"
#include <map>
#include <set>
#include <memory>
namespace Events {
struct NGCFT1_recv_request {
uint32_t group_number;
uint32_t peer_number;
NGCFT1_file_kind file_kind;
const uint8_t* file_id;
size_t file_id_size;
};
struct NGCFT1_recv_init {
uint32_t group_number;
uint32_t peer_number;
NGCFT1_file_kind file_kind;
const uint8_t* file_id;
size_t file_id_size;
const uint8_t transfer_id;
const size_t file_size;
// return true to accept, false to deny
bool& accept;
};
struct NGCFT1_recv_data {
uint32_t group_number;
uint32_t peer_number;
uint8_t transfer_id;
size_t data_offset;
const uint8_t* data;
size_t data_size;
};
// request to fill data_size bytes into data
struct NGCFT1_send_data {
uint32_t group_number;
uint32_t peer_number;
uint8_t transfer_id;
size_t data_offset;
uint8_t* data;
size_t data_size;
};
struct NGCFT1_recv_done {
uint32_t group_number;
uint32_t peer_number;
uint8_t transfer_id;
// TODO: reason
};
struct NGCFT1_send_done {
uint32_t group_number;
uint32_t peer_number;
uint8_t transfer_id;
// TODO: reason
};
struct NGCFT1_recv_message {
uint32_t group_number;
uint32_t peer_number;
uint32_t message_id;
NGCFT1_file_kind file_kind;
const uint8_t* file_id;
size_t file_id_size;
};
} // Events
enum class NGCFT1_Event : uint8_t {
recv_request,
recv_init,
recv_data,
send_data,
recv_done,
send_done,
recv_message,
MAX
};
struct NGCFT1EventI {
using enumType = NGCFT1_Event;
virtual bool onEvent(const Events::NGCFT1_recv_request&) { return false; }
virtual bool onEvent(const Events::NGCFT1_recv_init&) { return false; }
virtual bool onEvent(const Events::NGCFT1_recv_data&) { return false; }
virtual bool onEvent(const Events::NGCFT1_send_data&) { return false; } // const?
virtual bool onEvent(const Events::NGCFT1_recv_done&) { return false; }
virtual bool onEvent(const Events::NGCFT1_send_done&) { return false; }
virtual bool onEvent(const Events::NGCFT1_recv_message&) { return false; }
};
using NGCFT1EventProviderI = EventProviderI<NGCFT1EventI>;
class NGCFT1 : public ToxEventI, public NGCEXTEventI, public NGCFT1EventProviderI {
ToxI& _t;
ToxEventProviderI& _tep;
NGCEXTEventProviderI& _neep;
// TODO: config
size_t acks_per_packet {3u}; // 3
float init_retry_timeout_after {5.f}; // 10sec
float sending_give_up_after {30.f}; // 30sec
struct Group {
struct Peer {
std::unique_ptr<LEDBAT> cca = std::make_unique<LEDBAT>(500-4); // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
struct RecvTransfer {
uint32_t file_kind;
std::vector<uint8_t> file_id;
enum class State {
INITED, //init acked, but no data received yet (might be dropped)
RECV, // receiving data
} state;
// float time_since_last_activity ?
size_t file_size {0};
size_t file_size_current {0};
// sequence id based reassembly
RecvSequenceBuffer rsb;
};
std::array<std::optional<RecvTransfer>, 256> recv_transfers;
size_t next_recv_transfer_idx {0}; // next id will be 0
struct SendTransfer {
uint32_t file_kind;
std::vector<uint8_t> file_id;
enum class State {
INIT_SENT, // keep this state until ack or deny or giveup
SENDING, // we got the ack and are now sending data
FINISHING, // we sent all data but acks still outstanding????
// delete
} state;
size_t inits_sent {1}; // is sent when creating
float time_since_activity {0.f};
size_t file_size {0};
size_t file_size_current {0};
// sequence array
// list of sent but not acked seq_ids
SendSequenceBuffer ssb;
};
std::array<std::optional<SendTransfer>, 256> send_transfers;
size_t next_send_transfer_idx {0}; // next id will be 0
};
std::map<uint32_t, Peer> peers;
};
std::map<uint32_t, Group> groups;
protected:
bool sendPKG_FT1_REQUEST(uint32_t group_number, uint32_t peer_number, uint32_t file_kind, const uint8_t* file_id, size_t file_id_size);
bool sendPKG_FT1_INIT(uint32_t group_number, uint32_t peer_number, uint32_t file_kind, uint64_t file_size, uint8_t transfer_id, const uint8_t* file_id, size_t file_id_size);
bool sendPKG_FT1_INIT_ACK(uint32_t group_number, uint32_t peer_number, uint8_t transfer_id);
bool sendPKG_FT1_DATA(uint32_t group_number, uint32_t peer_number, uint8_t transfer_id, uint16_t sequence_id, const uint8_t* data, size_t data_size);
bool sendPKG_FT1_DATA_ACK(uint32_t group_number, uint32_t peer_number, uint8_t transfer_id, const uint16_t* seq_ids, size_t seq_ids_size);
bool sendPKG_FT1_MESSAGE(uint32_t group_number, uint32_t message_id, uint32_t file_kind, const uint8_t* file_id, size_t file_id_size);
void updateSendTransfer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<LEDBAT::SeqIDType>& timeouts_set);
void iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer);
public:
NGCFT1(
ToxI& t,
ToxEventProviderI& tep,
NGCEXTEventProviderI& neep
);
void iterate(float delta);
public: // ft1 api
// TODO: public variant?
void NGC_FT1_send_request_private(
uint32_t group_number, uint32_t peer_number,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size
);
// public does not make sense here
bool NGC_FT1_send_init_private(
uint32_t group_number, uint32_t peer_number,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size,
size_t file_size,
uint8_t* transfer_id
);
// sends the message and fills in message_id
bool NGC_FT1_send_message_public(
uint32_t group_number,
uint32_t& message_id,
uint32_t file_kind,
const uint8_t* file_id, size_t file_id_size
);
protected:
bool onEvent(const Events::NGCEXT_ft1_request&) override;
bool onEvent(const Events::NGCEXT_ft1_init&) override;
bool onEvent(const Events::NGCEXT_ft1_init_ack&) override;
bool onEvent(const Events::NGCEXT_ft1_data&) override;
bool onEvent(const Events::NGCEXT_ft1_data_ack&) override;
bool onEvent(const Events::NGCEXT_ft1_message&) override;
protected:
bool onToxEvent(const Tox_Event_Group_Peer_Exit* e) override;
//bool onToxEvent(const Tox_Event_Group_Custom_Packet* e) override;
//bool onToxEvent(const Tox_Event_Group_Custom_Private_Packet* e) override;
};

76
solanaceae/ngc_ft1/ngcft1_file_kind.hpp Normal file
View File

@ -0,0 +1,76 @@
#pragma once
#include <cstdint>
// uint32_t - same as tox friend ft
// TODO: fill in toxfriend file kinds
enum class NGCFT1_file_kind : uint32_t {
//INVALID = 0u, // DATA?
// id:
// group (implicit)
// peer pub key + msg_id
NGC_HS1_MESSAGE_BY_ID = 1u, // history sync PoC 1
// TODO: oops, 1 should be avatar v1
// id: TOX_FILE_ID_LENGTH (32) bytes
// this is basically and id and probably not a hash, like the tox friend api
// this id can be unique between 2 peers
ID = 8u,
// id: hash of the info, like a torrent infohash (using the same hash as the data)
// TODO: determain internal format
// draft: (for single file)
// - 256 bytes | filename
// - 8bytes | file size
// - 4bytes | chunk size
// - array of chunk hashes (ids) [
// - SHA1 bytes (20)
// - ]
HASH_SHA1_INFO,
// draft: (for single file) v2
// - c-string | filename
// - 8bytes | file size
// - 4bytes | chunk size
// - array of chunk hashes (ids) [
// - SHA1 bytes (20)
// - ]
HASH_SHA1_INFO2,
// draft: multiple files
// - 4bytes | number of filenames
// - array of filenames (variable length c-strings) [
// - c-string | filename (including path and '/' as dir seperator)
// - ]
// - 256 bytes | filename
// - 8bytes | file size
// - fixed chunk size of 4kb
// - array of chunk hashes (ids) [
// - SHAX bytes
// - ]
HASH_SHA1_INFO3,
HASH_SHA2_INFO, // hm?
// id: hash of the content
// TODO: fixed chunk size or variable (defined in info)
// if "variable" sized, it can be aliased with TORRENT_V1_CHUNK in the implementation
HASH_SHA1_CHUNK,
HASH_SHA2_CHUNK,
// TODO: design the same thing again for tox? (msg_pack instead of bencode?)
// id: infohash
TORRENT_V1_METAINFO,
// id: sha1
TORRENT_V1_PIECE, // alias with SHA1_CHUNK?
// TODO: fix all the v2 stuff here
// id: infohash
// in v2, metainfo contains only the root hashes of the merkletree(s)
TORRENT_V2_METAINFO,
// id: root hash
// contains all the leaf hashes for a file root hash
TORRENT_V2_FILE_HASHES,
// id: sha256
// always of size 16KiB, except if last piece in file
TORRENT_V2_PIECE,
};

44
solanaceae/ngc_ft1/rcv_buf.cpp Normal file
View File

@ -0,0 +1,44 @@
#include "./rcv_buf.hpp"
#include <cassert>
void RecvSequenceBuffer::erase(uint16_t seq) {
entries.erase(seq);
}
// inflight chunks
size_t RecvSequenceBuffer::size(void) const {
return entries.size();
}
void RecvSequenceBuffer::add(uint16_t seq_id, std::vector<uint8_t>&& data) {
entries[seq_id] = {data};
ack_seq_ids.push_back(seq_id);
if (ack_seq_ids.size() > 3) { // TODO: magic
ack_seq_ids.pop_front();
}
}
bool RecvSequenceBuffer::canPop(void) const {
return entries.count(next_seq_id);
}
std::vector<uint8_t> RecvSequenceBuffer::pop(void) {
assert(canPop());
auto tmp_data = entries.at(next_seq_id).data;
erase(next_seq_id);
next_seq_id++;
return tmp_data;
}
// for acking, might be bad since its front
std::vector<uint16_t> RecvSequenceBuffer::frontSeqIDs(size_t count) const {
std::vector<uint16_t> seq_ids;
auto it = entries.cbegin();
for (size_t i = 0; i < count && it != entries.cend(); i++, it++) {
seq_ids.push_back(it->first);
}
return seq_ids;
}

35
solanaceae/ngc_ft1/rcv_buf.hpp Normal file
View File

@ -0,0 +1,35 @@
#pragma once
#include <vector>
#include <map>
#include <deque>
#include <cstdint>
struct RecvSequenceBuffer {
struct RSBEntry {
std::vector<uint8_t> data;
};
// sequence_id -> entry
std::map<uint16_t, RSBEntry> entries;
uint16_t next_seq_id {0};
// list of seq_ids to ack, this is seperate bc rsbentries are deleted once processed
std::deque<uint16_t> ack_seq_ids;
void erase(uint16_t seq);
// inflight chunks
size_t size(void) const;
void add(uint16_t seq_id, std::vector<uint8_t>&& data);
bool canPop(void) const;
std::vector<uint8_t> pop(void);
// for acking, might be bad since its front
std::vector<uint16_t> frontSeqIDs(size_t count = 5) const;
};

16
solanaceae/ngc_ft1/snd_buf.cpp Normal file
View File

@ -0,0 +1,16 @@
#include "./snd_buf.hpp"
void SendSequenceBuffer::erase(uint16_t seq) {
entries.erase(seq);
}
// inflight chunks
size_t SendSequenceBuffer::size(void) const {
return entries.size();
}
uint16_t SendSequenceBuffer::add(std::vector<uint8_t>&& data) {
entries[next_seq_id] = {data, 0.f};
return next_seq_id++;
}

33
solanaceae/ngc_ft1/snd_buf.hpp Normal file
View File

@ -0,0 +1,33 @@
#pragma once
#include <vector>
#include <map>
#include <cstdint>
struct SendSequenceBuffer {
struct SSBEntry {
std::vector<uint8_t> data; // the data (variable size, but smaller than 500)
float time_since_activity {0.f};
};
// sequence_id -> entry
std::map<uint16_t, SSBEntry> entries;
uint16_t next_seq_id {0};
void erase(uint16_t seq);
// inflight chunks
size_t size(void) const;
uint16_t add(std::vector<uint8_t>&& data);
template<typename FN>
void for_each(float time_delta, FN&& fn) {
for (auto& [id, entry] : entries) {
entry.time_since_activity += time_delta;
fn(id, entry.data, entry.time_since_activity);
}
}
};

139
solanaceae/ngc_ft1_sha1/ft1_sha1_info.cpp Normal file
View File

@ -0,0 +1,139 @@
#include "./ft1_sha1_info.hpp"
#include <sodium.h>
SHA1Digest::SHA1Digest(const std::vector<uint8_t>& v) {
assert(v.size() == data.size());
for (size_t i = 0; i < data.size(); i++) {
data[i] = v[i];
}
}
SHA1Digest::SHA1Digest(const uint8_t* d, size_t s) {
assert(s == data.size());
for (size_t i = 0; i < data.size(); i++) {
data[i] = d[i];
}
}
std::ostream& operator<<(std::ostream& out, const SHA1Digest& v) {
std::string str{};
str.resize(v.size()*2, '?');
// HECK, std is 1 larger than size returns ('\0')
sodium_bin2hex(str.data(), str.size()+1, v.data.data(), v.data.size());
out << str;
return out;
}
size_t FT1InfoSHA1::chunkSize(size_t chunk_index) const {
if (chunk_index+1 == chunks.size()) {
// last chunk
return file_size - chunk_index * chunk_size;
} else {
return chunk_size;
}
}
std::vector<uint8_t> FT1InfoSHA1::toBuffer(void) const {
std::vector<uint8_t> buffer;
assert(!file_name.empty());
// TODO: optimize
for (size_t i = 0; i < 256; i++) {
if (i < file_name.size()) {
buffer.push_back(file_name.at(i));
} else {
buffer.push_back(0);
}
}
assert(buffer.size() == 256);
{ // HACK: endianess
buffer.push_back((file_size>>(0*8)) & 0xff);
buffer.push_back((file_size>>(1*8)) & 0xff);
buffer.push_back((file_size>>(2*8)) & 0xff);
buffer.push_back((file_size>>(3*8)) & 0xff);
buffer.push_back((file_size>>(4*8)) & 0xff);
buffer.push_back((file_size>>(5*8)) & 0xff);
buffer.push_back((file_size>>(6*8)) & 0xff);
buffer.push_back((file_size>>(7*8)) & 0xff);
}
assert(buffer.size() == 256+8);
// chunk size
{ // HACK: endianess
buffer.push_back((chunk_size>>(0*8)) & 0xff);
buffer.push_back((chunk_size>>(1*8)) & 0xff);
buffer.push_back((chunk_size>>(2*8)) & 0xff);
buffer.push_back((chunk_size>>(3*8)) & 0xff);
}
assert(buffer.size() == 256+8+4);
for (const auto& chunk : chunks) {
for (size_t i = 0; i < chunk.data.size(); i++) {
buffer.push_back(chunk.data[i]);
}
}
assert(buffer.size() == 256+8+4+20*chunks.size());
return buffer;
}
void FT1InfoSHA1::fromBuffer(const std::vector<uint8_t>& buffer) {
assert(buffer.size() >= 256+8+4);
// TODO: optimize
file_name.clear();
for (size_t i = 0; i < 256; i++) {
char next_char = static_cast<char>(buffer[i]);
if (next_char == 0) {
break;
}
file_name.push_back(next_char);
}
{ // HACK: endianess
file_size = 0;
file_size |= uint64_t(buffer[256+0]) << (0*8);
file_size |= uint64_t(buffer[256+1]) << (1*8);
file_size |= uint64_t(buffer[256+2]) << (2*8);
file_size |= uint64_t(buffer[256+3]) << (3*8);
file_size |= uint64_t(buffer[256+4]) << (4*8);
file_size |= uint64_t(buffer[256+5]) << (5*8);
file_size |= uint64_t(buffer[256+6]) << (6*8);
file_size |= uint64_t(buffer[256+7]) << (7*8);
}
{ // HACK: endianess
chunk_size = 0;
chunk_size |= uint32_t(buffer[256+8+0]) << (0*8);
chunk_size |= uint32_t(buffer[256+8+1]) << (1*8);
chunk_size |= uint32_t(buffer[256+8+2]) << (2*8);
chunk_size |= uint32_t(buffer[256+8+3]) << (3*8);
}
assert((buffer.size()-(256+8+4)) % 20 == 0);
for (size_t offset = 256+8+4; offset < buffer.size();) {
assert(buffer.size() >= offset + 20);
auto& chunk = chunks.emplace_back();
for (size_t i = 0; i < chunk.size(); i++, offset++) {
chunk.data[i] = buffer.at(offset);
}
// TODO: error/leftover checking
}
}
std::ostream& operator<<(std::ostream& out, const FT1InfoSHA1& v) {
out << " file_name: " << v.file_name << "\n";
out << " file_size: " << v.file_size << "\n";
out << " chunk_size: " << v.chunk_size << "\n";
out << " chunks.size(): " << v.chunks.size() << "\n";
return out;
}

55
solanaceae/ngc_ft1_sha1/ft1_sha1_info.hpp Normal file
View File

@ -0,0 +1,55 @@
#pragma once
#include <cstddef>
#include <cstdint>
#include <array>
#include <ostream>
#include <vector>
#include <cassert>
#include <string>
struct SHA1Digest {
std::array<uint8_t, 20> data;
SHA1Digest(void) = default;
SHA1Digest(const std::vector<uint8_t>& v);
SHA1Digest(const uint8_t* d, size_t s);
bool operator==(const SHA1Digest& other) const { return data == other.data; }
bool operator!=(const SHA1Digest& other) const { return data != other.data; }
size_t size(void) const { return data.size(); }
};
std::ostream& operator<<(std::ostream& out, const SHA1Digest& v);
namespace std { // inject
template<> struct hash<SHA1Digest> {
std::size_t operator()(const SHA1Digest& h) const noexcept {
return
size_t(h.data[0]) << (0*8) |
size_t(h.data[1]) << (1*8) |
size_t(h.data[2]) << (2*8) |
size_t(h.data[3]) << (3*8) |
size_t(h.data[4]) << (4*8) |
size_t(h.data[5]) << (5*8) |
size_t(h.data[6]) << (6*8) |
size_t(h.data[7]) << (7*8)
;
}
};
} // std
struct FT1InfoSHA1 {
std::string file_name;
uint64_t file_size {0};
uint32_t chunk_size {128*1024}; // 128KiB for now
std::vector<SHA1Digest> chunks;
size_t chunkSize(size_t chunk_index) const;
std::vector<uint8_t> toBuffer(void) const;
void fromBuffer(const std::vector<uint8_t>& buffer);
};
std::ostream& operator<<(std::ostream& out, const FT1InfoSHA1& v);

26
solanaceae/ngc_ft1_sha1/hash_utils.cpp Normal file
View File

@ -0,0 +1,26 @@
#include "./hash_utils.hpp"
#include <sha1.h>
// returns the 20bytes sha1 hash
std::vector<uint8_t> hash_sha1(const uint8_t* data, size_t size) {
SHA1_CTX ctx;
SHA1Init(&ctx);
{ // lib only takes uint32_t sizes, so chunk it
constexpr size_t hash_block_size {0xffffffff};
size_t i = 0;
for (; i + hash_block_size < size; i += hash_block_size) {
SHA1Update(&ctx, reinterpret_cast<const uint8_t*>(data) + i, hash_block_size);
}
if (i < size) {
SHA1Update(&ctx, reinterpret_cast<const uint8_t*>(data) + i, size - i);
}
}
std::vector<uint8_t> sha1_hash(20);
SHA1Final(sha1_hash.data(), &ctx);
return sha1_hash;
}

10
solanaceae/ngc_ft1_sha1/hash_utils.hpp Normal file
View File

@ -0,0 +1,10 @@
#pragma once
#include <cstdint>
#include <vector>
// returns the 20bytes sha1 hash
std::vector<uint8_t> hash_sha1(const uint8_t* data, size_t size);
inline std::vector<uint8_t> hash_sha1(const char* data, size_t size) { return hash_sha1(reinterpret_cast<const uint8_t*>(data), size); }

1311
solanaceae/ngc_ft1_sha1/sha1_ngcft1.cpp Normal file
View File

File diff suppressed because it is too large Load Diff

134
solanaceae/ngc_ft1_sha1/sha1_ngcft1.hpp Normal file
View File

@ -0,0 +1,134 @@
#pragma once
// solanaceae port of sha1 fts for NGCFT1
#include <solanaceae/contact/contact_model3.hpp>
#include <solanaceae/message3/registry_message_model.hpp>
#include <solanaceae/tox_contacts/tox_contact_model2.hpp>
#include "./ngcft1.hpp"
#include "./ft1_sha1_info.hpp"
#include <entt/entity/registry.hpp>
#include <entt/entity/handle.hpp>
#include <entt/container/dense_map.hpp>
#include <variant>
#include <random>
enum class Content : uint32_t {};
using ContentRegistry = entt::basic_registry<Content>;
using ContentHandle = entt::basic_handle<ContentRegistry>;
class SHA1_NGCFT1 : public RegistryMessageModelEventI, public NGCFT1EventI {
Contact3Registry& _cr;
RegistryMessageModel& _rmm;
NGCFT1& _nft;
ToxContactModel2& _tcm;
std::minstd_rand _rng {1337*11};
// registry per group?
ContentRegistry _contentr;
// limit this to each group?
entt::dense_map<SHA1Digest, ContentHandle> _info_to_content;
// sha1 chunk index
// TODO: optimize lookup
// TODO: multiple contents. hashes might be unique, but data is not
entt::dense_map<SHA1Digest, ContentHandle> _chunks;
// group_number, peer_number, content, chunk_hash, timer
std::deque<std::tuple<uint32_t, uint32_t, ContentHandle, SHA1Digest, float>> _queue_requested_chunk;
//void queueUpRequestInfo(uint32_t group_number, uint32_t peer_number, const SHA1Digest& hash);
void queueUpRequestChunk(uint32_t group_number, uint32_t peer_number, ContentHandle content, const SHA1Digest& hash);
struct SendingTransfer {
struct Info {
// copy of info data
// too large?
std::vector<uint8_t> info_data;
};
struct Chunk {
ContentHandle content;
size_t chunk_index; // <.< remove offset_into_file
//uint64_t offset_into_file;
// or data?
// if memmapped, this would be just a pointer
};
std::variant<Info, Chunk> v;
float time_since_activity {0.f};
};
// key is groupid + peerid
entt::dense_map<uint64_t, entt::dense_map<uint8_t, SendingTransfer>> _sending_transfers;
struct ReceivingTransfer {
struct Info {
ContentHandle content;
// copy of info data
// too large?
std::vector<uint8_t> info_data;
};
struct Chunk {
ContentHandle content;
std::vector<size_t> chunk_indices;
// or data?
// if memmapped, this would be just a pointer
};
std::variant<Info, Chunk> v;
float time_since_activity {0.f};
};
// key is groupid + peerid
entt::dense_map<uint64_t, entt::dense_map<uint8_t, ReceivingTransfer>> _receiving_transfers;
// makes request rotate around open content
std::deque<ContentHandle> _queue_content_want_info;
std::deque<ContentHandle> _queue_content_want_chunk;
static uint64_t combineIds(const uint32_t group_number, const uint32_t peer_number);
void updateMessages(ContentHandle ce);
std::optional<std::pair<uint32_t, uint32_t>> selectPeerForRequest(ContentHandle ce);
public: // TODO: config
bool _udp_only {false};
size_t _max_concurrent_in {4};
size_t _max_concurrent_out {6};
// TODO: probably also includes running transfers rn (meh)
size_t _max_pending_requests {16}; // per content
public:
SHA1_NGCFT1(
Contact3Registry& cr,
RegistryMessageModel& rmm,
NGCFT1& nft,
ToxContactModel2& tcm
);
void iterate(float delta);
protected: // rmm events (actions)
bool onEvent(const Message::Events::MessageUpdated&) override;
protected: // events
bool onEvent(const Events::NGCFT1_recv_request&) override;
bool onEvent(const Events::NGCFT1_recv_init&) override;
bool onEvent(const Events::NGCFT1_recv_data&) override;
bool onEvent(const Events::NGCFT1_send_data&) override; // const?
bool onEvent(const Events::NGCFT1_recv_done&) override;
bool onEvent(const Events::NGCFT1_send_done&) override;
bool onEvent(const Events::NGCFT1_recv_message&) override;
bool sendFilePath(const Contact3 c, std::string_view file_name, std::string_view file_path) override;
};