mirror of
https://github.com/OPSnet/Ocelot.git
synced 2026-11-07 17:39:02 -10:00
Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
021df7fc7a | ||
|
|
0c8b803b1e | ||
|
|
d4aaff7bb5 | ||
|
|
37685bd8da |
No files matched your search
@@ -1,3 +1,28 @@
|
||||
-- 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
|
||||
FEATURES
|
||||
* Peak number of simultaneous open connections is now tracked.
|
||||
* Maximum size of client http request is now tracked.
|
||||
* Number of errors (bad tracker secrets, bad client requests) is now tracked.
|
||||
* Show jemalloc stats in Gazelle and Grafana report endpoints.
|
||||
BUILD
|
||||
* refactor report functions to produce only report payload
|
||||
|
||||
-- 2.1.2 (2024-02-17)
|
||||
FEATURES
|
||||
* show resource usage in Gazelle and Grafana report endpoints
|
||||
|
||||
-- 2.1.1 (2024-02-15)
|
||||
BUILD
|
||||
* Detabbed source code in first step towards having the code lint cleanly, no code changes
|
||||
|
||||
-- 2.1.0 (2024-02-11)
|
||||
FEATURES
|
||||
* announce jitter is now a uniform distribution and can be modified on the fly
|
||||
|
||||
@@ -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
|
||||
)
|
||||
@@ -10,6 +10,7 @@ RUN apt-get update \
|
||||
libboost-iostreams-dev \
|
||||
libboost-system-dev \
|
||||
libev-dev \
|
||||
libfmt-dev \
|
||||
libjemalloc-dev \
|
||||
libmysql++-dev \
|
||||
netcat-traditional \
|
||||
@@ -27,6 +28,7 @@ RUN cmake -Wno-dev -S /srv -B /srv/build \
|
||||
pkg-config \
|
||||
&& apt-get autoremove -y \
|
||||
&& apt-get clean -y \
|
||||
&& mkdir -p /tmp/ocelot \
|
||||
&& mv /srv/build/ocelot /srv/ocelot \
|
||||
&& mv /srv/ocelot.conf.dist /srv/ocelot.conf
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -69,6 +71,10 @@ docker run -v $(pwd)/ocelot.conf:/srv/ocelot.conf ocelot
|
||||
* `-c <path/to/ocelot.conf>` - Path to config file. If unspecified, the current working directory is used.
|
||||
* `-v` - Print queue status every time a flush is initiated.
|
||||
|
||||
You can run a test ocelot daemon alongside a production daemon by specifying
|
||||
a separate configuration file, and setting `readonly = true`. This will
|
||||
prevent the database peer tables from being reset.
|
||||
|
||||
### Signals
|
||||
|
||||
* `SIGHUP` - Reload config
|
||||
|
||||
@@ -34,4 +34,9 @@ schedule_interval = 3
|
||||
log = false
|
||||
log_path = /tmp/ocelot
|
||||
|
||||
# the report path should be placed on a tmpfs partition in production
|
||||
report_path = /tmp/ocelot
|
||||
|
||||
# set to true if prevent peer data from being zeroed out on startup
|
||||
# useful to test a new version alongside an instance running in production
|
||||
readonly = false
|
||||
+5
-3
@@ -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;
|
||||
}
|
||||
@@ -100,6 +100,8 @@ void config::init() {
|
||||
add("log", false);
|
||||
add("log_path", "ocelot"); // path to where to write log + name of log (don't need to put file extension)
|
||||
|
||||
add("report_path", "/tmp"); // path for transfer to website (should be a tmpfs in production)
|
||||
|
||||
// Debugging
|
||||
add("readonly", false);
|
||||
}
|
||||
@@ -107,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;
|
||||
@@ -150,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
@@ -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
@@ -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;
|
||||
}
|
||||
@@ -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
|
||||
|
||||
+17
-9
@@ -12,6 +12,8 @@
|
||||
#include "response.h"
|
||||
#include "events.h"
|
||||
|
||||
std::mutex peak_open_mutex;
|
||||
|
||||
// Define the connection mother (first half) and connection middlemen (second half)
|
||||
|
||||
//---------- Connection mother - spawns middlemen and lets them deal with the connection
|
||||
@@ -52,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();
|
||||
@@ -75,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;
|
||||
}
|
||||
|
||||
@@ -86,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;
|
||||
}
|
||||
|
||||
@@ -111,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);
|
||||
}
|
||||
|
||||
@@ -119,7 +121,13 @@ void connection_mother::handle_connect(ev::io &watcher, int events_flags) {
|
||||
// Spawn a new middleman
|
||||
if (stats.open_connections < max_middlemen) {
|
||||
stats.opened_connections++;
|
||||
stats.open_connections++;
|
||||
{
|
||||
const std::lock_guard<std::mutex> lock(peak_open_mutex);
|
||||
unsigned int current_open = stats.open_connections++;
|
||||
if (stats.peak_connections < current_open) {
|
||||
stats.peak_connections = current_open;
|
||||
}
|
||||
}
|
||||
new connection_middleman(listen_socket, work, this);
|
||||
}
|
||||
}
|
||||
@@ -138,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;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,200 @@
|
||||
#include <ctype.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <string.h>
|
||||
|
||||
#include "jemalloc_parse.h"
|
||||
|
||||
/* parse a size_t decimal text representation */
|
||||
const char *parse_sz(const char *in, const char *find, size_t *target) {
|
||||
const char *p = strstr(in, find);
|
||||
if (!p) {
|
||||
// didn't find what we were looking for
|
||||
return NULL;
|
||||
}
|
||||
p += strlen(find);
|
||||
|
||||
// copy the digits to a null-terminated string in order to pass it to strtoul()
|
||||
// the largest unsigned long decimal representation is 4294967295 (10 bytes)
|
||||
char convert[11];
|
||||
char *c = convert;
|
||||
while (isspace(*p)) {
|
||||
p++;
|
||||
}
|
||||
while (isdigit(*p) && c < convert + sizeof(convert) - 1) {
|
||||
*c++ = *p++;
|
||||
}
|
||||
*c = '\0';
|
||||
*target = strtoul(convert, &c, 10);
|
||||
|
||||
return p + 1; // one past the last digit
|
||||
}
|
||||
|
||||
/* parse an unsigned long long decimal text representation */
|
||||
const char *parse_ull(const char *in, const char *find, unsigned long long *target) {
|
||||
const char *p = strstr(in, find);
|
||||
if (!p) {
|
||||
// didn't find what we were looking for
|
||||
return NULL;
|
||||
}
|
||||
p += strlen(find);
|
||||
|
||||
// copy the digits to a null-terminated string in order to pass it to strtoull()
|
||||
// the largest unsigned long long decimal representation is 18446744073709551615 (21 bytes)
|
||||
char convert[22];
|
||||
char *c = convert;
|
||||
while (isspace(*p)) {
|
||||
p++;
|
||||
}
|
||||
while(isdigit(*p) && c < convert + sizeof(convert) - 1) {
|
||||
*c++ = *p++;
|
||||
}
|
||||
*c = '\0';
|
||||
*target = strtoull(convert, &c, 10);
|
||||
|
||||
return p + 1; // one past the last digit
|
||||
}
|
||||
|
||||
int jemalloc_parse(const char *in, struct ocelot_alloc_info *out) {
|
||||
/* Parse the output of jemalloc_status() plain output
|
||||
* This is an ugly hack; it would be faster to build directly what we want
|
||||
* rather than walk down a formatted string that has generated lots of
|
||||
* information that is discarded.
|
||||
* If the input breaks, the error code indicates where the parsing failed.
|
||||
*/
|
||||
|
||||
// set everything to zero
|
||||
memset (out, 0, sizeof(struct ocelot_alloc_info));
|
||||
|
||||
/*
|
||||
___ Begin jemalloc statistics ___
|
||||
Version: "5.3.0-0-g54eaed1d8b56b1aa528be3bdd1877e59c56fa90c"
|
||||
Build-time option settings
|
||||
...
|
||||
Run-time option settings
|
||||
...
|
||||
Profiling settings
|
||||
...
|
||||
Arenas: 17
|
||||
Quantum size: 16
|
||||
Page size: 4096
|
||||
Maximum thread-cached size class: 32768
|
||||
Number of bin size classes: 36
|
||||
Number of thread-cache bin size classes: 41
|
||||
Number of large size classes: 196
|
||||
*/
|
||||
if (!(in = strstr(in, "___ Begin jemalloc statistics ___"))) {
|
||||
return 1;
|
||||
}
|
||||
|
||||
if (!(in = parse_sz(in, "Arenas: ", &out->nr_arena))) {
|
||||
return 10;
|
||||
}
|
||||
if (!(in = parse_sz(in, "Number of bin size classes: ", &out->nr_bin_small))) {
|
||||
return 11;
|
||||
}
|
||||
if (!(in = parse_sz(in, "Number of thread-cache bin size classes: ", &out->nr_bin_tc))) {
|
||||
return 12;
|
||||
}
|
||||
if (!(in = parse_sz(in, "Number of large size classes: ", &out->nr_bin_large))) {
|
||||
return 13;
|
||||
}
|
||||
|
||||
// Allocated: 895744, active: 1024000, metadata: 3047664 (n_thp 0), resident: 3985408, mapped: 7315456, retained: 1073152
|
||||
if (!(in = parse_ull(in, "Allocated: ", &out->mem_allocated))) {
|
||||
return 20;
|
||||
}
|
||||
if (!(in = parse_ull(in, "active: ", &out->mem_active))) {
|
||||
return 21;
|
||||
}
|
||||
if (!(in = parse_ull(in, "metadata: ", &out->mem_metadata))) {
|
||||
return 22;
|
||||
}
|
||||
if (!(in = parse_ull(in, "resident: ", &out->mem_resident))) {
|
||||
return 23;
|
||||
}
|
||||
if (!(in = parse_ull(in, "mapped: ", &out->mem_mapped))) {
|
||||
return 24;
|
||||
}
|
||||
if (!(in = parse_ull(in, "retained: ", &out->mem_retained))) {
|
||||
return 25;
|
||||
}
|
||||
|
||||
/*
|
||||
allocated nmalloc (#/sec) ndalloc (#/sec) nrequests (#/sec) nfill (#/sec) nflush (#/sec)
|
||||
small: 247840 131452 0 129036 0 24219 0 2402 0 8543 0
|
||||
large: 290816 815 0 808 0 971 0 815 0 2351 0
|
||||
total: 538656 132267 0 129844 0
|
||||
*/
|
||||
if (!(in = strstr(in, "(#/sec)\nsmall:"))) {
|
||||
return 30;
|
||||
}
|
||||
|
||||
if (!(in = parse_sz(in, " ", &out->small.allocated))) {
|
||||
return 40;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nmalloc_total))) {
|
||||
return 41;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.ndalloc_rate))) {
|
||||
return 42;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.ndalloc_total))) {
|
||||
return 43;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nrequests_rate))) {
|
||||
return 44;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nrequests_total))) {
|
||||
return 45;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nfill_rate))) {
|
||||
return 46;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nfill_total))) {
|
||||
return 47;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nflush_rate))) {
|
||||
return 48;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->small.nflush_total))) {
|
||||
return 49;
|
||||
}
|
||||
|
||||
if (!(in = strstr(in, "\nlarge:"))) {
|
||||
return 60;
|
||||
}
|
||||
|
||||
if (!(in = parse_sz(in, " ", &out->large.allocated))) {
|
||||
return 70;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nmalloc_total))) {
|
||||
return 71;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.ndalloc_rate))) {
|
||||
return 72;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.ndalloc_total))) {
|
||||
return 73;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nrequests_rate))) {
|
||||
return 74;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nrequests_total))) {
|
||||
return 75;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nfill_rate))) {
|
||||
return 76;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nfill_total))) {
|
||||
return 77;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nflush_rate))) {
|
||||
return 78;
|
||||
}
|
||||
if (!(in = parse_sz(in, " ", &out->large.nflush_total))) {
|
||||
return 79;
|
||||
}
|
||||
|
||||
return 0;
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
#ifndef SRC_PARSE_JEMALLOC_H_
|
||||
#define SRC_PARSE_JEMALLOC_H_
|
||||
|
||||
struct ocelot_alloc_stat {
|
||||
size_t allocated;
|
||||
size_t nmalloc_total;
|
||||
size_t nmalloc_rate;
|
||||
size_t ndalloc_total;
|
||||
size_t ndalloc_rate;
|
||||
size_t nrequests_total;
|
||||
size_t nrequests_rate;
|
||||
size_t nfill_total;
|
||||
size_t nfill_rate;
|
||||
size_t nflush_total;
|
||||
size_t nflush_rate;
|
||||
};
|
||||
|
||||
struct ocelot_alloc_info {
|
||||
size_t nr_arena;
|
||||
size_t nr_bin_small;
|
||||
size_t nr_bin_tc;
|
||||
size_t nr_bin_large;
|
||||
|
||||
unsigned long long mem_allocated;
|
||||
unsigned long long mem_active;
|
||||
unsigned long long mem_metadata;
|
||||
unsigned long long mem_resident;
|
||||
unsigned long long mem_mapped;
|
||||
unsigned long long mem_retained;
|
||||
|
||||
struct ocelot_alloc_stat small;
|
||||
struct ocelot_alloc_stat large;
|
||||
};
|
||||
|
||||
int jemalloc_parse(const char *in, struct ocelot_alloc_info *out);
|
||||
|
||||
#endif // SRC_PARSE_JEMALLOC_H_
|
||||
@@ -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;
|
||||
}
|
||||
@@ -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_
|
||||
+17
-5
@@ -26,10 +26,6 @@ static schedule *sched;
|
||||
|
||||
struct stats_t stats;
|
||||
|
||||
const char * version() {
|
||||
return "2.1.1";
|
||||
}
|
||||
|
||||
static void create_daemon() {
|
||||
pid_t pid;
|
||||
|
||||
@@ -113,7 +109,8 @@ int main(int argc, char **argv) {
|
||||
conf_arg = true;
|
||||
conf_file_path = argv[++i];
|
||||
} else if (strcmp(argv[i], "-V") == 0 || strcmp(argv[i], "--version") == 0) {
|
||||
std::cout << "Ocelot version " << version() << std::endl;
|
||||
// preprocessor concatenation
|
||||
std::cout << "Ocelot version " OCELOT_VERSION ", compiled " __DATE__ " " __TIME__ << std::endl;
|
||||
return 0;
|
||||
} else {
|
||||
std::cout << "Usage: " << argv[0] << " [-v] [-c configfile] [--daemonize]" << std::endl;
|
||||
@@ -146,6 +143,9 @@ int main(int argc, char **argv) {
|
||||
auto combined_logger = std::make_shared<spdlog::logger>("logger", begin(sinks), end(sinks));
|
||||
// If we don't set flush on info, the file log takes a long while to actually flush
|
||||
combined_logger->flush_on(spdlog::level::info);
|
||||
combined_logger->info(
|
||||
"Ocelot version " OCELOT_VERSION ", compiled " __DATE__ " " __TIME__
|
||||
);
|
||||
spdlog::register_logger(combined_logger);
|
||||
|
||||
db = new mysql(conf);
|
||||
@@ -167,17 +167,29 @@ int main(int argc, char **argv) {
|
||||
db->load_whitelist(whitelist);
|
||||
|
||||
stats.open_connections = 0;
|
||||
stats.peak_connections = 0;
|
||||
stats.opened_connections = 0;
|
||||
stats.connection_rate = 0;
|
||||
stats.requests = 0;
|
||||
stats.request_rate = 0;
|
||||
stats.leechers = 0;
|
||||
stats.seeders = 0;
|
||||
stats.user_queue_size = 0;
|
||||
stats.torrent_queue_size = 0;
|
||||
stats.peer_queue_size = 0;
|
||||
stats.snatch_queue_size = 0;
|
||||
stats.token_queue_size = 0;
|
||||
stats.max_client_request_len = 0;
|
||||
stats.announcements = 0;
|
||||
stats.succ_announcements = 0;
|
||||
stats.scrapes = 0;
|
||||
stats.bytes_read = 0;
|
||||
stats.bytes_written = 0;
|
||||
stats.auth_error_secret = 0;
|
||||
stats.auth_error_report = 0;
|
||||
stats.auth_error_announce_key = 0;
|
||||
stats.client_error = 0;
|
||||
stats.http_error = 0;
|
||||
stats.start_time = time(NULL);
|
||||
|
||||
// Create worker object, which handles announces and scrapes and all that jazz
|
||||
|
||||
@@ -3,6 +3,12 @@
|
||||
|
||||
// Copyright [2017-2024] Orpheus
|
||||
|
||||
#define OCELOT_VERSION_MAJOR 2
|
||||
#define OCELOT_VERSION_MINOR 1
|
||||
#define OCELOT_VERSION_BUGFIX 4
|
||||
#define OCELOT_VERSION_NREV 0
|
||||
#define OCELOT_VERSION "2.1.4"
|
||||
|
||||
#include <time.h>
|
||||
|
||||
#include <string>
|
||||
@@ -94,6 +100,7 @@ typedef std::unordered_map<std::string, std::string> params_type;
|
||||
|
||||
struct stats_t {
|
||||
std::atomic<uint32_t> open_connections;
|
||||
std::atomic<uint32_t> peak_connections;
|
||||
std::atomic<uint64_t> opened_connections;
|
||||
std::atomic<uint64_t> connection_rate;
|
||||
std::atomic<uint32_t> leechers;
|
||||
@@ -103,6 +110,7 @@ struct stats_t {
|
||||
std::atomic<uint32_t> peer_queue_size;
|
||||
std::atomic<uint32_t> snatch_queue_size;
|
||||
std::atomic<uint32_t> token_queue_size;
|
||||
std::atomic<uint32_t> max_client_request_len;
|
||||
std::atomic<uint64_t> requests;
|
||||
std::atomic<uint64_t> request_rate;
|
||||
std::atomic<uint64_t> announcements;
|
||||
@@ -110,6 +118,11 @@ struct stats_t {
|
||||
std::atomic<uint64_t> scrapes;
|
||||
std::atomic<uint64_t> bytes_read;
|
||||
std::atomic<uint64_t> bytes_written;
|
||||
std::atomic<uint64_t> auth_error_secret;
|
||||
std::atomic<uint64_t> auth_error_report;
|
||||
std::atomic<uint64_t> auth_error_announce_key;
|
||||
std::atomic<uint64_t> client_error;
|
||||
std::atomic<uint64_t> http_error;
|
||||
time_t start_time;
|
||||
};
|
||||
extern struct stats_t stats;
|
||||
|
||||
+264
-78
@@ -1,91 +1,277 @@
|
||||
// Copyright [2017-2024] Orpheus
|
||||
|
||||
#include <iostream>
|
||||
#include <map>
|
||||
#include <unistd.h>
|
||||
#include <fcntl.h>
|
||||
#include <string.h>
|
||||
#include <sys/resource.h>
|
||||
|
||||
#include <jemalloc/jemalloc.h>
|
||||
#include <spdlog/spdlog.h> // debug
|
||||
|
||||
#include <fstream>
|
||||
#include <sstream>
|
||||
#include <iostream>
|
||||
#include <iomanip> // for std::setfill() and std::setw()
|
||||
#include <map>
|
||||
#include <mutex>
|
||||
#include <thread>
|
||||
|
||||
#include "ocelot.h"
|
||||
#include "misc_functions.h"
|
||||
#include "jemalloc_parse.h"
|
||||
#include "report.h"
|
||||
#include "response.h"
|
||||
#include "user.h"
|
||||
|
||||
std::string report(params_type ¶ms, user_list &users_list, client_opts_t &client_opts, unsigned int announce_interval, unsigned int announce_jitter) {
|
||||
std::stringstream output;
|
||||
std::string action = params["get"];
|
||||
if (action == "stats") {
|
||||
time_t uptime = time(NULL) - stats.start_time;
|
||||
int up_d = uptime / 86400;
|
||||
uptime -= up_d * 86400;
|
||||
int up_h = uptime / 3600;
|
||||
uptime -= up_h * 3600;
|
||||
int up_m = uptime / 60;
|
||||
int up_s = uptime - up_m * 60;
|
||||
int jemalloc_fd;
|
||||
std::mutex jemalloc_fd_mutex;
|
||||
|
||||
output << "Uptime: " << up_d << " days, "
|
||||
<< (up_h < 10 ? "0" : "") << inttostr(up_h) << ':'
|
||||
<< (up_m < 10 ? "0" : "") << inttostr(up_m) << ':'
|
||||
<< (up_s < 10 ? "0" : "") << inttostr(up_s) << "\n"
|
||||
<< "version: " << version() << "\n"
|
||||
<< stats.opened_connections << " connections opened\n"
|
||||
<< stats.open_connections << " open connections\n"
|
||||
<< stats.connection_rate << " connections/s\n"
|
||||
<< stats.requests << " requests handled\n"
|
||||
<< stats.request_rate << " requests/s\n"
|
||||
<< stats.succ_announcements << " successful announcements\n"
|
||||
<< (stats.announcements - stats.succ_announcements) << " failed announcements\n"
|
||||
<< stats.scrapes << " scrapes\n"
|
||||
<< stats.leechers << " leechers tracked\n"
|
||||
<< stats.seeders << " seeders tracked\n"
|
||||
<< stats.user_queue_size << " items in user queue\n"
|
||||
<< stats.torrent_queue_size << " items in torrent queue\n"
|
||||
<< stats.peer_queue_size << " items in peer queue\n"
|
||||
<< stats.snatch_queue_size << " items in snatch queue\n"
|
||||
<< stats.token_queue_size << " items in token queue\n"
|
||||
<< stats.bytes_read << " bytes read\n"
|
||||
<< stats.bytes_written << " bytes written\n"
|
||||
<< announce_interval << " announce interval\n"
|
||||
<< announce_jitter << " announce jitter\n"
|
||||
;
|
||||
} else if (action == "prom_stats") {
|
||||
time_t uptime = time(NULL) - stats.start_time;
|
||||
output << "ocelot_uptime " << uptime << "\n"
|
||||
<< "ocelot_open_connections " << stats.open_connections << "\n"
|
||||
<< "ocelot_connection_rate " << stats.connection_rate << "\n"
|
||||
<< "ocelot_requests " << stats.requests << "\n"
|
||||
<< "ocelot_request_rate " << stats.request_rate << "\n"
|
||||
<< "ocelot_succ_announcements " << stats.succ_announcements << "\n"
|
||||
<< "ocelot_total_announcements " << stats.announcements << "\n"
|
||||
<< "ocelot_scrapes " << stats.scrapes << "\n"
|
||||
<< "ocelot_leechers " << stats.leechers << "\n"
|
||||
<< "ocelot_seeders " << stats.seeders << "\n"
|
||||
<< "ocelot_user_queue " << stats.user_queue_size << "\n"
|
||||
<< "ocelot_torrent_queue " << stats.torrent_queue_size << "\n"
|
||||
<< "ocelot_peer_queue " << stats.peer_queue_size << "\n"
|
||||
<< "ocelot_snatch_queue " << stats.snatch_queue_size << "\n"
|
||||
<< "ocelot_token_queue " << stats.token_queue_size << "\n"
|
||||
<< "ocelot_bytes_read " << stats.bytes_read << "\n"
|
||||
<< "ocelot_bytes_written " << stats.bytes_written << "\n#";
|
||||
} else if (action == "user") {
|
||||
std::string key = params["key"];
|
||||
if (key.empty()) {
|
||||
output << "Invalid action\n";
|
||||
} else {
|
||||
user_list::const_iterator u = users_list.find(key);
|
||||
if (u != users_list.end()) {
|
||||
output << "{\"id\":" << u->second->get_id()
|
||||
<< ",\"leeching\":" << u->second->get_leeching()
|
||||
<< ",\"seeding\":" << u->second->get_seeding()
|
||||
<< ",\"deleted\":" << u->second->is_deleted()
|
||||
<< ",\"protected\":" << u->second->is_protected()
|
||||
<< ",\"can_leech\":" << u->second->can_leech()
|
||||
<< "}" << "\n";
|
||||
return response(output.str(), client_opts);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
output << "Invalid action\n";
|
||||
/* Getting jemalloc stats back to Gazelle is a bit cumbersome
|
||||
* because all jemalloc can do is write to a file descriptor.
|
||||
* So we let it write to a file (which is why using a ramdisk
|
||||
* is important) and then read it back to return as an HTTP
|
||||
* response.
|
||||
*/
|
||||
|
||||
// callback function for malloc_stats_print()
|
||||
void jemalloc_stats_export(void *opaque, const char *buf) {
|
||||
if (jemalloc_fd > 0) {
|
||||
write(jemalloc_fd, buf, strlen(buf));
|
||||
}
|
||||
output << "success";
|
||||
return response(output.str(), client_opts);
|
||||
}
|
||||
|
||||
int dump_jemalloc(const char *filename, const char *opts) {
|
||||
// open the file descriptor
|
||||
std::lock_guard<std::mutex> guard(jemalloc_fd_mutex);
|
||||
jemalloc_fd = open(filename, O_CREAT|O_WRONLY, 0600);
|
||||
if (jemalloc_fd == -1) {
|
||||
return jemalloc_fd;
|
||||
}
|
||||
|
||||
// dump stats
|
||||
malloc_stats_print(jemalloc_stats_export, NULL, opts);
|
||||
|
||||
// tidy up
|
||||
int result = close(jemalloc_fd);
|
||||
jemalloc_fd = 0;
|
||||
return result;
|
||||
}
|
||||
|
||||
std::string report_jemalloc_plain(const char *opts, const std::string path) {
|
||||
std::string filename(path + "/jemalloc.json." + std::to_string(pthread_self()));
|
||||
if (dump_jemalloc(filename.c_str(), opts) != 0) {
|
||||
// return an empty string if something went bananas
|
||||
unlink(filename.c_str());
|
||||
return std::string();
|
||||
}
|
||||
|
||||
// slurp the contents into a string
|
||||
std::ifstream input(filename);
|
||||
std::ostringstream output;
|
||||
while(input >> output.rdbuf())
|
||||
;
|
||||
unlink(filename.c_str());
|
||||
output << "\n";
|
||||
|
||||
std::shared_ptr<spdlog::logger> logger(spdlog::get("logger"));
|
||||
logger->debug("jemalloc stats written to " + filename + ", size=" + std::to_string(output.str().size()));
|
||||
return output.str();
|
||||
}
|
||||
|
||||
#define ITEM_BYTE(LABEL, VALUE) "\"" LABEL "\":{\"type\":\"byte\",\"value\":" << VALUE << "}"
|
||||
#define ITEM_ELAPSED(LABEL, VALUE) "\"" LABEL "\":{\"type\":\"elapsed\",\"value\":" << VALUE << "}"
|
||||
#define ITEM_NUM(LABEL, VALUE) "\"" LABEL "\":{\"type\":\"number\",\"value\":" << VALUE << "}"
|
||||
#define ITEM_STR(LABEL, VALUE) "\"" LABEL "\":{\"type\":\"string\",\"value\":\"" VALUE "\"}"
|
||||
#define ITEM_VERSION(LABEL, MAJ, MIN, BUGFIX) "\"" LABEL "\":{\"type\":\"string\",\"value\":\"" << MAJ << "." << MIN << "." << BUGFIX << "\"}"
|
||||
|
||||
std::string report(const uint32_t announce_interval, const uint32_t announce_jitter) {
|
||||
std::ostringstream output;
|
||||
output << "{"
|
||||
<< ITEM_STR("version", OCELOT_VERSION) << ','
|
||||
<< ITEM_VERSION("jemalloc version", JEMALLOC_VERSION_MAJOR, JEMALLOC_VERSION_MINOR, JEMALLOC_VERSION_BUGFIX) << ','
|
||||
<< ITEM_NUM("connections opened", stats.opened_connections) << ','
|
||||
<< ITEM_NUM("open connections", stats.open_connections) << ','
|
||||
<< ITEM_NUM("peak connections", stats.peak_connections) << ','
|
||||
<< ITEM_NUM("connections/s", stats.connection_rate) << ','
|
||||
<< ITEM_NUM("requests handled", stats.requests) << ','
|
||||
<< ITEM_NUM("requests/s", stats.request_rate) << ','
|
||||
<< ITEM_NUM("successful announcements", stats.succ_announcements) << ','
|
||||
<< ITEM_NUM("failed announcements", (stats.announcements - stats.succ_announcements)) << ','
|
||||
<< ITEM_NUM("scrapes", stats.scrapes) << ','
|
||||
<< ITEM_NUM("leechers tracked", stats.leechers) << ','
|
||||
<< ITEM_NUM("seeders tracked", stats.seeders) << ','
|
||||
<< ITEM_NUM("items in user queue", stats.user_queue_size) << ','
|
||||
<< ITEM_NUM("items in torrent queue", stats.torrent_queue_size) << ','
|
||||
<< ITEM_NUM("items in peer queue", stats.peer_queue_size) << ','
|
||||
<< ITEM_NUM("items in snatch queue", stats.snatch_queue_size) << ','
|
||||
<< ITEM_NUM("items in token queue", stats.token_queue_size) << ','
|
||||
<< ITEM_BYTE("max client request length", stats.max_client_request_len) << ','
|
||||
<< ITEM_BYTE("bytes read", stats.bytes_read) << ','
|
||||
<< ITEM_BYTE("bytes written", stats.bytes_written) << ','
|
||||
<< ITEM_NUM("bad tracker secret received", stats.auth_error_secret) << ','
|
||||
<< ITEM_NUM("bad report secret received", stats.auth_error_report) << ','
|
||||
<< ITEM_NUM("bad announce key received", stats.auth_error_announce_key) << ','
|
||||
<< ITEM_NUM("bad client configuration", stats.client_error) << ','
|
||||
<< ITEM_NUM("bad http request", stats.http_error) << ','
|
||||
<< ITEM_NUM("announce interval", announce_interval) << ','
|
||||
<< ITEM_NUM("announce jitter", announce_jitter) << ','
|
||||
<< ITEM_ELAPSED("uptime", (time(NULL) - stats.start_time))
|
||||
;
|
||||
|
||||
struct rusage r;
|
||||
if (getrusage(RUSAGE_SELF, &r) == 0) {
|
||||
output << ','
|
||||
<< ITEM_ELAPSED("user time", r.ru_utime.tv_sec << '.' << std::setfill('0') << std::setw(6) << r.ru_utime.tv_usec) << ','
|
||||
<< ITEM_ELAPSED("system time", r.ru_stime.tv_sec << '.' << std::setfill('0') << std::setw(6) << r.ru_stime.tv_usec) << ','
|
||||
<< ITEM_BYTE("maximum resident set size", r.ru_maxrss * 1024) << ','
|
||||
<< ITEM_NUM("minor page faults", r.ru_minflt) << ','
|
||||
<< ITEM_NUM("major page faults", r.ru_majflt) << ','
|
||||
<< ITEM_NUM("blocks in", r.ru_inblock) << ','
|
||||
<< ITEM_NUM("blocks out", r.ru_oublock) << ','
|
||||
<< ITEM_NUM("voluntary context switches", r.ru_nvcsw ) << ','
|
||||
<< ITEM_NUM("involuntary context switches", r.ru_nivcsw )
|
||||
;
|
||||
}
|
||||
output << "}\n";
|
||||
return output.str();
|
||||
}
|
||||
|
||||
std::string report_prom_stats(const char *jemalloc) {
|
||||
std::ostringstream output;
|
||||
|
||||
struct ocelot_alloc_info ji;
|
||||
int result = jemalloc_parse(jemalloc, &ji);
|
||||
|
||||
output << "ocelot_uptime " << time(NULL) - stats.start_time << "\n"
|
||||
"ocelot_version " << OCELOT_VERSION_MAJOR
|
||||
<< std::setfill('0') << std::setw(3) << OCELOT_VERSION_MINOR
|
||||
<< std::setfill('0') << std::setw(3) << OCELOT_VERSION_BUGFIX
|
||||
<< std::setfill('0') << std::setw(3) << OCELOT_VERSION_NREV << "\n"
|
||||
"jemalloc_version " << JEMALLOC_VERSION_MAJOR
|
||||
<< std::setfill('0') << std::setw(3) << JEMALLOC_VERSION_MINOR
|
||||
<< std::setfill('0') << std::setw(3) << JEMALLOC_VERSION_BUGFIX << "\n"
|
||||
"ocelot_open_connections " << stats.open_connections << "\n"
|
||||
|
||||
"#TYPE ocelot_peak_connections counter\n"
|
||||
"ocelot_peak_connections " << stats.peak_connections << "\n"
|
||||
"ocelot_connection_rate " << stats.connection_rate << "\n"
|
||||
"ocelot_requests " << stats.requests << "\n"
|
||||
"ocelot_request_rate " << stats.request_rate << "\n"
|
||||
"ocelot_succ_announcements " << stats.succ_announcements << "\n"
|
||||
"ocelot_total_announcements " << stats.announcements << "\n"
|
||||
"ocelot_scrapes " << stats.scrapes << "\n"
|
||||
"ocelot_leechers " << stats.leechers << "\n"
|
||||
"ocelot_seeders " << stats.seeders << "\n"
|
||||
"ocelot_user_queue " << stats.user_queue_size << "\n"
|
||||
"ocelot_torrent_queue " << stats.torrent_queue_size << "\n"
|
||||
"ocelot_peer_queue " << stats.peer_queue_size << "\n"
|
||||
"ocelot_snatch_queue " << stats.snatch_queue_size << "\n"
|
||||
"ocelot_token_queue " << stats.token_queue_size << "\n"
|
||||
"ocelot_bytes_read " << stats.bytes_read << "\n"
|
||||
"ocelot_bytes_written " << stats.bytes_written << "\n"
|
||||
|
||||
"#TYPE ocelot_max_client_request_len counter\n"
|
||||
"ocelot_max_client_request_len " << stats.max_client_request_len << "\n"
|
||||
|
||||
"#TYPE ocelot_error counter counter\n"
|
||||
"ocelot_error{kind=\"secret\"} " << stats.auth_error_secret << "\n"
|
||||
"ocelot_error{kind=\"report\"} " << stats.auth_error_report << "\n"
|
||||
"ocelot_error{kind=\"announce\"} " << stats.auth_error_announce_key << "\n"
|
||||
"ocelot_error{kind=\"client\"} " << stats.client_error << "\n"
|
||||
"ocelot_error{kind=\"http\"} " << stats.http_error << "\n"
|
||||
"ocelot_error{kind=\"jemalloc_parse\"} " << result << "\n"
|
||||
|
||||
"#TYPE jemalloc_arena_total counter\n"
|
||||
"jemalloc_arena_total " << ji.nr_arena << "\n"
|
||||
|
||||
"#TYPE jemalloc_bin_total counter\n"
|
||||
"jemalloc_bin_total{bin=\"small\"} " << ji.nr_bin_small << "\n"
|
||||
"jemalloc_bin_total{bin=\"thread_cache\"} " << ji.nr_bin_tc << "\n"
|
||||
"jemalloc_bin_total{bin=\"large\"} " << ji.nr_bin_large << "\n"
|
||||
|
||||
"#TYPE jemalloc_allocated counter\n"
|
||||
"jemalloc_allocated{bucket=\"small\"} " << ji.small.allocated << "\n"
|
||||
"jemalloc_allocated{bucket=\"large\"} " << ji.large.allocated << "\n"
|
||||
|
||||
"#TYPE jemalloc_nmalloc_total counter\n"
|
||||
"jemalloc_nmalloc_total{bucket=\"small\"} " << ji.small.nmalloc_total << "\n"
|
||||
"jemalloc_nmalloc_total{bucket=\"large\"} " << ji.large.nmalloc_total << "\n"
|
||||
|
||||
"#TYPE jemalloc_ndalloc_total counter\n"
|
||||
"jemalloc_ndalloc_total{bucket=\"small\"} " << ji.small.ndalloc_total << "\n"
|
||||
"jemalloc_ndalloc_total{bucket=\"large\"} " << ji.large.ndalloc_total << "\n"
|
||||
|
||||
"#TYPE jemalloc_nrequests_total counter\n"
|
||||
"jemalloc_nrequests_total{bucket=\"small\"} " << ji.small.nrequests_total << "\n"
|
||||
"jemalloc_nrequests_total{bucket=\"large\"} " << ji.large.nrequests_total << "\n"
|
||||
|
||||
"#TYPE jemalloc_nfill_total counter\n"
|
||||
"jemalloc_nfill_total{bucket=\"small\"} " << ji.small.nfill_total << "\n"
|
||||
"jemalloc_nfill_total{bucket=\"large\"} " << ji.large.nfill_total << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_total counter\n"
|
||||
"jemalloc_nflush_total{bucket=\"small\"} " << ji.small.nflush_total << "\n"
|
||||
"jemalloc_nflush_total{bucket=\"large\"} " << ji.large.nflush_total << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_total gauge\n"
|
||||
"jemalloc_memory{global=\"allocated\"} " << ji.mem_allocated << "\n"
|
||||
"jemalloc_memory{global=\"active\"} " << ji.mem_active << "\n"
|
||||
"jemalloc_memory{global=\"metadata\"} " << ji.mem_metadata << "\n"
|
||||
"jemalloc_memory{global=\"resident\"} " << ji.mem_resident << "\n"
|
||||
"jemalloc_memory{global=\"mapped\"} " << ji.mem_mapped << "\n"
|
||||
"jemalloc_memory{global=\"retained\"} " << ji.mem_retained << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_rate gauge\n"
|
||||
"jemalloc_allocated{bucket=\"small\"} " << ji.small.allocated << "\n"
|
||||
"jemalloc_allocated{bucket=\"large\"} " << ji.large.allocated << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_rate gauge\n"
|
||||
"jemalloc_nmalloc_rate{bucket=\"small\"} " << ji.small.nmalloc_rate << "\n"
|
||||
"jemalloc_nmalloc_rate{bucket=\"large\"} " << ji.large.nmalloc_rate << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_total gauge\n"
|
||||
"jemalloc_ndalloc_rate{bucket=\"small\"} " << ji.small.ndalloc_rate << "\n"
|
||||
"jemalloc_ndalloc_rate{bucket=\"large\"} " << ji.large.ndalloc_rate << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_rate gauge\n"
|
||||
"jemalloc_nrequests_rate{bucket=\"small\"} " << ji.small.nrequests_rate << "\n"
|
||||
"jemalloc_nrequests_rate{bucket=\"large\"} " << ji.large.nrequests_rate << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_rate gauge\n"
|
||||
"jemalloc_fill_rate{bucket=\"small\"} " << ji.small.nfill_rate << "\n"
|
||||
"jemalloc_fill_rate{bucket=\"large\"} " << ji.large.nfill_rate << "\n"
|
||||
|
||||
"#TYPE jemalloc_nflush_rate gauge\n"
|
||||
"jemalloc_nflush_rate{bucket=\"small\"} " << ji.small.nflush_rate << "\n"
|
||||
"jemalloc_nflush_rate{bucket=\"large\"} " << ji.large.nflush_rate << "\n"
|
||||
;
|
||||
|
||||
struct rusage r;
|
||||
if (getrusage(RUSAGE_SELF, &r) == 0) {
|
||||
output << "ocelot_time_user " << r.ru_utime.tv_sec << '.' << std::setfill('0') << std::setw(6) << r.ru_utime.tv_usec << "\n"
|
||||
"ocelot_time_system " << r.ru_stime.tv_sec << '.' << std::setfill('0') << std::setw(6) << r.ru_stime.tv_usec << "\n"
|
||||
"ocelot_max_rss " << r.ru_maxrss << "\n"
|
||||
"ocelot_minor_fault " << r.ru_minflt << "\n"
|
||||
"ocelot_major_fault " << r.ru_majflt << "\n"
|
||||
"#TYPE ocelot_block counter\n"
|
||||
"ocelot_block{op=\"in\"} " << r.ru_inblock << "\n"
|
||||
"ocelot_block{op=\"out\"} " << r.ru_oublock << "\n"
|
||||
"ocelot_nvcsw " << r.ru_nvcsw << "\n"
|
||||
"ocelot_nivcsw " << r.ru_nivcsw << "\n"
|
||||
;
|
||||
}
|
||||
output << '#';
|
||||
return output.str();
|
||||
}
|
||||
|
||||
std::string report_user(const user_ptr u) {
|
||||
std::ostringstream output;
|
||||
output << "{\"id\":" << u->get_id()
|
||||
<< ",\"leeching\":" << u->get_leeching()
|
||||
<< ",\"seeding\":" << u->get_seeding()
|
||||
<< ",\"deleted\":" << u->is_deleted()
|
||||
<< ",\"protected\":" << u->is_protected()
|
||||
<< ",\"can_leech\":" << u->can_leech()
|
||||
<< "}\n";
|
||||
return output.str();
|
||||
}
|
||||
+14
-7
@@ -5,12 +5,19 @@
|
||||
|
||||
#include <string>
|
||||
|
||||
std::string report(
|
||||
params_type ¶ms,
|
||||
user_list &users_list,
|
||||
client_opts_t &client_opts,
|
||||
unsigned int announce_interval,
|
||||
unsigned int announce_jitter
|
||||
);
|
||||
// take a snapshot of the current stats
|
||||
void copy_stats(const struct stats_t *in, struct stats_t *out);
|
||||
|
||||
// status report
|
||||
std::string report(const uint32_t announce_interval, const uint32_t announce_jitter);
|
||||
|
||||
// a snapshot of jemalloc statistics
|
||||
std::string report_jemalloc_plain(const char *opts, const std::string path);
|
||||
|
||||
// a prometheus scrape
|
||||
std::string report_prom_stats(const char *jemalloc_stats);
|
||||
|
||||
// user report
|
||||
std::string report_user(const user_ptr u);
|
||||
|
||||
#endif // SRC_REPORT_H_
|
||||
+32
-23
@@ -1,5 +1,7 @@
|
||||
// Copyright [2017-2024] Orpheus
|
||||
|
||||
#include <fmt/format.h>
|
||||
|
||||
#include <algorithm>
|
||||
#include <sstream>
|
||||
|
||||
@@ -10,7 +12,26 @@
|
||||
#include "misc_functions.h"
|
||||
#include "response.h"
|
||||
|
||||
const std::string response(const std::string &body, client_opts_t &client_opts) {
|
||||
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(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";
|
||||
}
|
||||
if (client_opts.http_close) {
|
||||
head += "Connection: Close\r\n";
|
||||
}
|
||||
return head + "\r\n";
|
||||
}
|
||||
|
||||
const std::string http_response(const std::string &body, client_opts_t &client_opts) {
|
||||
std::string out;
|
||||
bool processed = false;
|
||||
if (client_opts.html) {
|
||||
@@ -28,29 +49,17 @@ const std::string response(const std::string &body, client_opts_t &client_opts)
|
||||
processed = true;
|
||||
}
|
||||
if (processed) {
|
||||
return response_head(out.length(), client_opts) + out;
|
||||
return http_head(out.length(), client_opts) + out;
|
||||
}
|
||||
return response_head(body.length(), client_opts) + body;
|
||||
return http_head(body.length(), client_opts) + body;
|
||||
}
|
||||
|
||||
const std::string response_head(size_t content_length, client_opts_t &client_opts) {
|
||||
const std::string content_type = client_opts.html ? "text/html" : "text/plain";
|
||||
std::string head = "HTTP/1.1 200 OK\r\nServer: Ocelot 1.0";
|
||||
head += "\r\nContent-Type: " + content_type;
|
||||
if (client_opts.gzip) {
|
||||
head += "\r\nContent-Encoding: gzip";
|
||||
}
|
||||
if (client_opts.http_close) {
|
||||
head += "\r\nConnection: Close";
|
||||
}
|
||||
head += "\r\nContent-Length: " + inttostr(content_length) + "\r\n\r\n";
|
||||
return head;
|
||||
}
|
||||
|
||||
const std::string error(const std::string &err, client_opts_t &client_opts) {
|
||||
return response("d14:failure reason" + inttostr(err.length()) + ':' + err + "12:min intervali5400e8:intervali5400ee", client_opts);
|
||||
}
|
||||
|
||||
const std::string warning(const std::string &msg) {
|
||||
return "15:warning message" + inttostr(msg.length()) + ':' + msg;
|
||||
const std::string error(const std::string &message, client_opts_t &client_opts) {
|
||||
return http_response(
|
||||
fmt::format(
|
||||
"d14:failure reason{}:{}12:min intervali5400e8:intervali5400ee",
|
||||
message.length(), message
|
||||
),
|
||||
client_opts
|
||||
);
|
||||
}
|
||||
+1
-3
@@ -7,9 +7,7 @@
|
||||
|
||||
#include "ocelot.h"
|
||||
|
||||
const std::string response(const std::string &body, client_opts_t &client_opts);
|
||||
const std::string response_head(size_t content_length, client_opts_t &client_opts);
|
||||
const std::string http_response(const std::string &body, client_opts_t &client_opts);
|
||||
const std::string error(const std::string &err, client_opts_t &client_opts);
|
||||
const std::string warning(const std::string &msg);
|
||||
|
||||
#endif // SRC_RESPONSE_H_
|
||||
+4
-3
@@ -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
@@ -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;
|
||||
}
|
||||
|
||||
+383
-212
@@ -1,6 +1,7 @@
|
||||
// Copyright [2017-2024] Orpheus
|
||||
|
||||
#include <spdlog/spdlog.h>
|
||||
#include <fmt/format.h>
|
||||
|
||||
#include <iostream>
|
||||
#include <string>
|
||||
@@ -24,6 +25,8 @@
|
||||
#include "report.h"
|
||||
#include "user.h"
|
||||
|
||||
std::mutex worker::client_len_mutex;
|
||||
|
||||
//---------- Worker - does stuff with input
|
||||
worker::worker(config * conf_obj, torrent_list &torrents, user_list &users, std::vector<std::string> &_whitelist, mysql * db_obj, site_comm * sc) :
|
||||
conf(conf_obj), db(db_obj), s_comm(sc), torrents_list(torrents), users_list(users), whitelist(_whitelist), status(OPEN), reaper_active(false), randgen((std::random_device())()) {
|
||||
@@ -69,26 +72,26 @@ bool worker::shutdown() {
|
||||
|
||||
std::string worker::work(const std::string &input, std::string &ip, client_opts_t &client_opts) {
|
||||
unsigned int input_length = input.length();
|
||||
{
|
||||
const std::lock_guard<std::mutex> lock(worker::client_len_mutex);
|
||||
unsigned int max_len = stats.max_client_request_len;
|
||||
if (max_len < input_length) {
|
||||
stats.max_client_request_len = input_length;
|
||||
}
|
||||
}
|
||||
|
||||
//---------- Parse request - ugly but fast. Using substr exploded.
|
||||
if (input_length < 60) { // Way too short to be anything useful
|
||||
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 {
|
||||
@@ -96,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++;
|
||||
@@ -120,7 +124,7 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
|
||||
if (input[pos] != '?') {
|
||||
// No parameters given. Probably means we're not talking to a torrent client
|
||||
client_opts.html = true;
|
||||
return response("Nothing to see here", client_opts);
|
||||
return http_response("Nothing to see here", client_opts);
|
||||
}
|
||||
|
||||
// Parse URL params
|
||||
@@ -158,6 +162,7 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
|
||||
++pos;
|
||||
|
||||
if (input.compare(pos, 5, "HTTP/") != 0) {
|
||||
stats.http_error++;
|
||||
return error("Malformed HTTP request", client_opts);
|
||||
}
|
||||
|
||||
@@ -211,94 +216,132 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
|
||||
|
||||
if (status != OPEN) {
|
||||
return error("The tracker is temporarily unavailable.", client_opts);
|
||||
}
|
||||
|
||||
if (action == INVALID) {
|
||||
} else if (action == INVALID) {
|
||||
stats.http_error++;
|
||||
return error("Invalid action", client_opts);
|
||||
}
|
||||
|
||||
if (action == UPDATE) {
|
||||
if (passkey == site_password) {
|
||||
return update(params, client_opts);
|
||||
} else {
|
||||
} else if (action == UPDATE) {
|
||||
if (passkey != site_password) {
|
||||
stats.auth_error_secret++;
|
||||
logger->error("incorrect TRACKER_SECRET received");
|
||||
return error("Authentication failure", client_opts);
|
||||
}
|
||||
}
|
||||
|
||||
if (action == REPORT) {
|
||||
if (passkey == report_password) {
|
||||
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
|
||||
return report(params, users_list, client_opts, announce_interval, conf->get_uint("announce_jitter"));
|
||||
} else {
|
||||
return update(params, client_opts);
|
||||
} else if (action == REPORT) {
|
||||
if (passkey != report_password) {
|
||||
stats.auth_error_report++;
|
||||
logger->error("incorrect TRACKER_REPORT received");
|
||||
return error("Authentication failure", client_opts);
|
||||
}
|
||||
}
|
||||
|
||||
// Either a scrape or an announce
|
||||
|
||||
std::unique_lock<std::mutex> ul_lock(db->user_list_mutex);
|
||||
auto user_it = users_list.find(passkey);
|
||||
if (user_it == users_list.end()) {
|
||||
return error("Passkey not found", client_opts);
|
||||
}
|
||||
user_ptr u = user_it->second;
|
||||
ul_lock.unlock();
|
||||
|
||||
if (action == ANNOUNCE) {
|
||||
// Let's translate the infohash into something nice
|
||||
// info_hash is a url encoded (hex) base 20 number
|
||||
std::string info_hash_decoded = hex_decode(params["info_hash"]);
|
||||
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
|
||||
auto tor = torrents_list.find(info_hash_decoded);
|
||||
if (tor == torrents_list.end()) {
|
||||
std::lock_guard<std::mutex> dr_lock(del_reasons_lock);
|
||||
auto msg = del_reasons.find(info_hash_decoded);
|
||||
if (msg != del_reasons.end()) {
|
||||
if (msg->second.reason != -1) {
|
||||
return error("Unregistered torrent: " + get_del_reason(msg->second.reason), client_opts);
|
||||
} else {
|
||||
return error("Unregistered torrent", client_opts);
|
||||
std::string report_action(params["get"]);
|
||||
if (report_action == "prom_stats") {
|
||||
// exclude per-arena (a), destroyed merged (d), mutex (m) and extents (e) statistics
|
||||
std::string jemalloc_stats(report_jemalloc_plain("adex", conf->get_str("report_path")));
|
||||
return http_response(
|
||||
report_prom_stats(jemalloc_stats.c_str()),
|
||||
client_opts
|
||||
);
|
||||
} else if (report_action == "stats") {
|
||||
return http_response(
|
||||
report(announce_interval, conf->get_uint("announce_jitter")),
|
||||
client_opts
|
||||
);
|
||||
} else if (report_action == "jemalloc") {
|
||||
return http_response(
|
||||
report_jemalloc_plain("adex", conf->get_str("report_path")),
|
||||
client_opts
|
||||
);
|
||||
} else if (report_action == "user") {
|
||||
std::string announce_key = params["key"];
|
||||
if (announce_key.empty()) {
|
||||
stats.auth_error_announce_key++;
|
||||
logger->error("user report with no announce key");
|
||||
return error("Announce key missing", client_opts);
|
||||
}
|
||||
user_ptr u;
|
||||
{
|
||||
// lock scope
|
||||
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
|
||||
auto user_it = users_list.find(announce_key);
|
||||
if (user_it == users_list.end()) {
|
||||
stats.auth_error_announce_key++;
|
||||
logger->error("user report announce key not found");
|
||||
return error("Announce key not found", client_opts);
|
||||
}
|
||||
u = user_it->second;
|
||||
}
|
||||
return http_response(
|
||||
report_user(u),
|
||||
client_opts
|
||||
);
|
||||
}
|
||||
|
||||
stats.http_error++;
|
||||
logger->error("unrecognized report query");
|
||||
return error("unrecognized report query", client_opts);
|
||||
}
|
||||
|
||||
// Either a scrape or an announce, find the user
|
||||
user_ptr u;
|
||||
{
|
||||
// lock scope
|
||||
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
|
||||
auto user_it = users_list.find(passkey);
|
||||
if (user_it == users_list.end()) {
|
||||
stats.auth_error_announce_key++;
|
||||
return error("Passkey not found", client_opts);
|
||||
}
|
||||
u = user_it->second;
|
||||
}
|
||||
|
||||
if (action == SCRAPE) {
|
||||
return scrape(infohashes, headers, client_opts);
|
||||
}
|
||||
|
||||
// Let's translate the infohash into something nice
|
||||
// info_hash is a url encoded (hex) base 20 number
|
||||
std::string info_hash_decoded = hex_decode(params["info_hash"]);
|
||||
std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
|
||||
auto tor = torrents_list.find(info_hash_decoded);
|
||||
if (tor == torrents_list.end()) {
|
||||
std::lock_guard<std::mutex> dr_lock(del_reasons_lock);
|
||||
auto msg = del_reasons.find(info_hash_decoded);
|
||||
if (msg != del_reasons.end()) {
|
||||
if (msg->second.reason != -1) {
|
||||
return error("Unregistered torrent: " + get_del_reason(msg->second.reason), client_opts);
|
||||
} else {
|
||||
return error("Unregistered torrent", client_opts);
|
||||
}
|
||||
} else {
|
||||
return error("Unregistered torrent", client_opts);
|
||||
}
|
||||
return announce(input, tor->second, u, params, headers, ip, client_opts);
|
||||
} else {
|
||||
return scrape(infohashes, headers, client_opts);
|
||||
}
|
||||
return announce(input, tor->second, u, params, headers, ip, client_opts);
|
||||
}
|
||||
|
||||
const std::string bencode_warning(const std::string &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 ¶ms, params_type &headers, std::string &ip, client_opts_t &client_opts) {
|
||||
cur_time = time(NULL);
|
||||
|
||||
if (params["compact"] != "1") {
|
||||
stats.client_error++;
|
||||
return error("Your client does not support compact announces", client_opts);
|
||||
}
|
||||
|
||||
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"]));
|
||||
|
||||
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
|
||||
bool inserted = false; // If we insert the peer as opposed to update
|
||||
bool update_torrent = false; // Whether or not we should update the torrent in the DB
|
||||
bool completed_torrent = false; // Whether or not the current announcement is a snatch
|
||||
bool stopped_torrent = false; // Was the torrent just stopped?
|
||||
bool expire_token = false; // Whether or not to expire a token after torrent completion
|
||||
bool peer_changed = false; // Whether or not the peer is new or has changed since the last announcement
|
||||
bool invalid_ip = false;
|
||||
bool inc_l = false, inc_s = false, dec_l = false, dec_s = false;
|
||||
userid_t userid = u->get_id();
|
||||
|
||||
params_type::const_iterator peer_id_iterator = params.find("peer_id");
|
||||
if (peer_id_iterator == params.end()) {
|
||||
stats.client_error++;
|
||||
return error("No peer ID", client_opts);
|
||||
}
|
||||
const std::string peer_id = hex_decode(peer_id_iterator->second);
|
||||
if (peer_id.length() != 20) {
|
||||
stats.client_error++;
|
||||
return error("Invalid peer ID", client_opts);
|
||||
}
|
||||
|
||||
@@ -312,17 +355,70 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
|
||||
}
|
||||
}
|
||||
if (!found) {
|
||||
stats.client_error++;
|
||||
return error("Your client is not on the whitelist", client_opts);
|
||||
}
|
||||
}
|
||||
wl_lock.unlock();
|
||||
|
||||
std::stringstream peer_key_stream;
|
||||
// "Randomize" the element order in the peer map by prefixing with a peer id byte
|
||||
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());
|
||||
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
|
||||
bool inserted = false; // If we insert the peer as opposed to update
|
||||
bool update_torrent = false; // Whether or not we should update the torrent in the DB
|
||||
bool completed_torrent = false; // Whether or not the current announcement is a snatch
|
||||
bool stopped_torrent = false; // Was the torrent just stopped?
|
||||
bool expire_token = false; // Whether or not to expire a token after torrent completion
|
||||
bool peer_changed = false; // Whether or not the peer is new or has changed since the last announcement
|
||||
bool invalid_ip = false;
|
||||
bool inc_l = false, inc_s = false, dec_l = false, dec_s = false;
|
||||
userid_t userid = u->get_id();
|
||||
|
||||
if (params["event"] == "completed") {
|
||||
// Don't update <snatched> here as we may decide to use other conditions later on
|
||||
@@ -333,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) {
|
||||
@@ -440,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
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -475,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] == '.') {
|
||||
@@ -514,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!
|
||||
@@ -535,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) {
|
||||
@@ -550,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) {
|
||||
@@ -697,39 +794,38 @@ 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 += warning("Illegal character found in IP address. IPv6 is not supported");
|
||||
output += bencode_warning("Illegal character found in IP address. IPv6 is not supported");
|
||||
}
|
||||
output += 'e';
|
||||
|
||||
@@ -740,14 +836,13 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
|
||||
/*if (headers["accept-encoding"].find("gzip") != std::string::npos) {
|
||||
client_opts.gzip = true;
|
||||
}*/
|
||||
return response(output, client_opts);
|
||||
return http_response(output, client_opts);
|
||||
}
|
||||
|
||||
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()) {
|
||||
@@ -755,50 +850,57 @@ 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) {
|
||||
client_opts.gzip = true;
|
||||
}
|
||||
return response(output, client_opts);
|
||||
return http_response(output, client_opts);
|
||||
}
|
||||
|
||||
// TODO: Restrict to local IPs
|
||||
std::string worker::update(params_type ¶ms, client_opts_t &client_opts) {
|
||||
if (params["action"] == "change_passkey") {
|
||||
std::string action(params["action"]);
|
||||
if (action == "change_passkey") {
|
||||
std::string oldpasskey = params["oldpasskey"];
|
||||
std::string newpasskey = params["newpasskey"];
|
||||
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 (params["action"] == "add_torrent") {
|
||||
} 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;
|
||||
}
|
||||
@@ -809,10 +911,13 @@ std::string worker::update(params_type ¶ms, 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"]);
|
||||
} else if (params["action"] == "update_torrent") {
|
||||
std::string info_hash = params["info_hash"];
|
||||
info_hash = hex_decode(info_hash);
|
||||
logger->info(
|
||||
"added torrent {} with FL {} {}",
|
||||
t->id, t->free_torrent, params["freetorrent"]
|
||||
);
|
||||
} else if (action == "update_torrent") {
|
||||
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;
|
||||
@@ -825,11 +930,17 @@ std::string worker::update(params_type ¶ms, 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 (params["action"] == "update_torrents") {
|
||||
} else if (action == "update_torrents") {
|
||||
// Each decoded infohash is exactly 20 characters long.
|
||||
std::string info_hashes = params["info_hashes"];
|
||||
info_hashes = hex_decode(info_hashes);
|
||||
@@ -847,34 +958,48 @@ std::string worker::update(params_type ¶ms, 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 (params["action"] == "add_token") {
|
||||
std::string info_hash = hex_decode(params["info_hash"]);
|
||||
} else if (action == "add_token") {
|
||||
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 (params["action"] == "remove_token") {
|
||||
std::string info_hash = hex_decode(params["info_hash"]);
|
||||
} else if (action == "remove_token") {
|
||||
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 (params["action"] == "delete_torrent") {
|
||||
std::string info_hash = params["info_hash"];
|
||||
info_hash = hex_decode(info_hash);
|
||||
} else if (action == "delete_torrent") {
|
||||
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()) {
|
||||
@@ -882,8 +1007,12 @@ std::string worker::update(params_type ¶ms, 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) {
|
||||
@@ -898,33 +1027,44 @@ std::string worker::update(params_type ¶ms, 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 (params["action"] == "add_user") {
|
||||
} 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 (params["action"] == "remove_user") {
|
||||
} 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);
|
||||
}
|
||||
} else if (params["action"] == "remove_users") {
|
||||
} else if (action == "remove_users") {
|
||||
// Each passkey is exactly 32 characters long.
|
||||
std::string passkeys = params["passkeys"];
|
||||
std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
|
||||
@@ -932,12 +1072,15 @@ std::string worker::update(params_type ¶ms, 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);
|
||||
}
|
||||
}
|
||||
} else if (params["action"] == "update_user") {
|
||||
} else if (action == "update_user") {
|
||||
std::string passkey = params["passkey"];
|
||||
bool can_leech = true;
|
||||
bool protect_ip = false;
|
||||
@@ -948,20 +1091,29 @@ std::string worker::update(params_type ¶ms, 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 (params["action"] == "add_whitelist") {
|
||||
} 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);
|
||||
} else if (params["action"] == "remove_whitelist") {
|
||||
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);
|
||||
for (unsigned int i = 0; i < whitelist.size(); i++) {
|
||||
@@ -970,8 +1122,11 @@ std::string worker::update(params_type ¶ms, client_opts_t &client_opts) {
|
||||
break;
|
||||
}
|
||||
}
|
||||
logger->info("De-whitelisted " + peer_id);
|
||||
} else if (params["action"] == "edit_whitelist") {
|
||||
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"];
|
||||
std::lock_guard<std::mutex> wl_lock(db->whitelist_mutex);
|
||||
@@ -982,24 +1137,35 @@ std::string worker::update(params_type ¶ms, client_opts_t &client_opts) {
|
||||
}
|
||||
}
|
||||
whitelist.push_back(new_peer_id);
|
||||
logger->info("Edited whitelist item from " + old_peer_id + " to " + new_peer_id);
|
||||
} else if (params["action"] == "update_announce_interval") {
|
||||
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));
|
||||
} else if (params["action"] == "update_announce_jitter") {
|
||||
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")));
|
||||
} else if (params["action"] == "info_torrent") {
|
||||
std::stringstream output;
|
||||
logger->info(
|
||||
"modified announce jitter from {} to {}",
|
||||
prev, conf->get_uint("announce_jitter")
|
||||
);
|
||||
} else if (action == "info_torrent") {
|
||||
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);
|
||||
@@ -1007,9 +1173,9 @@ std::string worker::update(params_type ¶ms, client_opts_t &client_opts) {
|
||||
output << ",\"fail\":1";
|
||||
}
|
||||
output << "}\n";
|
||||
return response(output.str(), client_opts);
|
||||
return http_response(output.str(), client_opts);
|
||||
}
|
||||
return response("success", client_opts);
|
||||
return http_response("success", client_opts);
|
||||
}
|
||||
|
||||
peer_list::iterator worker::add_peer(peer_list &peer_list, const std::string &peer_key) {
|
||||
@@ -1078,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() {
|
||||
@@ -1097,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) {
|
||||
@@ -1172,7 +1343,7 @@ std::string worker::get_del_reason(int code) {
|
||||
return "Audience Recording";
|
||||
break;
|
||||
default:
|
||||
return "";
|
||||
return std::string();
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
+2
-1
@@ -34,11 +34,12 @@ class worker {
|
||||
std::shared_ptr<spdlog::logger> logger;
|
||||
std::mt19937 randgen;
|
||||
std::uniform_int_distribution<int> jitter;
|
||||
static std::mutex client_len_mutex;
|
||||
|
||||
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;
|
||||
|
||||
Reference in new issue
Block a user