lock entire reap operations in a mutex

CHANGES

@@ -1,8 +1,14 @@
--- 2.1.4.(2024-03-24)
+-- 2.1.5 (2026-02-12)
 FEATURES
-* Refactor the handling of SQL insert buffers
-BUILD
-* now requires libfmt, used for formatting strings
+* connection_mother() does not need db access
+* validate client whitelist before acquiring torrent lock
+* add timers to measure lock acquisition, announce() and reap_peers() duration
+BUGFIXES
+* Wrap peer reaper code in a mutex to avoid stats desynchronization.
+* Prevent upspeed and downspeed calculation from exceeding 2^31-1.
+* Fix error when seeders reannounce with left > 0.
+NOTE
+* 2.1.4 was not officially released.
 
 -- 2.1.3 (2024-03-09)
 BREAKING

CMakeLists.txt

@@ -49,9 +49,6 @@ 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)
@@ -63,7 +60,6 @@ SET(LINK_LIBRARIES
     ${Boost_SYSTEM_LIBRARY}
     ${Boost_IOSTREAMS_LIBRARY}
     ${LIBEV_LIBRARY}
-    ${LIBFMT_IBRARY}
     ${MYSQLPP_LIBRARY}
     Threads::Threads
 )
@@ -79,5 +75,4 @@ endif()
 target_link_libraries(
     ocelot
     ${LINK_LIBRARIES}
-    fmt::fmt
 )

Dockerfile

@@ -10,7 +10,6 @@ RUN apt-get update \
         libboost-iostreams-dev \
         libboost-system-dev \
         libev-dev \
-        libfmt-dev \
         libjemalloc-dev \
         libmysql++-dev \
         netcat-traditional \
@@ -28,9 +27,10 @@ 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
+    && mv /srv/ocelot.conf.dist /srv/ocelot.conf \
+    && rm -rf /srv/build \
+    && mkdir -p /tmp/ocelot
 
 # default listen_port value in ocelot.conf
 EXPOSE 34000

README.md

@@ -8,7 +8,6 @@ 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)
@@ -24,10 +23,9 @@ 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
@@ -44,7 +42,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 or libraries are added).
+* Prepare the build environment. (This must be re-executed when new source files are added).
 
         cmake -Wno-dev . -B build
 

src/config.cpp

@@ -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 = atoi(value.c_str());
+        uint_val = strtoint32(value);
     } else if (val_type == CONF_STR) {
         str_val = value;
     }
@@ -109,7 +109,7 @@ void config::init() {
 confval * config::get(const std::string &setting_name) {
     const auto setting = settings.find(setting_name);
     if (setting == settings.end()) {
-        spdlog::get("logger")->info("WARNING: Unrecognized setting '{}'", setting_name);
+        spdlog::get("logger")->info("WARNING: Unrecognized setting '" + setting_name + "'");
         return dummy_setting;
     }
     return &setting->second;
@@ -152,7 +152,7 @@ void config::reload() {
     const std::string conf_file_path(get_str("conf_file_path"));
     std::ifstream conf_file(conf_file_path);
     if (conf_file.fail()) {
-        spdlog::get("logger")->error("Config file '{}' couldn't be opened", conf_file_path);
+        spdlog::get("logger")->error("Config file '" + conf_file_path + "' couldn't be opened");
     } else {
         init();
         load(conf_file);

src/config.h

@@ -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();

src/db.cpp

@@ -21,11 +21,7 @@
 
 #define DB_LOCK_TIMEOUT 50
 
-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) {
+mysql::mysql(config * conf) : u_active(false), t_active(false), p_active(false), s_active(false), tok_active(false) {
     logger = spdlog::get("logger");
     load_config(conf);
     if (mysql_db.empty()) {
@@ -38,7 +34,7 @@ mysql::mysql(config * conf) :
         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 ({})", er.what());
+        logger->error("Failed to connect to MySQL (" + std::string(er.what()) + ")");
         return;
     }
 
@@ -73,12 +69,6 @@ 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;");
@@ -90,9 +80,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: {}", er.what());
+        logger->error("Query error in clear_peer_data: " + std::string(er.what()));
     } catch (const mysqlpp::Exception &er) {
-        logger->error("Query error in clear_peer_data: {}", er.what());
+        logger->error("Query error in clear_peer_data: " + std::string(er.what()));
     }
 }
 
@@ -157,19 +147,15 @@ void mysql::load_torrents(torrent_list &torrents) {
             }
         }
     } catch (const mysqlpp::BadQuery &er) {
-        logger->error("Query error in load_torrents: {}", er.what());
+        logger->error("Query error in load_torrents: " + std::string(er.what()));
         return;
     }
-    logger->info("Loaded {} torrents", torrents.size());
+    logger->info("Loaded " + std::to_string(torrents.size()) + " torrents");
     load_tokens(torrents);
 }
 
 void mysql::load_users(user_list &users) {
-    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);
+    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';");
     try {
         mysqlpp::StoreQueryResult res = query.store();
         size_t num_rows = res.num_rows();
@@ -206,18 +192,14 @@ void mysql::load_users(user_list &users) {
             }
         }
     } catch (const mysqlpp::BadQuery &er) {
-        logger->error("Query error in load_users: {} [{}]", er.what(), sql);
+        logger->error("Query error in load_users: " + std::string(er.what()));
         return;
     }
