2 Commits
18 changed files with 619 additions and 582 deletions

No files matched your search

+6
View File
@@ -1,3 +1,9 @@
-- 2.1.4.(2024-03-24)
FEATURES
* Refactor the handling of SQL insert buffers
BUILD
* now requires libfmt, used for formatting strings
-- 2.1.3 (2024-03-09)
BREAKING
* API user endpoint output changed to JSON
+5
View File
@@ -49,6 +49,9 @@ find_package(MYSQLPP REQUIRED)
find_package(LibEv REQUIRED)
include_directories(${LIBEV_INCLUDE_DIR})
find_package(fmt REQUIRED)
message(STATUS "Found libfmt version ${fmt_VERSION}")
include_directories(${CMAKE_SOURCE_DIR}/libs)
file(GLOB HEADERS src/*.h)
@@ -60,6 +63,7 @@ SET(LINK_LIBRARIES
${Boost_SYSTEM_LIBRARY}
${Boost_IOSTREAMS_LIBRARY}
${LIBEV_LIBRARY}
${LIBFMT_IBRARY}
${MYSQLPP_LIBRARY}
Threads::Threads
)
@@ -75,4 +79,5 @@ endif()
target_link_libraries(
ocelot
${LINK_LIBRARIES}
fmt::fmt
)
+1
View File
@@ -10,6 +10,7 @@ RUN apt-get update \
libboost-iostreams-dev \
libboost-system-dev \
libev-dev \
libfmt-dev \
libjemalloc-dev \
libmysql++-dev \
netcat-traditional \
+4 -2
View File
@@ -8,6 +8,7 @@ It supports requests over TCP and can only track IPv4 peers.
* [GCC/G++](http://gcc.gnu.org/) (11+ required; 11.4.0+ recommended)
* [Boost](http://www.boost.org/) (1.74.0+ required)
* [libev](http://software.schmorp.de/pkg/libev.html) (required)
* [libfmt](https://github.com/fmtlib/fmt) (8.1+ required)
* [MySQL++](http://tangentsoft.net/mysql++/) (3.2.0+ required)
* [jemalloc](http://jemalloc.net/) (optional, but strongly recommended)
* [TCMalloc](http://goog-perftools.sourceforge.net/doc/tcmalloc.html) (optional)
@@ -23,9 +24,10 @@ sudo apt-get install \
libboost-iostreams-dev \
libboost-system-dev \
libev-dev \
libfmt-dev \
libjemalloc-dev \
libmysql++-dev \
netcat-traditional \
netcat-traditional \
pkg-config
cmake -Wno-dev . -B build
make -C build
@@ -42,7 +44,7 @@ docker run -v $(pwd)/ocelot.conf:/srv/ocelot.conf ocelot
### Standalone Installation
* Prepare the build environment. (This must be re-executed when new source files are added).
* Prepare the build environment. (This must be re-executed when new source files or libraries are added).
cmake -Wno-dev . -B build
+3 -3
View File
@@ -47,7 +47,7 @@ void confval::set(const std::string &value) {
if (val_type == CONF_BOOL) {
bool_val = value == "1" || value == "true" || value == "yes";
} else if (val_type == CONF_UINT) {
uint_val = strtoint32(value);
uint_val = atoi(value.c_str());
} else if (val_type == CONF_STR) {
str_val = value;
}
@@ -109,7 +109,7 @@ void config::init() {
confval * config::get(const std::string &setting_name) {
const auto setting = settings.find(setting_name);
if (setting == settings.end()) {
spdlog::get("logger")->info("WARNING: Unrecognized setting '" + setting_name + "'");
spdlog::get("logger")->info("WARNING: Unrecognized setting '{}'", setting_name);
return dummy_setting;
}
return &setting->second;
@@ -152,7 +152,7 @@ void config::reload() {
const std::string conf_file_path(get_str("conf_file_path"));
std::ifstream conf_file(conf_file_path);
if (conf_file.fail()) {
spdlog::get("logger")->error("Config file '" + conf_file_path + "' couldn't be opened");
spdlog::get("logger")->error("Config file '{}' couldn't be opened", conf_file_path);
} else {
init();
load(conf_file);
+1 -1
View File
@@ -21,7 +21,7 @@ class confval {
confval();
confval(bool value);
confval(uint32_t value);
confval(const char * value);
confval(const char *value);
uint32_t get_uint();
bool get_bool();
std::string get_str();
+263 -329
View File
@@ -21,7 +21,11 @@
#define DB_LOCK_TIMEOUT 50
mysql::mysql(config * conf) : u_active(false), t_active(false), p_active(false), s_active(false), tok_active(false) {
mysql::mysql(config * conf) :
update_heavy_peer_buffer(*this), update_light_peer_buffer(*this),
update_snatch_buffer(*this), update_token_buffer(*this),
update_torrent_buffer(*this), update_user_buffer(*this),
peer_flush_active(false), snatch_flush_active(false), token_flush_active(false), torrent_flush_active(false), user_flush_active(false) {
logger = spdlog::get("logger");
load_config(conf);
if (mysql_db.empty()) {
@@ -34,7 +38,7 @@ mysql::mysql(config * conf) : u_active(false), t_active(false), p_active(false),
conn.set_option(&reconnect);
conn.connect(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), mysql_port);
} catch (const mysqlpp::Exception &er) {
logger->error("Failed to connect to MySQL (" + std::string(er.what()) + ")");
logger->error("Failed to connect to MySQL ({})", er.what());
return;
}
@@ -69,6 +73,12 @@ mysqlpp::Connection mysql::create_connection() {
return c;
}
std::string mysql::quote(std::string value) {
mysqlpp::Query query = conn.query();
query.escape_string(&value);
return '\'' + value + '\'';
}
void mysql::clear_peer_data() {
try {
mysqlpp::Query query = conn.query("TRUNCATE xbt_files_users;");
@@ -80,9 +90,9 @@ void mysql::clear_peer_data() {
logger->error("Unable to reset seeder and leecher count!");
}
} catch (const mysqlpp::BadQuery &er) {
logger->error("Query error in clear_peer_data: " + std::string(er.what()));
logger->error("Query error in clear_peer_data: {}", er.what());
} catch (const mysqlpp::Exception &er) {
logger->error("Query error in clear_peer_data: " + std::string(er.what()));
logger->error("Query error in clear_peer_data: {}", er.what());
}
}
@@ -147,15 +157,19 @@ void mysql::load_torrents(torrent_list &torrents) {
}
}
} catch (const mysqlpp::BadQuery &er) {
logger->error("Query error in load_torrents: " + std::string(er.what()));
logger->error("Query error in load_torrents: {}", er.what());
return;
}
logger->info("Loaded " + std::to_string(torrents.size()) + " torrents");
logger->info("Loaded {} torrents", torrents.size());
load_tokens(torrents);
}
void mysql::load_users(user_list &users) {
mysqlpp::Query query = conn.query("SELECT ID, can_leech, torrent_pass, (Visible='0' OR IP='127.0.0.1') AS Protected FROM users_main WHERE Enabled='1';");
std::string sql(
"SELECT ID, can_leech, torrent_pass, (Visible='0' OR IP='127.0.0.1') AS Protected"
" FROM users_main WHERE Enabled='1'"
);
mysqlpp::Query query = conn.query(sql);
try {
mysqlpp::StoreQueryResult res = query.store();
size_t num_rows = res.num_rows();
@@ -192,14 +206,18 @@ void mysql::load_users(user_list &users) {
}
}
} catch (const mysqlpp::BadQuery &er) {
logger->error("Query error in load_users: " + std::string(er.what()));
logger->error("Query error in load_users: {} [{}]", er.what(), sql);
return;
}
logger->info("Loaded " + std::to_string(users.size()) + " users");
logger->info("Loaded {} users", users.size());
}
void mysql::load_tokens(torrent_list &torrents) {
mysqlpp::Query query = conn.query("SELECT uf.UserID, t.info_hash FROM users_freeleeches AS uf INNER JOIN torrents AS t ON t.ID = uf.TorrentID WHERE uf.Expired = '0';");
std::string sql(
"SELECT uf.UserID, t.info_hash FROM users_freeleeches AS uf "
"INNER JOIN torrents AS t ON t.ID = uf.TorrentID WHERE uf.Expired = '0'"
);
mysqlpp::Query query = conn.query(sql);
int token_count = 0;
try {
mysqlpp::StoreQueryResult res = query.store();
@@ -216,15 +234,16 @@ void mysql::load_tokens(torrent_list &torrents) {
}
}
} catch (const mysqlpp::BadQuery &er) {
logger->error("Query error in load_tokens: " + std::string(er.what()));
logger->error("Query error in load_tokens: {} [{}]", er.what(), sql);
return;
}
logger->info("Loaded " + std::to_string(token_count) + " tokens");
logger->info("Loaded {} tokens", token_count);
}
void mysql::load_whitelist(std::vector<std::string> &whitelist) {
mysqlpp::Query query = conn.query("SELECT peer_id FROM xbt_client_whitelist;");
std::string sql("SELECT peer_id FROM xbt_client_whitelist");
mysqlpp::Query query = conn.query(sql);
try {
mysqlpp::StoreQueryResult res = query.store();
size_t num_rows = res.num_rows();
@@ -236,64 +255,40 @@ void mysql::load_whitelist(std::vector<std::string> &whitelist) {
whitelist.push_back(peer_id);
}
} catch (const mysqlpp::BadQuery &er) {
logger->error("Query error in load_whitelist: " + std::string(er.what()));
logger->error("Query error in load_whitelist: {} [{}]", er.what(), sql);
return;
}
if (whitelist.size() == 0) {
logger->info("Assuming no whitelist desired, disabling");
} else {
logger->info("Loaded " + std::to_string(whitelist.size()) + " clients into the whitelist");
}
logger->info("Loaded {} clients into the whitelist", whitelist.size());
}
void mysql::record_peer(const std::string &record) {
std::lock_guard<std::mutex> buf_lock(peer_buffer_lock);
update_heavy_peer_buffer += record;
}
void mysql::record_peer_light(const std::string &record) {
std::lock_guard<std::mutex> buf_lock(peer_light_buffer_lock);
update_light_peer_buffer += record;
}
void mysql::record_snatch(const std::string &record) {
std::lock_guard<std::mutex> buf_lock(snatch_buffer_lock);
update_snatch_buffer += record;
}
void mysql::record_token(const std::string &record) {
if (!update_token_buffer.empty()) {
update_token_buffer += ",";
}
std::lock_guard<std::mutex> buf_lock(token_buffer_lock);
update_token_buffer += record;
}
void mysql::record_user(const std::string &record) {
if (!update_user_buffer.empty()) {
update_user_buffer += ",";
}
update_user_buffer += record;
}
void mysql::record_torrent(const std::string &record) {
std::lock_guard<std::mutex> tb_lock(torrent_buffer_lock);
if (!update_torrent_buffer.empty()) {
update_torrent_buffer += ",";
}
std::lock_guard<std::mutex> buf_lock(torrent_buffer_lock);
update_torrent_buffer += record;
}
void mysql::record_peer(const std::string &record, const std::string &ip, const std::string &peer_id, const std::string &useragent) {
if (!update_heavy_peer_buffer.empty()) {
update_heavy_peer_buffer += ",";
}
mysqlpp::Query q = conn.query();
q << record << mysqlpp::quote << ip << ',' << mysqlpp::quote << peer_id << ',' << mysqlpp::quote << useragent << "," << time(NULL) << ')';
update_heavy_peer_buffer += q.str();
}
void mysql::record_peer(const std::string &record, const std::string &peer_id) {
if (!update_light_peer_buffer.empty()) {
update_light_peer_buffer += ",";
}
mysqlpp::Query q = conn.query();
q << record << mysqlpp::quote << peer_id << ',' << time(NULL) << ')';
update_light_peer_buffer += q.str();
}
void mysql::record_snatch(const std::string &record, const std::string &ip) {
if (!update_snatch_buffer.empty()) {
update_snatch_buffer += ",";
}
mysqlpp::Query q = conn.query();
q << record << ',' << mysqlpp::quote << ip << ')';
update_snatch_buffer += q.str();
void mysql::record_user(const std::string &record) {
std::lock_guard<std::mutex> buf_lock(user_buffer_lock);
update_user_buffer += record;
}
bool mysql::all_clear() {
@@ -308,350 +303,289 @@ void mysql::flush() {
flush_tokens();
}
void mysql::flush_users() {
if (readonly) {
update_user_buffer.clear();
return;
}
std::string sql;
std::lock_guard<std::mutex> uq_lock(user_queue_lock);
size_t qsize = user_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("User flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(user_queue.front().size()));
}
if (update_user_buffer.empty()) {
return;
}
sql = "INSERT INTO users_leech_stats (UserID, Uploaded, Downloaded) VALUES " + update_user_buffer +
" ON DUPLICATE KEY UPDATE Uploaded = Uploaded + VALUES(Uploaded), Downloaded = Downloaded + VALUES(Downloaded)";
user_queue.push(sql);
update_user_buffer.clear();
stats.user_queue_size = qsize + 1;
if (!u_active) {
std::thread thread(&mysql::do_flush_users, this);
thread.detach();
}
}
void mysql::flush_torrents() {
std::lock_guard<std::mutex> tb_lock(torrent_buffer_lock);
if (readonly) {
update_torrent_buffer.clear();
return;
}
std::string sql;
std::lock_guard<std::mutex> tq_lock(torrent_queue_lock);
size_t qsize = torrent_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Torrent flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(torrent_queue.front().size()));
}
if (update_torrent_buffer.empty()) {
return;
}
sql = "INSERT INTO torrents_leech_stats (TorrentID,Seeders,Leechers,Snatched,Balance) VALUES " + update_torrent_buffer +
" ON DUPLICATE KEY UPDATE Seeders=VALUES(Seeders), Leechers=VALUES(Leechers), " +
"Snatched=Snatched+VALUES(Snatched), Balance=VALUES(Balance), last_action = " +
"IF(VALUES(Seeders) > 0, NOW(), last_action)";
torrent_queue.push(sql);
update_torrent_buffer.clear();
sql.clear();
sql = "DELETE FROM torrents WHERE info_hash = ''";
torrent_queue.push(sql);
stats.torrent_queue_size = qsize + 2;
if (!t_active) {
std::thread thread(&mysql::do_flush_torrents, this);
thread.detach();
}
}
void mysql::flush_snatches() {
if (readonly) {
update_snatch_buffer.clear();
return;
}
std::string sql;
std::lock_guard<std::mutex> sq_lock(snatch_queue_lock);
size_t qsize = snatch_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Snatch flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(snatch_queue.front().size()));
}
if (update_snatch_buffer.empty()) {
return;
}
sql = "INSERT IGNORE INTO xbt_snatched (uid, fid, tstamp, IP) VALUES " + update_snatch_buffer;
snatch_queue.push(sql);
update_snatch_buffer.clear();
stats.snatch_queue_size = qsize + 1;
if (!s_active) {
std::thread thread(&mysql::do_flush_snatches, this);
thread.detach();
}
}
void mysql::flush_peers() {
std::unique_lock<std::mutex> q_lock(peer_queue_lock);
size_t qsize = peer_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Peer flush queue size: {}, next query length: {}", qsize, peer_queue.front().size());
}
q_lock.unlock();
std::unique_lock<std::mutex> buf_lock(peer_buffer_lock);
std::unique_lock<std::mutex> buf_light_lock(peer_light_buffer_lock);
if (readonly) {
update_light_peer_buffer.clear();
update_heavy_peer_buffer.clear();
return;
}
std::string sql;
std::lock_guard<std::mutex> pq_lock(peer_queue_lock);
size_t qsize = peer_queue.size();
short qsize_added = 0;
if (verbose_flush || qsize > 0) {
logger->info("Peer flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(peer_queue.front().size()));
}
// Nothing to do
if (update_light_peer_buffer.empty() && update_heavy_peer_buffer.empty()) {
/* The first step to update the database is to transform the tuples that
* have been stored into an SQL insert statement. The tuple buffer can then
* be cleared and the statement saved in a queue. The next step is to see
* if the queued statements can be sent to the database. If a database
* flush is already active then nothing else happens.
*
* Otherwise, a new thread is spun off to take the queued statements and
* execute them on the database in a FIFO manner. If for whatever reason
* the database cannot process the statement (bad credentials,
* incorrect/missing) the statements remain queued. If you are lucky, you
* can make the necessary changes in the database (alter a table, create
* the mysql user), and then the queue will empty.
*/
if (update_light_peer_buffer.empty() && update_heavy_peer_buffer.empty() && qsize == 0) {
// the buffers are empty and nothing is queued
return;
}
int16_t qsize_added = 0;
if (!update_heavy_peer_buffer.empty()) {
// Because xfu inserts are slow and ram is not infinite we need to
// limit this queue's size
// xfu will be messed up if the light query inserts a new row,
// but that's better than an oom crash
if (qsize >= 1000) {
std::lock_guard<std::mutex> q_lock(peer_queue_lock);
peer_queue.pop();
logger->error("Peer queue overflowed, update lost");
} else {
qsize_added += 1;
}
sql = "INSERT INTO xbt_files_users (uid,fid,active,uploaded,downloaded,upspeed,downspeed,remaining,corrupt," +
std::string("timespent,announced,ip,peer_id,useragent,mtime) VALUES ") + update_heavy_peer_buffer +
" ON DUPLICATE KEY UPDATE active=VALUES(active), uploaded=VALUES(uploaded), " +
"downloaded=VALUES(downloaded), upspeed=VALUES(upspeed), " +
"downspeed=VALUES(downspeed), remaining=VALUES(remaining), " +
"corrupt=VALUES(corrupt), timespent=VALUES(timespent), " +
"announced=VALUES(announced), mtime=VALUES(mtime)";
peer_queue.push(sql);
std::lock_guard<std::mutex> q_lock(peer_queue_lock);
peer_queue.push(fmt::format(
"INSERT INTO xbt_files_users"
"(uid,fid,active,uploaded,downloaded,upspeed,downspeed,remaining"
",corrupt,timespent,announced,ip,peer_id,useragent,mtime)"
"VALUES {} ON DUPLICATE KEY UPDATE active=VALUES(active)"
",uploaded=VALUES(uploaded),downloaded=VALUES(downloaded)"
",upspeed=VALUES(upspeed),downspeed=VALUES(downspeed)"
",remaining=VALUES(remaining),corrupt=VALUES(corrupt)"
",timespent=VALUES(timespent),announced=VALUES(announced)"
",mtime=VALUES(mtime)",
update_heavy_peer_buffer.str()
));
update_heavy_peer_buffer.clear();
sql.clear();
}
buf_lock.unlock();
if (!update_light_peer_buffer.empty()) {
// See comment above
if (qsize >= 1000) {
std::lock_guard<std::mutex> q_lock(peer_queue_lock);
peer_queue.pop();
logger->error("Peer queue overflowed, update lost");
} else {
qsize_added += 1;
}
sql = "INSERT INTO xbt_files_users (uid,fid,timespent,announced,peer_id,mtime) VALUES " +
update_light_peer_buffer +
" ON DUPLICATE KEY UPDATE upspeed=0, downspeed=0, timespent=VALUES(timespent), " +
"announced=VALUES(announced), mtime=VALUES(mtime)";
peer_queue.push(sql);
std::lock_guard<std::mutex> q_lock(peer_queue_lock);
peer_queue.push(fmt::format(
"INSERT INTO xbt_files_users(uid,fid,timespent,announced,peer_id,mtime)"
"VALUES {} ON DUPLICATE KEY UPDATE upspeed=0,downspeed=0"
",timespent=VALUES(timespent),announced=VALUES(announced)"
",mtime=VALUES(mtime)",
update_light_peer_buffer.str()
));
update_light_peer_buffer.clear();
sql.clear();
}
buf_light_lock.unlock();
stats.peer_queue_size = qsize + qsize_added;
if (!p_active) {
std::thread thread(&mysql::do_flush_peers, this);
if (!peer_flush_active) {
std::thread thread([this]() {
do_flush("peer", peer_queue, peer_queue_lock, peer_flush_active, stats.peer_queue_size);
});
thread.detach();
}
}
void mysql::flush_snatches() {
std::unique_lock<std::mutex> q_lock(snatch_queue_lock);
size_t qsize = snatch_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Snatch flush queue size: {}, next query length: {}", qsize, snatch_queue.front().size());
}
q_lock.unlock();
std::unique_lock<std::mutex> buf_lock(snatch_buffer_lock);
if (readonly) {
update_snatch_buffer.clear();
return;
}
if (update_snatch_buffer.empty() && qsize == 0) {
return;
}
if (!update_snatch_buffer.empty()) {
std::lock_guard<std::mutex> q_lock(snatch_queue_lock);
snatch_queue.push(fmt::format(
"INSERT IGNORE INTO xbt_snatched (uid, fid, tstamp, IP) VALUES {}",
update_snatch_buffer.str()
));
update_snatch_buffer.clear();
stats.snatch_queue_size = qsize + 1;
}
buf_lock.unlock();
if (!snatch_flush_active) {
std::thread thread([this]() {
do_flush("snatch", snatch_queue, snatch_queue_lock, snatch_flush_active, stats.snatch_queue_size);
});
thread.detach();
}
}
void mysql::flush_tokens() {
std::unique_lock<std::mutex> q_lock(token_queue_lock);
size_t qsize = token_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Token flush queue size: {}, next query length: {}", qsize, token_queue.front().size());
}
q_lock.unlock();
std::unique_lock<std::mutex> buf_lock(token_buffer_lock);
if (readonly) {
update_token_buffer.clear();
return;
}
std::string sql;
std::lock_guard<std::mutex> tq_lock(token_queue_lock);
size_t qsize = token_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Token flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(token_queue.front().size()));
}
if (update_token_buffer.empty()) {
if (update_token_buffer.empty() && qsize == 0) {
return;
}
sql = "INSERT INTO users_freeleeches (UserID, TorrentID, Downloaded) VALUES " + update_token_buffer +
" ON DUPLICATE KEY UPDATE Downloaded = Downloaded + VALUES(Downloaded)";
token_queue.push(sql);
update_token_buffer.clear();
stats.token_queue_size = qsize + 1;
if (!tok_active) {
std::thread thread(&mysql::do_flush_tokens, this);
if (!update_token_buffer.empty()) {
std::lock_guard<std::mutex> q_lock(token_queue_lock);
token_queue.push(fmt::format(
"INSERT INTO users_freeleeches(UserID,TorrentID,Downloaded)VALUES"
"{} ON DUPLICATE KEY UPDATE Downloaded=Downloaded+VALUES(Downloaded)",
update_token_buffer.str()
));
update_token_buffer.clear();
stats.token_queue_size = qsize + 1;
}
buf_lock.unlock();
if (!token_flush_active) {
std::thread thread([this]() {
do_flush("token", token_queue, token_queue_lock, token_flush_active, stats.token_queue_size);
});
thread.detach();
}
}
void mysql::do_flush_users() {
u_active = true;
try {
mysqlpp::Connection c = create_connection();
while (user_queue.size() > 0) {
try {
std::string sql = user_queue.front();
mysqlpp::Query query = c.query(sql);
if (!query.exec()) {
logger->info("User flush failed (" + std::to_string(user_queue.size()) + " remain)");
sleep(3);
break;
} else {
std::lock_guard<std::mutex> uq_lock(user_queue_lock);
user_queue.pop();
stats.user_queue_size -= 1;
}
}
catch (const mysqlpp::BadQuery &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush users with a qlength: " + std::to_string(user_queue.front().size()) + " queue size: " + std::to_string(user_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} catch (const mysqlpp::Exception &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush users with a qlength: " + std::to_string(user_queue.front().size()) + " queue size: " + std::to_string(user_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
}
}
void mysql::flush_torrents() {
std::unique_lock<std::mutex> q_lock(torrent_queue_lock);
size_t qsize = torrent_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Torrent flush queue size: {}, next query length: {}", qsize, torrent_queue.front().size());
}
catch (const mysqlpp::Exception &er) {
logger->error("MySQL error in flush_users: " + std::string(er.what()));
q_lock.unlock();
std::unique_lock<std::mutex> buf_lock(torrent_buffer_lock);
if (readonly) {
update_torrent_buffer.clear();
return;
}
if (update_torrent_buffer.empty() && qsize == 0) {
return;
}
if (!update_torrent_buffer.empty()) {
std::lock_guard<std::mutex> q_lock(torrent_queue_lock);
torrent_queue.push(fmt::format(
"INSERT INTO torrents_leech_stats"
"(TorrentID,Seeders,Leechers,Snatched,Balance)VALUES{}"
"ON DUPLICATE KEY UPDATE Seeders=VALUES(Seeders),Leechers=VALUES(Leechers)"
",Snatched=Snatched+VALUES(Snatched),Balance=VALUES(Balance)"
",last_action=IF(VALUES(Seeders)>0,now(),last_action)",
update_torrent_buffer.str()
));
update_torrent_buffer.clear();
}
buf_lock.unlock();
stats.torrent_queue_size = qsize + 1;
if (!torrent_flush_active) {
std::thread thread([this]() {
do_flush("torrent", torrent_queue, torrent_queue_lock, torrent_flush_active, stats.torrent_queue_size);
});
thread.detach();
}
u_active = false;
}
void mysql::do_flush_torrents() {
t_active = true;
try {
mysqlpp::Connection c = create_connection();
while (torrent_queue.size() > 0) {
try {
std::string sql = torrent_queue.front();
if (sql.empty()) {
torrent_queue.pop();
stats.torrent_queue_size -= 1;
continue;
}
mysqlpp::Query query = c.query(sql);
if (!query.exec()) {
logger->info("Torrent flush failed (" + std::to_string(torrent_queue.size()) + " remain)");
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} else {
std::lock_guard<std::mutex> tq_lock(torrent_queue_lock);
torrent_queue.pop();
stats.torrent_queue_size -= 1;
}
}
catch (const mysqlpp::BadQuery &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush torrents with a qlength: " + std::to_string(torrent_queue.front().size()) + " queue size: " + std::to_string(torrent_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} catch (const mysqlpp::Exception &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush torrents with a qlength: " + std::to_string(torrent_queue.front().size()) + " queue size: " + std::to_string(torrent_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
}
}
void mysql::flush_users() {
std::unique_lock<std::mutex> q_lock(user_queue_lock);
size_t qsize = user_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("User flush queue size: {}, next query length: {}", qsize, user_queue.front().size());
}
catch (const mysqlpp::Exception &er) {
logger->error("MySQL error in flush_torrents: " + std::string(er.what()));
q_lock.unlock();
std::unique_lock<std::mutex> buf_lock(user_buffer_lock);
if (readonly) {
update_user_buffer.clear();
return;
}
if (update_user_buffer.empty() && qsize == 0) {
return;
}
if (!update_user_buffer.empty()) {
std::lock_guard<std::mutex> q_lock(user_queue_lock);
user_queue.push(fmt::format(
"INSERT INTO users_leech_stats(UserID,Uploaded,Downloaded)VALUES{}"
"ON DUPLICATE KEY UPDATE Uploaded=Uploaded+VALUES(Uploaded)"
",Downloaded=Downloaded+VALUES(Downloaded)",
update_user_buffer.str()
));
update_user_buffer.clear();
stats.user_queue_size = qsize + 1;
}
buf_lock.unlock();
if (!user_flush_active) {
std::thread thread([this]() {
do_flush("user", user_queue, user_queue_lock, user_flush_active, stats.user_queue_size);
});
thread.detach();
}
t_active = false;
}
void mysql::do_flush_peers() {
p_active = true;
void mysql::do_flush(const char *name, std::queue<std::string> &queue, std::mutex &mtx, bool &active, std::atomic<uint32_t>& counter) {
active = true;
try {
mysqlpp::Connection c = create_connection();
while (peer_queue.size() > 0) {
try {
std::string sql = peer_queue.front();
mysqlpp::Query query = c.query(sql);
if (!query.exec()) {
logger->info("Peer flush failed (" + std::to_string(peer_queue.size()) + " remain)");
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} else {
std::lock_guard<std::mutex> pq_lock(peer_queue_lock);
peer_queue.pop();
stats.peer_queue_size -= 1;
}
}
catch (const mysqlpp::BadQuery &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush peers with a qlength: " + std::to_string(peer_queue.front().size()) + " queue size: " + std::to_string(peer_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} catch (const mysqlpp::Exception &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush peers with a qlength: " + std::to_string(peer_queue.front().size()) + " queue size: " + std::to_string(peer_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
}
}
}
catch (const mysqlpp::Exception &er) {
logger->error("MySQL error in flush_peers: " + std::string(er.what()));
}
p_active = false;
}
std::unique_lock<std::mutex> q_lock(mtx);
size_t q_size = queue.size();
q_lock.unlock();
void mysql::do_flush_snatches() {
s_active = true;
try {
mysqlpp::Connection c = create_connection();
while (snatch_queue.size() > 0) {
std::string sql;
while (q_size > 0) {
try {
std::string sql = snatch_queue.front();
std::unique_lock<std::mutex> q_lock(mtx);
sql = queue.front();
q_lock.unlock();
mysqlpp::Query query = c.query(sql);
if (!query.exec()) {
logger->info("Snatch flush failed (" + std::to_string(snatch_queue.size()) + " remain)");
logger->info("{} flush failed ({} remain)", name, q_size);
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} else {
std::lock_guard<std::mutex> sq_lock(snatch_queue_lock);
snatch_queue.pop();
stats.snatch_queue_size -= 1;
}
q_lock.lock();
queue.pop();
q_size = queue.size();
q_lock.unlock();
counter -= 1;
}
catch (const mysqlpp::BadQuery &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush snatches with a qlength: " + std::to_string(snatch_queue.front().size()) + " queue size: " + std::to_string(snatch_queue.size()));
logger->error("SQL error: {} in {} flush with a qlength: {} queue size: {} sql=[{}]",
er.what(), name, queue.front().size(), q_size, sql
);
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} catch (const mysqlpp::Exception &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush snatches with a qlength: " + std::to_string(snatch_queue.front().size()) + " queue size: " + std::to_string(snatch_queue.size()));
logger->error("DB error: {} in {} flush with a qlength: {} queue size: {}",
er.what(), name, queue.front().size(), q_size
);
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
}
}
}
catch (const mysqlpp::Exception &er) {
logger->error("MySQL error in flush_snatches: " + std::string(er.what()));
logger->error("General DB error in {} flush: {}", name, er.what());
}
s_active = false;
}
void mysql::do_flush_tokens() {
tok_active = true;
try {
mysqlpp::Connection c = create_connection();
while (token_queue.size() > 0) {
try {
std::string sql = token_queue.front();
mysqlpp::Query query = c.query(sql);
if (!query.exec()) {
logger->info("Token flush failed (" + std::to_string(token_queue.size()) + " remain)");
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} else {
std::lock_guard<std::mutex> tq_lock(token_queue_lock);
token_queue.pop();
stats.token_queue_size -= 1;
}
}
catch (const mysqlpp::BadQuery &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush tokens with a qlength: " + std::to_string(token_queue.front().size()) + " queue size: " + std::to_string(token_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
} catch (const mysqlpp::Exception &er) {
logger->error("Query error: " + std::string(er.what()) + " in flush tokens with a qlength: " + std::to_string(token_queue.front().size()) + " queue size: " + std::to_string(token_queue.size()));
std::this_thread::sleep_for(std::chrono::seconds(3));
break;
}
}
}
catch (const mysqlpp::Exception &er) {
logger->error("MySQL error in flush_tokens: " + std::string(er.what()));
}
tok_active = false;
active = false;
}
+63 -37
View File
@@ -16,13 +16,44 @@
class mysql {
private:
class query_buffer {
std::string buf;
const mysql &db;
public:
/* pre-extend the buffer to avoid repeated reallocs,
* the peer queue can easily reach 500Kb
*/
query_buffer(const mysql& db_):db(db_) { buf.reserve(57320); }
query_buffer& operator+=(const std::string& chars) {
if (!db.readonly) {
if (!buf.empty()) {
buf += ',';
}
buf += chars;
}
return *this;
}
void clear() { buf.clear(); buf.reserve(57320); }
bool empty() const { return buf.empty(); }
const std::string& str() const { return buf; }
};
friend class query_buffer;
mysqlpp::Connection conn;
std::string update_user_buffer;
std::string update_torrent_buffer;
std::string update_heavy_peer_buffer;
std::string update_light_peer_buffer;
std::string update_snatch_buffer;
std::string update_token_buffer;
query_buffer update_heavy_peer_buffer;
query_buffer update_light_peer_buffer;
query_buffer update_snatch_buffer;
query_buffer update_token_buffer;
query_buffer update_torrent_buffer;
query_buffer update_user_buffer;
std::mutex peer_buffer_lock;
std::mutex peer_light_buffer_lock;
std::mutex snatch_buffer_lock;
std::mutex token_buffer_lock;
std::mutex torrent_buffer_lock;
std::mutex user_buffer_lock;
std::queue<std::string> user_queue;
std::queue<std::string> torrent_queue;
@@ -30,19 +61,16 @@ class mysql {
std::queue<std::string> snatch_queue;
std::queue<std::string> token_queue;
std::string mysql_db, mysql_host, mysql_username, mysql_password;
unsigned int mysql_port;
bool u_active, t_active, p_active, s_active, tok_active;
bool readonly;
// These locks prevent more than one thread from reading/writing the buffers.
// These should be held for the minimum time possible.
std::mutex user_queue_lock;
std::mutex torrent_buffer_lock;
std::mutex torrent_queue_lock;
std::mutex peer_queue_lock;
std::mutex snatch_queue_lock;
std::mutex token_queue_lock;
std::mutex torrent_queue_lock;
std::mutex user_queue_lock;
std::string mysql_db, mysql_host, mysql_username, mysql_password;
unsigned int mysql_port;
bool peer_flush_active, snatch_flush_active, token_flush_active, torrent_flush_active, user_flush_active;
bool readonly;
std::shared_ptr<spdlog::logger> logger;
@@ -50,11 +78,7 @@ class mysql {
void load_tokens(torrent_list &torrents);
mysqlpp::Connection create_connection();
void do_flush_users();
void do_flush_torrents();
void do_flush_snatches();
void do_flush_peers();
void do_flush_tokens();
void do_flush(const char *name, std::queue<std::string> &queue, std::mutex &mtx, bool &active, std::atomic<uint32_t>& counter);
void flush_users();
void flush_torrents();
@@ -66,6 +90,10 @@ class mysql {
public:
bool verbose_flush;
std::mutex torrent_list_mutex;
std::mutex user_list_mutex;
std::mutex whitelist_mutex;
mysql(config * conf);
void reload_config(config * conf);
void flush();
@@ -74,27 +102,25 @@ class mysql {
void load_torrents(torrent_list &torrents);
void load_users(user_list &users);
void load_whitelist(std::vector<std::string> &whitelist);
std::string quote(std::string value);
// (id,uploaded_change,downloaded_change)
void record_user(const std::string &record);
// (uid,fid,active,peerid,useragent,ip,uploaded,downloaded,upspeed,downspeed,left,timespent,announces,tstamp)
void record_peer(const std::string &record);
// (fid,peerid,timespent,announces,tstamp)
void record_peer_light(const std::string &record);
// (uid,fid,tstamp)
void record_snatch(const std::string &record);
// (uid,fid,downloaded_change)
void record_token(const std::string &record);
// (id,seeders,leechers,snatched_change,balance)
void record_torrent(const std::string &record);
// (uid,fid,tstamp)
void record_snatch(const std::string &record, const std::string &ip);
// (uid,fid,active,peerid,useragent,ip,uploaded,downloaded,upspeed,downspeed,left,timespent,announces,tstamp)
void record_peer(const std::string &record, const std::string &ip, const std::string &peer_id, const std::string &useragent);
// (fid,peerid,timespent,announces,tstamp)
void record_peer(const std::string &record, const std::string &peer_id);
void record_token(const std::string &record);
std::mutex torrent_list_mutex;
std::mutex user_list_mutex;
std::mutex whitelist_mutex;
// (id,uploaded_change,downloaded_change)
void record_user(const std::string &record);
};
#pragma GCC visibility pop
+8 -8
View File
@@ -54,7 +54,7 @@ void connection_mother::reload_config(config * conf) {
unsigned int old_max_connections = max_connections;
load_config(conf);
if (old_listen_port != listen_port) {
logger->info("Changing listen port from " + std::to_string(old_listen_port) +" to " + std::to_string(listen_port));
logger->info("Changing listen port from {} to {}", old_listen_port, listen_port);
int new_listen_socket = create_listen_socket();
if (new_listen_socket != 0) {
listen_event.stop();
@@ -77,7 +77,7 @@ int connection_mother::create_listen_socket() {
// Stop old sockets from hogging the port
int yes = 1;
if (setsockopt(new_listen_socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) {
logger->error("Could not reuse socket: " + std::string(strerror(errno)));
logger->error("Could not reuse socket: {}", strerror(errno));
return 0;
}
@@ -88,24 +88,24 @@ int connection_mother::create_listen_socket() {
// Bind
if (bind(new_listen_socket, (sockaddr *) &address, sizeof(address)) == -1) {
logger->error("Bind failed: " + std::string(strerror(errno)));
logger->error("Bind failed: {}", strerror(errno));
return 0;
}
// Listen
if (listen(new_listen_socket, max_connections) == -1) {
logger->error("Listen failed: " + std::string(strerror(errno)));
logger->error("Listen failed: {}", strerror(errno));
return 0;
}
// Set non-blocking
int flags = fcntl(new_listen_socket, F_GETFL);
if (flags == -1) {
logger->error("Could not get socket flags: " + std::string(strerror(errno)));
logger->error("Could not get socket flags: {}", strerror(errno));
return 0;
}
if (fcntl(new_listen_socket, F_SETFL, flags | O_NONBLOCK) == -1) {
logger->error("Could not set non-blocking: " + std::string(strerror(errno)));
logger->error("Could not set non-blocking: {}", strerror(errno));
return 0;
}
@@ -113,7 +113,7 @@ int connection_mother::create_listen_socket() {
}
void connection_mother::run() {
logger->info("Sockets up on port " + std::to_string(listen_port) + ", starting event loop!");
logger->info("Sockets up on port {}, starting event loop!", listen_port);
ev_loop(ev_default_loop(0), 0);
}
@@ -146,7 +146,7 @@ connection_middleman::connection_middleman(int &listen_socket, worker * new_work
auto logger = spdlog::get("logger");
connect_sock = accept(listen_socket, NULL, NULL);
if (connect_sock == -1) {
logger->error("Accept failed, errno " + std::to_string(errno) + ": " + std::string(strerror(errno)));
logger->error("Accept failed, errno {}: {}", errno, strerror(errno));
delete this;
return;
}
-46
View File
@@ -6,29 +6,6 @@
#include "misc_functions.h"
int32_t strtoint32(const std::string& str) {
std::istringstream stream(str);
int32_t i = 0;
stream >> i;
return i;
}
int64_t strtoint64(const std::string& str) {
std::istringstream stream(str);
int64_t i = 0;
stream >> i;
return i;
}
std::string inttostr(const int i) {
std::string str;
std::stringstream out;
out << i;
str = out.str();
return str;
}
std::string hex_decode(const std::string &in) {
std::string out;
out.reserve(20);
@@ -60,26 +37,3 @@ std::string hex_decode(const std::string &in) {
}
return out;
}
std::string bintohex(const std::string &in) {
std::string out;
size_t length = in.length();
out.reserve(2*length);
for (unsigned int i = 0; i < length; i++) {
unsigned char x = static_cast<unsigned char>((in[i] & 0xF0) >> 4);
if (x > 9) {
x += 'a' - 10;
} else {
x += '0';
}
out.push_back(x);
x = in[i] & 0x0F;
if (x > 9) {
x += 'a' - 10;
} else {
x += '0';
}
out.push_back(x);
}
return out;
}
-4
View File
@@ -5,10 +5,6 @@
#include <string>
int32_t strtoint32(const std::string& str);
int64_t strtoint64(const std::string& str);
std::string inttostr(int i);
std::string hex_decode(const std::string &in);
std::string bintohex(const std::string &in);
#endif // SRC_MISC_FUNCTIONS_H_
+2 -2
View File
@@ -5,9 +5,9 @@
#define OCELOT_VERSION_MAJOR 2
#define OCELOT_VERSION_MINOR 1
#define OCELOT_VERSION_BUGFIX 3
#define OCELOT_VERSION_BUGFIX 4
#define OCELOT_VERSION_NREV 0
#define OCELOT_VERSION "2.1.3"
#define OCELOT_VERSION "2.1.4"
#include <time.h>
+1 -1
View File
@@ -62,7 +62,7 @@ std::string report_jemalloc_plain(const char *opts, const std::string path) {
if (dump_jemalloc(filename.c_str(), opts) != 0) {
// return an empty string if something went bananas
unlink(filename.c_str());
return std::string("");
return std::string();
}
// slurp the contents into a string
+12 -5
View File
@@ -1,5 +1,7 @@
// Copyright [2017-2024] Orpheus
#include <fmt/format.h>
#include <algorithm>
#include <sstream>
@@ -14,9 +16,12 @@ const std::string http_head(size_t content_length, client_opts_t &client_opts) {
const std::string content_type = client_opts.html ? "text/html" : "text/plain";
// rely on preprocessor string concatenation
std::string head = "HTTP/1.1 200 OK\r\nServer: Ocelot " OCELOT_VERSION "\r\n"
"Content-Type: " + content_type + "\r\n"
"Content-Length: " + inttostr(content_length) + "\r\n";
std::string head(fmt::format(
"HTTP/1.1 200 OK\r\nServer: Ocelot " OCELOT_VERSION "\r\n"
"Content-Type: {}\r\n"
"Content-Length: {}\r\n",
content_type, content_length
));
if (client_opts.gzip) {
head += "Content-Encoding: gzip\r\n";
}
@@ -51,8 +56,10 @@ const std::string http_response(const std::string &body, client_opts_t &client_o
const std::string error(const std::string &message, client_opts_t &client_opts) {
return http_response(
"d14:failure reason" + inttostr(message.length()) + ':' + message
+ "12:min intervali5400e8:intervali5400ee",
fmt::format(
"d14:failure reason{}:{}12:min intervali5400e8:intervali5400ee",
message.length(), message
),
client_opts
);
}
+4 -3
View File
@@ -34,9 +34,10 @@ void schedule::handle(ev::timer &watcher, int events_flags) {
stats.connection_rate = (stats.opened_connections - last_opened_connections) / cur_schedule_interval;
stats.request_rate = (stats.requests - last_request_count) / cur_schedule_interval;
if (counter % 20 == 0) {
logger->info(std::to_string(stats.open_connections) + " open, "
+ std::to_string(stats.opened_connections) + " connections (" + std::to_string(stats.connection_rate) + "/s), "
+ std::to_string(stats.requests) + " requests (" + std::to_string(stats.request_rate) + "/s)");
logger->info("{} open, {} connections ({}/s), {} requests ({}/s)",
stats.open_connections, stats.opened_connections, stats.connection_rate,
stats.requests, stats.request_rate
);
}
if (work->get_status() == CLOSING && db->all_clear() && sc->all_clear()) {
+6 -6
View File
@@ -65,7 +65,7 @@ void site_comm::flush_tokens()
std::lock_guard<std::mutex> lock(expire_queue_lock);
size_t qsize = token_queue.size();
if (verbose_flush || qsize > 0) {
logger->info("Token expire queue size: " + std::to_string(qsize));
logger->info("Token expire queue size: {}", qsize);
}
if (expire_token_buffer.empty()) {
return;
@@ -104,9 +104,9 @@ void site_comm::do_flush_tokens()
std::ostream request_stream(&request);
request_stream << "GET " << site_path << "/tools.php?key=" << site_password
<< "&type=expiretoken&action=ocelot&tokens=" << token_queue.front() << " HTTP/1.0\r\n"
<< "Host: " << site_host << "\r\n"
<< "Accept: */*\r\n"
<< "Connection: close\r\n\r\n";
"Host: " << site_host << "\r\n"
"Accept: */*\r\n"
"Connection: close\r\n\r\n";
boost::asio::write(socket, request);
@@ -130,13 +130,13 @@ void site_comm::do_flush_tokens()
std::lock_guard<std::mutex> lock(expire_queue_lock);
token_queue.pop();
} else {
logger->error("Response returned with status code " + std::to_string(status_code) + " when trying to expire a token!");
logger->error("Response returned with status code {} when trying to expire a token!", status_code);
// TODO: add exponential backoff
std::this_thread::sleep_for(std::chrono::seconds(5));
}
}
} catch (std::exception &er) {
logger->error("Exception: " + std::string(er.what()));
logger->error("Exception: {}", er.what());
}
t_active = false;
}
+239 -134
View File
@@ -1,6 +1,7 @@
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <fmt/format.h>
#include <iostream>
#include <string>
@@ -84,22 +85,13 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
stats.http_error++;
return error("GET string too short", client_opts);
}
size_t pos = 5; // skip 'GET /'
// Get the passkey
std::string passkey;
passkey.reserve(32);
if (input[37] != '/') {
stats.http_error++;
return error("Malformed announce", client_opts);
}
for (; pos < 37; pos++) {
passkey.push_back(input[pos]);
}
pos = 38;
// Get the passkey: skip the 5 chars of 'GET /' and take the next 32
std::string passkey(input.substr(5, 32));
// Get the action
enum action_t {
@@ -107,6 +99,7 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
};
action_t action = INVALID;
size_t pos = 38;
switch (input[pos]) {
case 'a':
stats.announcements++;
@@ -327,7 +320,10 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
}
const std::string bencode_warning(const std::string &message) {
return "15:warning message" + inttostr(message.length()) + ':' + message;
return fmt::format(
"15:warning message{}:{}",
message.length(), message
);
}
std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u, params_type &params, params_type &headers, std::string &ip, client_opts_t &client_opts) {
@@ -365,10 +361,52 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
}
wl_lock.unlock();
int64_t left = std::max((int64_t)0, strtoint64(params["left"]));
int64_t uploaded = std::max((int64_t)0, strtoint64(params["uploaded"]));
int64_t downloaded = std::max((int64_t)0, strtoint64(params["downloaded"]));
int64_t corrupt = std::max((int64_t)0, strtoint64(params["corrupt"]));
int64_t uploaded = 0;
int64_t downloaded = 0;
int64_t corrupt = 0;
int64_t left = 0;
{
// local scope for conversions
params_type::const_iterator p;
size_t uploaded_convert = 0;
size_t downloaded_convert = 0;
size_t corrupt_convert = 0;
size_t left_convert = 0;
try {
p = params.find("uploaded");
if (p != params.end()) {
uploaded = std::stoull(p->second, &uploaded_convert, 10);
}
p = params.find("downloaded");
if (p != params.end()) {
downloaded = std::stoull(p->second, &downloaded_convert, 10);
}
p = params.find("corrupt");
if (p != params.end()) {
corrupt = std::stoull(p->second, &corrupt_convert, 10);
}
p = params.find("left");
if (p != params.end()) {
left = std::stoull(p->second, &left_convert, 10);
}
} catch (std::invalid_argument &er) {
return error(
fmt::format(
"corrupt parameter in announce u{}/d{}/c{}/l{}",
uploaded_convert, downloaded_convert, corrupt_convert, left_convert
),
client_opts
);
} catch (std::out_of_range &er) {
return error(
fmt::format(
"value out of range in announce u{}/d{}/c{}/l{}",
uploaded_convert, downloaded_convert, corrupt_convert, left_convert
),
client_opts
);
}
}
int snatched = 0; // This is the value that gets sent to the database on a snatch
int active = 1; // This is the value that marks a peer as active/inactive in the database
@@ -382,13 +420,6 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
bool inc_l = false, inc_s = false, dec_l = false, dec_s = false;
userid_t userid = u->get_id();
// "Randomize" the element order in the peer map by prefixing with a peer id byte
std::stringstream peer_key_stream;
peer_key_stream << peer_id[12 + (tor.id & 7)]
<< userid // Include user id in the key to lower chance of peer id collisions
<< peer_id;
const std::string peer_key(peer_key_stream.str());
if (params["event"] == "completed") {
// Don't update <snatched> here as we may decide to use other conditions later on
completed_torrent = (left == 0); // Sanity check just to be extra safe
@@ -398,7 +429,14 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
update_torrent = true;
active = 0;
}
peer * p;
// "Randomize" the element order in the peer map by prefixing with a peer id byte
// Include user id in the key to lower chance of peer id collisions
const std::string peer_key = fmt::format(
"{}{}{}",
peer_id[12 + (tor.id & 7)], userid, peer_id
);
peer *p;
peer_list::iterator peer_it;
// Insert/find the peer in the torrent list
if (left > 0) {
@@ -505,19 +543,19 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
} else if (tor.free_torrent == FREE || sit != tor.tokened_users.end()) {
if (sit != tor.tokened_users.end()) {
expire_token = true;
std::stringstream record;
record << '(' << userid << ',' << tor.id << ',' << downloaded_change << ')';
std::string record_str = record.str();
db->record_token(record_str);
db->record_token(fmt::format(
"({},{},{})",
userid, tor.id, downloaded_change
));
}
downloaded_change = 0;
}
if (uploaded_change || downloaded_change) {
std::stringstream record;
record << '(' << userid << ',' << uploaded_change << ',' << downloaded_change << ')';
std::string record_str = record.str();
db->record_user(record_str);
db->record_user(fmt::format(
"({},{},{})",
userid, uploaded_change, downloaded_change
));
}
}
}
@@ -540,12 +578,12 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
}
}
uint16_t port = strtoint32(params["port"]) & 0xFFFF;
uint16_t port = atoi(params["port"].c_str()) & 0xFFFF;
// Generate compact ip/port string
if (inserted || port != p->port || ip != p->ip) {
p->port = port;
p->ip = ip;
p->ip_port = "";
p->ip_port.clear();
char x = 0;
for (size_t pos = 0, end = ip.length(); pos < end; pos++) {
if (ip[pos] == '.') {
@@ -579,19 +617,19 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
// Add peer data to the database
std::stringstream record;
if (peer_changed) {
record << '(' << userid << ',' << tor.id << ',' << active << ',' << uploaded << ',' << downloaded << ',' << upspeed << ',' << downspeed << ',' << left << ',' << corrupt << ',' << (cur_time - p->first_announced) << ',' << p->announces << ',';
std::string record_str = record.str();
std::string record_ip;
if (u->is_protected()) {
record_ip = "";
} else {
record_ip = ip;
}
db->record_peer(record_str, record_ip, peer_id, headers["user-agent"]);
db->record_peer(fmt::format(
"({},{},{},{},{},{},{}" ",{},{},{},{}" ",{},{}" ",{},{})",
userid, tor.id, active, uploaded, downloaded, upspeed, downspeed,
left, corrupt, cur_time - p->first_announced, p->announces,
u->is_protected() ? "'0.0.0.0'" : db->quote(ip), db->quote(peer_id),
db->quote(headers["user-agent"].substr(0, 51)), time(NULL)
));
} else {
record << '(' << userid << ',' << tor.id << ',' << (cur_time - p->first_announced) << ',' << p->announces << ',';
std::string record_str = record.str();
db->record_peer(record_str, peer_id);
db->record_peer_light(fmt::format(
"({},{},{},{}" ",{},{})",
userid, tor.id, cur_time - p->first_announced, p->announces,
db->quote(peer_id), time(NULL)
));
}
// Select peers!
@@ -600,7 +638,7 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
if (param_numwant == params.end()) {
numwant = numwant_limit;
} else {
numwant = std::min((int32_t)numwant_limit, strtoint32(param_numwant->second));
numwant = std::min(numwant_limit, atoi(param_numwant->second.c_str()));
}
if (stopped_torrent) {
@@ -615,16 +653,10 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
update_torrent = true;
tor.completed++;
std::stringstream record;
std::string record_ip;
if (u->is_protected()) {
record_ip = "";
} else {
record_ip = ip;
}
record << '(' << userid << ',' << tor.id << ',' << cur_time;
std::string record_str = record.str();
db->record_snatch(record_str, record_ip);
db->record_snatch(fmt::format(
"({},{},{},{})",
userid, tor.id, cur_time, u->is_protected() ? "'0.0.0.0'" : db->quote(ip)
));
// User is a seeder now!
if (!inserted) {
@@ -762,36 +794,35 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
// Putting this after the peer deletion gives us accurate swarm sizes
if (update_torrent || tor.last_flushed + 3600 < cur_time) {
db->record_torrent(fmt::format(
"({},{},{},{},{})",
tor.id, tor.seeders.size(), tor.leechers.size(), snatched, tor.balance
));
tor.last_flushed = cur_time;
std::stringstream record;
record << '(' << tor.id << ',' << tor.seeders.size() << ',' << tor.leechers.size() << ',' << snatched << ',' << tor.balance << ')';
std::string record_str = record.str();
db->record_torrent(record_str);
}
if (!u->can_leech() && left > 0) {
return error("Access denied, leeching forbidden", client_opts);
}
std::string output = "d8:completei";
std::string output;
output.reserve(350);
output += inttostr(tor.seeders.size());
output += "e10:downloadedi";
output += inttostr(tor.completed);
output += "e10:incompletei";
output += inttostr(tor.leechers.size());
output += "e8:intervali";
output += inttostr(announce_interval + jitter(randgen));
output += "e12:min intervali";
output += inttostr(announce_interval);
output += "e5:peers";
output = fmt::format(
"d8:completei{}"
"e10:downloadedi{}"
"e10:incompletei{}"
"e8:intervali{}"
"e12:min intervali{}",
tor.seeders.size(),
tor.completed,
tor.leechers.size(),
announce_interval + jitter(randgen),
announce_interval
);
if (peers.length() == 0) {
output += "0:";
output += "e5:peers0:";
} else {
output += inttostr(peers.length());
output += ":";
output += peers;
output += fmt::format("e5:peers{}:{}", peers.length(), peers);
}
if (invalid_ip) {
output += bencode_warning("Illegal character found in IP address. IPv6 is not supported");
@@ -811,8 +842,7 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
std::string worker::scrape(const std::list<std::string> &infohashes, params_type &headers, client_opts_t &client_opts) {
std::string output = "d5:filesd";
for (std::list<std::string>::const_iterator i = infohashes.begin(); i != infohashes.end(); ++i) {
std::string infohash = *i;
infohash = hex_decode(infohash);
std::string infohash(hex_decode(*i));
torrent_list::iterator tor = torrents_list.find(infohash);
if (tor == torrents_list.end()) {
@@ -820,16 +850,16 @@ std::string worker::scrape(const std::list<std::string> &infohashes, params_type
}
torrent *t = &(tor->second);
output += inttostr(infohash.length());
output += ':';
output += infohash;
output += "d8:completei";
output += inttostr(t->seeders.size());
output += "e10:incompletei";
output += inttostr(t->leechers.size());
output += "e10:downloadedi";
output += inttostr(t->completed);
output += "ee";
output += fmt::format(
"{}:{}"
"d8:completei{}"
"e10:incompletei{}"
"e10:downloadedi{}ee",
infohash.length(), infohash,
t->seeders.size(),
t->leechers.size(),
t->completed
);
}
output += "ee";
if (headers["accept-encoding"].find("gzip") != std::string::npos) {
@@ -847,24 +877,30 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
auto u = users_list.find(oldpasskey);
if (u == users_list.end()) {
logger->warn("No user with passkey " + oldpasskey + " exists when attempting to change passkey to " + newpasskey);
logger->warn(
"user {} not found when attempting to change passkey to {}",
oldpasskey, newpasskey
);
} else {
users_list[newpasskey] = u->second;
users_list.erase(oldpasskey);
logger->info("Changed passkey from " + oldpasskey + " to " + newpasskey + " for user " + std::to_string(u->second->get_id()));
logger->info(
"user {} changed passkey from {} to {}",
u->second->get_id(), oldpasskey, newpasskey
);
}
} else if (action == "add_torrent") {
torrent *t;
std::string info_hash = params["info_hash"];
info_hash = hex_decode(info_hash);
std::string info_hash_hex(params["info_hash"]);
std::string info_hash(hex_decode(info_hash_hex));
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
auto i = torrents_list.find(info_hash);
if (i == torrents_list.end()) {
t = &torrents_list[info_hash];
t->id = static_cast<torid_t>(strtoint32(params["id"]));
t->id = static_cast<torid_t>(atoi(params["id"].c_str()));
t->balance = 0;
t->completed = 0;
t->last_selected_seeder = "";
t->last_selected_seeder.clear();
} else {
t = &i->second;
}
@@ -875,10 +911,13 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
} else {
t->free_torrent = NEUTRAL;
}
logger->info("Added torrent " + std::to_string(t->id) + ". FL: " + std::to_string(t->free_torrent) + " " + params["freetorrent"]);
logger->info(
"added torrent {} with FL {} {}",
t->id, t->free_torrent, params["freetorrent"]
);
} else if (action == "update_torrent") {
std::string info_hash = params["info_hash"];
info_hash = hex_decode(info_hash);
std::string info_hash_hex(params["info_hash"]);
std::string info_hash(hex_decode(info_hash_hex));
freetype fl;
if (params["freetorrent"] == "0") {
fl = NORMAL;
@@ -891,9 +930,15 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
auto torrent_it = torrents_list.find(info_hash);
if (torrent_it != torrents_list.end()) {
torrent_it->second.free_torrent = fl;
logger->info("Updated torrent " + std::to_string(torrent_it->second.id) + " to FL " + std::to_string(fl));
logger->info(
"modify torrent {} to FL {}",
torrent_it->second.id, fl
);
} else {
logger->warn("Failed to find torrent " + std::string(info_hash) + " to FL " + std::to_string(fl));
logger->warn(
"failed to find torrent {} to FL {}",
info_hash_hex, fl
);
}
} else if (action == "update_torrents") {
// Each decoded infohash is exactly 20 characters long.
@@ -913,34 +958,48 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
auto torrent_it = torrents_list.find(info_hash);
if (torrent_it != torrents_list.end()) {
torrent_it->second.free_torrent = fl;
logger->info("Updated torrent " + std::to_string(torrent_it->second.id) + " to FL " + std::to_string(fl));
logger->info(
"bulk modify torrent {} to FL {}",
torrent_it->second.id, fl
);
} else {
logger->warn("Failed to find torrent " + info_hash + " to FL " + std::to_string(fl));
logger->warn(
"bulk modify cannot find torrent {} to FL {}",
info_hash, fl
);
}
}
} else if (action == "add_token") {
std::string info_hash = hex_decode(params["info_hash"]);
std::string info_hash_hex(params["info_hash"]);
std::string info_hash(hex_decode(info_hash_hex));
int userid = atoi(params["userid"].c_str());
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
auto torrent_it = torrents_list.find(info_hash);
if (torrent_it != torrents_list.end()) {
torrent_it->second.tokened_users.insert(userid);
} else {
logger->warn("Failed to find torrent to add a token for user " + std::to_string(userid));
logger->warn(
"user {} failed to find torrent {} to add token",
userid, info_hash_hex
);
}
} else if (action == "remove_token") {
std::string info_hash = hex_decode(params["info_hash"]);
std::string info_hash_hex(params["info_hash"]);
std::string info_hash(hex_decode(info_hash_hex));
int userid = atoi(params["userid"].c_str());
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
auto torrent_it = torrents_list.find(info_hash);
if (torrent_it != torrents_list.end()) {
torrent_it->second.tokened_users.erase(userid);
} else {
logger->warn("Failed to find torrent " + info_hash + " to remove token for user " + std::to_string(userid));
logger->warn(
"user {} failed to find torrent {} to remove token",
userid, info_hash_hex
);
}
} else if (action == "delete_torrent") {
std::string info_hash = params["info_hash"];
info_hash = hex_decode(info_hash);
std::string info_hash_hex(params["info_hash"]);
std::string info_hash(hex_decode(info_hash_hex));
int reason = -1;
auto reason_it = params.find("reason");
if (reason_it != params.end()) {
@@ -948,8 +1007,12 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
}
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
auto torrent_it = torrents_list.find(info_hash);
if (torrent_it != torrents_list.end()) {
logger->info("Deleting torrent " + std::to_string(torrent_it->second.id) + " for the reason '" + get_del_reason(reason) + "'");
if (torrent_it == torrents_list.end()) {
logger->warn(
"torrent {} not found to delete",
info_hash_hex
);
} else {
stats.leechers -= torrent_it->second.leechers.size();
stats.seeders -= torrent_it->second.seeders.size();
for (auto &p : torrent_it->second.leechers) {
@@ -964,29 +1027,40 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
msg.time = time(NULL);
del_reasons[info_hash] = msg;
torrents_list.erase(torrent_it);
} else {
logger->warn("Failed to find torrent " + bintohex(info_hash) + " to delete ");
logger->info(
"torrent {} deleted for reason '{}'",
torrent_it->second.id, get_del_reason(reason)
);
}
} else if (action == "add_user") {
std::string passkey = params["passkey"];
userid_t userid = strtoint32(params["id"]);
userid_t userid = atoi(params["id"].c_str());
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
auto u = users_list.find(passkey);
if (u == users_list.end()) {
bool protect_ip = params["visible"] == "0";
user_ptr tmp_user = std::make_shared<user>(userid, true, protect_ip);
users_list.insert(std::pair<std::string, user_ptr>(passkey, tmp_user));
logger->info("Added user " + passkey + " with id " + std::to_string(userid));
logger->info(
"added user {} via {}",
userid, passkey
);
} else {
logger->warn("Tried to add already known user " + passkey + " with id " + std::to_string(userid));
u->second->set_deleted(false);
logger->warn(
"ignored add user {} via {}",
userid, passkey
);
}
} else if (action == "remove_user") {
std::string passkey = params["passkey"];
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
auto u = users_list.find(passkey);
if (u != users_list.end()) {
logger->info("Removed user " + passkey + " with id " + std::to_string(u->second->get_id()));
logger->info(
"removed user {} via {}",
u->second->get_id(), passkey
);
u->second->set_deleted(true);
users_list.erase(u);
}
@@ -998,7 +1072,10 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
std::string passkey = passkeys.substr(pos, 32);
auto u = users_list.find(passkey);
if (u != users_list.end()) {
logger->info("Removed user " + passkey);
logger->info(
"bulk remove user {}",
passkey
);
u->second->set_deleted(true);
users_list.erase(passkey);
}
@@ -1014,19 +1091,28 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
protect_ip = true;
}
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
user_list::iterator i = users_list.find(passkey);
if (i == users_list.end()) {
logger->warn("No user with passkey " + passkey + " found when attempting to change leeching status!");
user_list::iterator u = users_list.find(passkey);
if (u == users_list.end()) {
logger->warn(
"user {} not found for update",
passkey
);
} else {
i->second->set_protected(protect_ip);
i->second->set_leechstatus(can_leech);
logger->info("Updated user " + passkey);
u->second->set_protected(protect_ip);
u->second->set_leechstatus(can_leech);
logger->info(
"updated user {} via {} leech={} protect_ip={}",
u->second->get_id(), passkey, can_leech, protect_ip
);
}
} else if (action == "add_whitelist") {
std::string peer_id = params["peer_id"];
std::lock_guard<std::mutex> wl_lock(db->whitelist_mutex);
whitelist.push_back(peer_id);
logger->info("Whitelisted " + peer_id);
logger->info(
"whitelisted client {}",
peer_id
);
} else if (action == "remove_whitelist") {
std::string peer_id = params["peer_id"];
std::lock_guard<std::mutex> wl_lock(db->whitelist_mutex);
@@ -1036,7 +1122,10 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
break;
}
}
logger->info("De-whitelisted " + peer_id);
logger->info(
"dewhitelisted cient {}",
peer_id
);
} else if (action == "edit_whitelist") {
std::string new_peer_id = params["new_peer_id"];
std::string old_peer_id = params["old_peer_id"];
@@ -1048,24 +1137,35 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
}
}
whitelist.push_back(new_peer_id);
logger->info("Edited whitelist item from " + old_peer_id + " to " + new_peer_id);
logger->info(
"modified whitelist client from {} to {}",
old_peer_id, new_peer_id
);
} else if (action == "update_announce_interval") {
uint32_t prev = conf->get_uint("announce_interval");
const std::string interval = params["new_announce_interval"];
conf->set("announce_interval", interval);
announce_interval = conf->get_uint("announce_interval");
logger->info("Edited announce interval to " + std::to_string(announce_interval));
logger->info(
"modified announce interval from {} to {}",
prev, announce_interval
);
} else if (action == "update_announce_jitter") {
uint32_t prev = conf->get_uint("announce_jitter");
const std::string new_jitter = params["new_announce_jitter"];
conf->set("announce_jitter", new_jitter);
jitter = std::uniform_int_distribution<int>(0, conf->get_uint("announce_jitter"));
logger->info("Edited announce jitter to " + std::to_string(conf->get_uint("announce_jitter")));
logger->info(
"modified announce jitter from {} to {}",
prev, conf->get_uint("announce_jitter")
);
} else if (action == "info_torrent") {
std::stringstream output;
std::string info_hash_hex = params["info_hash"];
std::string info_hash = hex_decode(info_hash_hex);
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
auto torrent_it = torrents_list.find(info_hash);
output << "{\"hash\":" << std::string(info_hash_hex);
std::ostringstream output;
output << "{\"hash\":" << info_hash_hex;
if (torrent_it != torrents_list.end()) {
output << ",\"id\":" << std::to_string(torrent_it->second.id)
<< ",\"free\":" << std::to_string(torrent_it->second.free_torrent);
@@ -1144,8 +1244,10 @@ void worker::reap_peers() {
stats.leechers -= reaped_l;
stats.seeders -= reaped_s;
}
logger->info("Reaped " + std::to_string(reaped_l) + " leechers and " + std::to_string(reaped_s)
+ " seeders. Reset " + std::to_string(cleared_torrents) + " torrents");
logger->info(
"Reaped {} leechers and {} seeders. Reset {} torrents",
reaped_l, reaped_s, cleared_torrents
);
}
void worker::reap_del_reasons() {
@@ -1163,7 +1265,10 @@ void worker::reap_del_reasons() {
}
++it;
}
logger->info("Reaped " + std::to_string(reaped) + " del reasons");
logger->info(
"Reaped {} del reasons",
reaped
);
}
std::string worker::get_del_reason(int code) {
@@ -1238,7 +1343,7 @@ std::string worker::get_del_reason(int code) {
return "Audience Recording";
break;
default:
return "";
return std::string();
break;
}
}
+1 -1
View File
@@ -39,7 +39,7 @@ class worker {
unsigned int announce_interval;
unsigned int del_reason_lifetime;
unsigned int peers_timeout;
unsigned int numwant_limit;
int numwant_limit;
bool keepalive_enabled;
std::string site_password;
std::string report_password;