-    logger->info("Loaded {} users", users.size());
+    logger->info("Loaded " + std::to_string(users.size()) + " users");
 }
 
 void mysql::load_tokens(torrent_list &torrents) {
-    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);
+    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';");
     int token_count = 0;
     try {
         mysqlpp::StoreQueryResult res = query.store();
@@ -234,16 +216,15 @@ void mysql::load_tokens(torrent_list &torrents) {
             }
         }
     } catch (const mysqlpp::BadQuery &er) {
-        logger->error("Query error in load_tokens: {} [{}]", er.what(), sql);
+        logger->error("Query error in load_tokens: " + std::string(er.what()));
         return;
     }
-    logger->info("Loaded {} tokens", token_count);
+    logger->info("Loaded " + std::to_string(token_count) + " tokens");
 }
 
 
 void mysql::load_whitelist(std::vector<std::string> &whitelist) {
-    std::string sql("SELECT peer_id FROM xbt_client_whitelist");
-    mysqlpp::Query query = conn.query(sql);
+    mysqlpp::Query query = conn.query("SELECT peer_id FROM xbt_client_whitelist;");
     try {
         mysqlpp::StoreQueryResult res = query.store();
         size_t num_rows = res.num_rows();
@@ -255,40 +236,64 @@ 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: {} [{}]", er.what(), sql);
+        logger->error("Query error in load_whitelist: " + std::string(er.what()));
         return;
     }
-    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;
+    if (whitelist.size() == 0) {
+        logger->info("Assuming no whitelist desired, disabling");
+    } else {
+        logger->info("Loaded " + std::to_string(whitelist.size()) + " clients into the whitelist");
+    }
 }
 
 void mysql::record_token(const std::string &record) {
-    std::lock_guard<std::mutex> buf_lock(token_buffer_lock);
+    if (!update_token_buffer.empty()) {
+        update_token_buffer += ",";
+    }
     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> buf_lock(torrent_buffer_lock);
+    std::lock_guard<std::mutex> tb_lock(torrent_buffer_lock);
+    if (!update_torrent_buffer.empty()) {
+        update_torrent_buffer += ",";
+    }
     update_torrent_buffer += record;
 }
 
-void mysql::record_user(const std::string &record) {
-    std::lock_guard<std::mutex> buf_lock(user_buffer_lock);
-    update_user_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();
 }
 
 bool mysql::all_clear() {
@@ -303,289 +308,350 @@ void mysql::flush() {
     flush_tokens();
 }
 
-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());
+void mysql::flush_users() {
+    if (readonly) {
+        update_user_buffer.clear();
+        return;
     }
-    q_lock.unlock();
+    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();
+    }
+}
 
-    std::unique_lock<std::mutex> buf_lock(peer_buffer_lock);
-    std::unique_lock<std::mutex> buf_light_lock(peer_light_buffer_lock);
+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() {
     if (readonly) {
         update_light_peer_buffer.clear();
         update_heavy_peer_buffer.clear();
         return;
     }
-    /* 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
+    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()) {
         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;
         }
-        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()
-        ));
+        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);
         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;
         }
-        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()
-        ));
+        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);
         update_light_peer_buffer.clear();
+        sql.clear();
     }
-    buf_light_lock.unlock();
     stats.peer_queue_size = qsize + qsize_added;
 
-    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);
-        });
+    if (!p_active) {
+        std::thread thread(&mysql::do_flush_peers, this);
         thread.detach();
     }
 }
 
 void mysql::flush_tokens() {
-    std::unique_lock<std::mutex> q_lock(token_queue_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: {}, next query length: {}", qsize, token_queue.front().size());
+        logger->info("Token flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(token_queue.front().size()));
     }
-    q_lock.unlock();
-
-    std::unique_lock<std::mutex> buf_lock(token_buffer_lock);
-    if (readonly) {
-        update_token_buffer.clear();
+    if (update_token_buffer.empty()) {
         return;
     }
-    if (update_token_buffer.empty() && qsize == 0) {
-        return;
-    }
-    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);
-        });
+    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);
         thread.detach();
     }
 }
 
-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());
-    }
-    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();
-    }
-}
-
-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());
-    }
-    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();
-    }
-}
-
-void mysql::do_flush(const char *name, std::queue<std::string> &queue, std::mutex &mtx, bool &active, std::atomic<uint32_t>& counter) {
-    active = true;
+void mysql::do_flush_users() {
+    u_active = true;
     try {
-        std::unique_lock<std::mutex> q_lock(mtx);
-        size_t q_size = queue.size();
-        q_lock.unlock();
-
         mysqlpp::Connection c = create_connection();
-        std::string sql;
-        while (q_size > 0) {
+        while (user_queue.size() > 0) {
             try {
-                std::unique_lock<std::mutex> q_lock(mtx);
-                sql = queue.front();
-                q_lock.unlock();
+                std::string sql = user_queue.front();
                 mysqlpp::Query query = c.query(sql);
                 if (!query.exec()) {
-                    logger->info("{} flush failed ({} remain)", name, q_size);
-                    std::this_thread::sleep_for(std::chrono::seconds(3));
+                    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;
                 }
-                q_lock.lock();
-                queue.pop();
-                q_size = queue.size();
-                q_lock.unlock();
-                counter -= 1;
             }
             catch (const mysqlpp::BadQuery &er) {
-                logger->error("SQL error: {} in {} flush with a qlength: {} queue size: {} sql=[{}]",
-                    er.what(), name, queue.front().size(), q_size, sql
-                );
+                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("DB error: {} in {} flush with a qlength: {} queue size: {}",
-                    er.what(), name, queue.front().size(), q_size
-                );
+                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("General DB error in {} flush: {}", name, er.what());
+        logger->error("MySQL error in flush_users: " + std::string(er.what()));
     }
-    active = false;
+    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;
+            }
+        }
+    }
+    catch (const mysqlpp::Exception &er) {
+        logger->error("MySQL error in flush_torrents: " + std::string(er.what()));
+    }
+    t_active = false;
+}
+
+void mysql::do_flush_peers() {
+    p_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;
+}
+
+void mysql::do_flush_snatches() {
+    s_active = true;
+    try {
+        mysqlpp::Connection c = create_connection();
+        while (snatch_queue.size() > 0) {
+            try {
+                std::string sql = snatch_queue.front();
+                mysqlpp::Query query = c.query(sql);
+                if (!query.exec()) {
+                    logger->info("Snatch flush failed (" + std::to_string(snatch_queue.size()) + " remain)");
+                    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;
+                }
+            }
+            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()));
+                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()));
+                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()));
+    }
+    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;
 }

src/db.h

@@ -16,44 +16,13 @@
 
 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;
-    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::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;
 
     std::queue<std::string> user_queue;
     std::queue<std::string> torrent_queue;
@@ -61,16 +30,19 @@ 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;
 
@@ -78,7 +50,11 @@ class mysql {
     void load_tokens(torrent_list &torrents);
     mysqlpp::Connection create_connection();
 
-    void do_flush(const char *name, std::queue<std::string> &queue, std::mutex &mtx, bool &active, std::atomic<uint32_t>& counter);
+    void do_flush_users();
+    void do_flush_torrents();
+    void do_flush_snatches();
+    void do_flush_peers();
+    void do_flush_tokens();
 
     void flush_users();
     void flush_torrents();
@@ -90,10 +66,6 @@ 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();
@@ -102,25 +74,27 @@ 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);
 
-    // (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,uploaded_change,downloaded_change)
+    void record_user(const std::string &record);
 
     // (id,seeders,leechers,snatched_change,balance)
     void record_torrent(const std::string &record);
 
-    // (id,uploaded_change,downloaded_change)
-    void record_user(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;
 };
 
 #pragma GCC visibility pop

src/events.cpp

@@ -1,4 +1,4 @@
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #include <spdlog/spdlog.h>
 
@@ -18,7 +18,7 @@ std::mutex peak_open_mutex;
 
 //---------- Connection mother - spawns middlemen and lets them deal with the connection
 
-connection_mother::connection_mother(config * conf, worker * worker_obj, mysql * db_obj, site_comm * sc_obj, schedule * sched) : work(worker_obj), db(db_obj) {
+connection_mother::connection_mother(config * conf, worker * worker_obj, site_comm * sc_obj, schedule * sched) : work(worker_obj) {
     logger = spdlog::get("logger");
 
     // Handle config stuff first
@@ -54,7 +54,7 @@ void connection_mother::reload_config(config * conf) {
     unsigned int old_max_connections = max_connections;
     load_config(conf);
     if (old_listen_port != listen_port) {
-        logger->info("Changing listen port from {} to {}", old_listen_port, listen_port);
+        logger->info("Changing listen port from " + std::to_string(old_listen_port) +" to " + std::to_string(listen_port));
         int new_listen_socket = create_listen_socket();
         if (new_listen_socket != 0) {
             listen_event.stop();
@@ -77,7 +77,7 @@ int connection_mother::create_listen_socket() {
     // Stop old sockets from hogging the port
     int yes = 1;
     if (setsockopt(new_listen_socket, SOL_SOCKET, SO_REUSEADDR, &yes, sizeof(int)) == -1) {
-        logger->error("Could not reuse socket: {}", strerror(errno));
+        logger->error("Could not reuse socket: " + std::string(strerror(errno)));
         return 0;
     }
 
@@ -88,24 +88,24 @@ int connection_mother::create_listen_socket() {
 
     // Bind
     if (bind(new_listen_socket, (sockaddr *) &address, sizeof(address)) == -1) {
-        logger->error("Bind failed: {}", strerror(errno));
+        logger->error("Bind failed: " + std::string(strerror(errno)));
         return 0;
     }
 
     // Listen
     if (listen(new_listen_socket, max_connections) == -1) {
-        logger->error("Listen failed: {}", strerror(errno));
+        logger->error("Listen failed: " + std::string(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: {}", strerror(errno));
+        logger->error("Could not get socket flags: " + std::string(strerror(errno)));
         return 0;
     }
     if (fcntl(new_listen_socket, F_SETFL, flags | O_NONBLOCK) == -1) {
-        logger->error("Could not set non-blocking: {}", strerror(errno));
+        logger->error("Could not set non-blocking: " + std::string(strerror(errno)));
         return 0;
     }
 
@@ -113,7 +113,7 @@ int connection_mother::create_listen_socket() {
 }
 
 void connection_mother::run() {
-    logger->info("Sockets up on port {}, starting event loop!", listen_port);
+    logger->info("Sockets up on port " + std::to_string(listen_port) + ", starting event loop!");
     ev_loop(ev_default_loop(0), 0);
 }
 
@@ -146,7 +146,7 @@ connection_middleman::connection_middleman(int &listen_socket, worker * new_work
     auto logger = spdlog::get("logger");
     connect_sock = accept(listen_socket, NULL, NULL);
     if (connect_sock == -1) {
-        logger->error("Accept failed, errno {}: {}", errno, strerror(errno));
+        logger->error("Accept failed, errno " + std::to_string(errno) + ": " + std::string(strerror(errno)));
         delete this;
         return;
     }

src/events.h

@@ -1,7 +1,7 @@
 #ifndef SRC_EVENTS_H_
 #define SRC_EVENTS_H_
 
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #ifdef EV_ERROR
 #undef EV_ERROR
@@ -29,7 +29,6 @@
 #include "config.h"
 #include "worker.h"
 #include "schedule.h"
-#include "db.h"
 #include "site_comm.h"
 
 /*
@@ -58,13 +57,12 @@ class connection_mother {
 
     int listen_socket;
     worker * work;
-    mysql * db;
     ev::io listen_event;
     ev::timer schedule_event;
     std::shared_ptr<spdlog::logger> logger;
 
  public:
-    connection_mother(config * conf, worker * worker_obj, mysql * db_obj, site_comm * sc_obj, schedule * sched_obj);
+    connection_mother(config * conf, worker * worker_obj, site_comm * sc_obj, schedule * sched_obj);
     ~connection_mother();
     void reload_config(config * conf);
     int create_listen_socket();

src/misc_functions.cpp

@@ -6,6 +6,29 @@
 
 #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);
@@ -37,3 +60,26 @@ 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;
+}

src/misc_functions.h

@@ -5,6 +5,10 @@
 
 #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_

src/ocelot.cpp

@@ -1,4 +1,4 @@
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #include <sys/stat.h>
 #include <syslog.h>
@@ -183,6 +183,12 @@ int main(int argc, char **argv) {
     stats.announcements = 0;
     stats.succ_announcements = 0;
     stats.scrapes = 0;
+    stats.announce_duration = 0;
+    stats.torrent_lock_total = 0;
+    stats.torrent_lock_duration = 0;
+    stats.reap_total = 0;
+    stats.reap_lock_duration = 0;
+    stats.reap_duration = 0;
     stats.bytes_read = 0;
     stats.bytes_written = 0;
     stats.auth_error_secret = 0;
@@ -199,7 +205,7 @@ int main(int argc, char **argv) {
     sched = new schedule(conf, work, db, sc);
 
     // Create connection mother, which binds to its socket and handles the event stuff
-    mother = new connection_mother(conf, work, db, sc, sched);
+    mother = new connection_mother(conf, work, sc, sched);
 
     // Add signal handlers now that all objects have been created
     struct sigaction handler{}, ignore{};

src/ocelot.h

@@ -1,13 +1,13 @@
 #ifndef SRC_OCELOT_H_
 #define SRC_OCELOT_H_
 
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #define OCELOT_VERSION_MAJOR 2
 #define OCELOT_VERSION_MINOR 1
-#define OCELOT_VERSION_BUGFIX 4
+#define OCELOT_VERSION_BUGFIX 5
 #define OCELOT_VERSION_NREV 0
-#define OCELOT_VERSION "2.1.4"
+#define OCELOT_VERSION "2.1.5"
 
 #include <time.h>
 
@@ -116,6 +116,12 @@ struct stats_t {
     std::atomic<uint64_t> announcements;
     std::atomic<uint64_t> succ_announcements;
     std::atomic<uint64_t> scrapes;
+    std::atomic<uint64_t> announce_duration; // all durations measure microseconds
+    std::atomic<uint64_t> torrent_lock_total;
+    std::atomic<uint64_t> torrent_lock_duration;
+    std::atomic<uint64_t> reap_total;
+    std::atomic<uint64_t> reap_lock_duration;
+    std::atomic<uint64_t> reap_duration;
     std::atomic<uint64_t> bytes_read;
     std::atomic<uint64_t> bytes_written;
     std::atomic<uint64_t> auth_error_secret;

src/report.cpp

@@ -1,4 +1,4 @@
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #include <unistd.h>
 #include <fcntl.h>
@@ -62,7 +62,7 @@ std::string report_jemalloc_plain(const char *opts, const std::string path) {
     if (dump_jemalloc(filename.c_str(), opts) != 0) {
         // return an empty string if something went bananas
         unlink(filename.c_str());
-        return std::string();
+        return std::string("");
     }
 
     // slurp the contents into a string
@@ -100,6 +100,12 @@ std::string report(const uint32_t announce_interval, const uint32_t announce_jit
         << ITEM_NUM("scrapes", stats.scrapes) << ','
         << ITEM_NUM("leechers tracked", stats.leechers) << ','
         << ITEM_NUM("seeders tracked", stats.seeders) << ','
+        << ITEM_NUM("announce duration", stats.announce_duration) << ','
+        << ITEM_NUM("torrent lock total", stats.torrent_lock_total) << ','
+        << ITEM_NUM("torrent lock duration", stats.torrent_lock_duration) << ','
+        << ITEM_NUM("reap lock total", stats.reap_total) << ','
+        << ITEM_NUM("reap lock duration", stats.reap_lock_duration) << ','
+        << ITEM_NUM("reap duration", stats.reap_duration) << ','
         << 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) << ','
@@ -173,7 +179,7 @@ std::string report_prom_stats(const char *jemalloc) {
         "#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"
+        "#TYPE ocelot_error 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"
@@ -181,6 +187,14 @@ std::string report_prom_stats(const char *jemalloc) {
         "ocelot_error{kind=\"http\"} "           << stats.http_error << "\n"
         "ocelot_error{kind=\"jemalloc_parse\"} " << result << "\n"
 
+        "#TYPE ocelot_timing counter\n"
+        "ocelot_timing{kind=\"announce_duration\"} "     << stats.announce_duration << "\n"
+        "ocelot_timing{kind=\"torrent_lock_total\"} "    << stats.torrent_lock_total << "\n"
+        "ocelot_timing{kind=\"torrent_lock_duration\"} " << stats.torrent_lock_duration << "\n"
+        "ocelot_timing{kind=\"reap_total\"} "            << stats.reap_total << "\n"
+        "ocelot_timing{kind=\"reap_lock_duration\"} "    << stats.reap_lock_duration << "\n"
+        "ocelot_timing{kind=\"reap_duration\"} "         << stats.reap_duration << "\n"
+
         "#TYPE jemalloc_arena_total counter\n"
         "jemalloc_arena_total " << ji.nr_arena << "\n"
 

src/report.h

@@ -1,7 +1,7 @@
 #ifndef SRC_REPORT_H_
 #define SRC_REPORT_H_
 
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #include <string>
 

src/response.cpp

@@ -1,7 +1,5 @@
 // Copyright [2017-2024] Orpheus
 
-#include <fmt/format.h>
-
 #include <algorithm>
 #include <sstream>
 
@@ -16,12 +14,9 @@ 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
-    ));
+    std::string head = "HTTP/1.1 200 OK\r\nServer: Ocelot " OCELOT_VERSION "\r\n"
+        "Content-Type: " + content_type + "\r\n"
+        "Content-Length: " + inttostr(content_length) + "\r\n";
     if (client_opts.gzip) {
         head += "Content-Encoding: gzip\r\n";
     }
@@ -56,10 +51,8 @@ const std::string http_response(const std::string &body, client_opts_t &client_o
 
 const std::string error(const std::string &message, client_opts_t &client_opts) {
     return http_response(
-        fmt::format(
-            "d14:failure reason{}:{}12:min intervali5400e8:intervali5400ee",
-            message.length(), message
-        ),
+        "d14:failure reason" + inttostr(message.length()) + ':' + message
+            + "12:min intervali5400e8:intervali5400ee",
         client_opts
     );
 }

src/schedule.cpp

@@ -34,10 +34,9 @@ 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("{} open, {} connections ({}/s), {} requests ({}/s)",
-            stats.open_connections, stats.opened_connections, stats.connection_rate,
-            stats.requests, stats.request_rate
-        );
+        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)");
     }
 
     if (work->get_status() == CLOSING && db->all_clear() && sc->all_clear()) {

src/site_comm.cpp

@@ -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: {}", qsize);
+        logger->info("Token expire queue size: " + std::to_string(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 {} when trying to expire a token!", status_code);
+                logger->error("Response returned with status code " + std::to_string(status_code) + " when trying to expire a token!");
                 // TODO: add exponential backoff
                 std::this_thread::sleep_for(std::chrono::seconds(5));
             }
         }
     } catch (std::exception &er) {
-        logger->error("Exception: {}", er.what());
+        logger->error("Exception: " + std::string(er.what()));
     }
     t_active = false;
 }

src/worker.h

@@ -1,7 +1,7 @@
 #ifndef SRC_WORKER_H_
 #define SRC_WORKER_H_
 
-// Copyright [2017-2024] Orpheus
+// Copyright [2017-2026] Orpheus
 
 #include <spdlog/spdlog.h>
 
@@ -39,7 +39,7 @@ class worker {
     unsigned int announce_interval;
     unsigned int del_reason_lifetime;
     unsigned int peers_timeout;
-    int numwant_limit;
+    unsigned int numwant_limit;
     bool keepalive_enabled;
     std::string site_password;
     std::string report_password;
@@ -57,7 +57,7 @@ class worker {
     worker(config * conf_obj, torrent_list &torrents, user_list &users, std::vector<std::string> &_whitelist, mysql * db_obj, site_comm * sc);
     void reload_config(config * conf);
     std::string work(const std::string &input, std::string &ip, client_opts_t &client_opts);
-    std::string announce(const std::string &input, torrent &tor, user_ptr &u, params_type &params, params_type &headers, std::string &ip, client_opts_t &client_opts);
+    std::string announce(const std::string &input, const std::string &peer_id, torrent &tor, user_ptr &u, params_type &params, params_type &headers, std::string &ip, client_opts_t &client_opts);
     std::string scrape(const std::list<std::string> &infohashes, params_type &headers, client_opts_t &client_opts);
     std::string update(params_type &params, client_opts_t &client_opts);