2 Commits
Author SHA1 Message Date
Spine ce2192c018 detabbed source code, no code changes 2024-02-15 16:05:57 +01:00
Spine 2a8a7eb24b Merge branch 'announce-jitter' into 'master'
redesign announce jitter to better spread the load after startup

See merge request tracker/ocelot!8
2024-02-11 15:04:57 +00:00
22 changed files with 2789 additions and 2709 deletions

No files matched your search

+102 -98
View File
@@ -1,166 +1,170 @@
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <iostream>
#include <fstream>
#include <string>
#include <spdlog/spdlog.h>
#include "config.h"
#include "misc_functions.h"
confval::confval() {
bool_val = 0;
uint_val = 0;
str_val = "";
val_type = CONF_NONEXISTENT;
bool_val = 0;
uint_val = 0;
str_val = "";
val_type = CONF_NONEXISTENT;
}
confval::confval(bool value) {
bool_val = value;
val_type = CONF_BOOL;
bool_val = value;
val_type = CONF_BOOL;
}
confval::confval(unsigned int value) {
uint_val = value;
val_type = CONF_UINT;
uint_val = value;
val_type = CONF_UINT;
}
confval::confval(const char * value) {
str_val = value;
val_type = CONF_STR;
str_val = value;
val_type = CONF_STR;
}
bool confval::get_bool() {
return bool_val;
return bool_val;
}
unsigned int confval::get_uint() {
return uint_val;
return uint_val;
}
std::string confval::get_str() {
return str_val;
return str_val;
}
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);
} else if (val_type == CONF_STR) {
str_val = value;
}
if (val_type == CONF_BOOL) {
bool_val = value == "1" || value == "true" || value == "yes";
} else if (val_type == CONF_UINT) {
uint_val = strtoint32(value);
} else if (val_type == CONF_STR) {
str_val = value;
}
}
config::config() {
init();
dummy_setting = new confval(); // Safety value to use if we're accessing nonexistent settings
init();
dummy_setting = new confval(); // Safety value to use if we're accessing nonexistent settings
}
void config::init() {
// Internal stuff
add("listen_port", 34000u);
add("max_connections", 1024u);
add("max_middlemen", 20000u);
add("max_read_buffer", 4096u);
add("connection_timeout", 10u);
add("keepalive_timeout", 0u);
add("daemonize", false);
// Internal stuff
add("listen_port", 34000u);
add("max_connections", 1024u);
add("max_middlemen", 20000u);
add("max_read_buffer", 4096u);
add("connection_timeout", 10u);
add("keepalive_timeout", 0u);
add("daemonize", false);
// Tracker requests
add("announce_interval", 1800u);
add("announce_jitter", 240u);
add("max_request_size", 4096u);
add("numwant_limit", 50u);
add("request_log_size", 500u);
// Tracker requests
add("announce_interval", 1800u);
add("announce_jitter", 240u);
add("max_request_size", 4096u);
add("numwant_limit", 50u);
add("request_log_size", 500u);
// Timers
add("del_reason_lifetime", 86400u);
add("peers_timeout", 7200u);
add("reap_peers_interval", 1800u);
add("schedule_interval", 3u);
// Timers
add("del_reason_lifetime", 86400u);
add("peers_timeout", 7200u);
add("reap_peers_interval", 1800u);
add("schedule_interval", 3u);
// MySQL
add("mysql_db", "gazelle");
add("mysql_host", "localhost");
add("mysql_port", 3306u);
add("mysql_username", "");
add("mysql_password", "");
add("mysql_port", 3306u);
// MySQL
add("mysql_db", "gazelle");
add("mysql_host", "localhost");
add("mysql_port", 3306u);
add("mysql_username", "");
add("mysql_password", "");
add("mysql_port", 3306u);
// Site communication
add("site_host", "127.0.0.1");
add("site_service", "http");
add("site_path", "");
add("site_password", "00000000000000000000000000000000");
add("report_password", "00000000000000000000000000000000");
// Site communication
add("site_host", "127.0.0.1");
add("site_service", "http");
add("site_path", "");
add("site_password", "00000000000000000000000000000000");
add("report_password", "00000000000000000000000000000000");
// Logging
add("log", false);
add("log_path", "ocelot"); // path to where to write log + name of log (don't need to put file extension)
// Logging
add("log", false);
add("log_path", "ocelot"); // path to where to write log + name of log (don't need to put file extension)
// Debugging
add("readonly", false);
// Debugging
add("readonly", false);
}
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 + "'");
return dummy_setting;
}
return &setting->second;
const auto setting = settings.find(setting_name);
if (setting == settings.end()) {
spdlog::get("logger")->info("WARNING: Unrecognized setting '" + setting_name + "'");
return dummy_setting;
}
return &setting->second;
}
bool config::get_bool(const std::string &setting_name) {
return get(setting_name)->get_bool();
return get(setting_name)->get_bool();
}
unsigned int config::get_uint(const std::string &setting_name) {
return get(setting_name)->get_uint();
return get(setting_name)->get_uint();
}
std::string config::get_str(const std::string &setting_name) {
return get(setting_name)->get_str();
return get(setting_name)->get_str();
}
void config::set(const std::string &setting_name, const std::string &value) {
get(setting_name)->set(value);
get(setting_name)->set(value);
}
void config::load(const std::string &conf_file_path, std::istream &conf_file) {
load(conf_file);
add("conf_file_path", conf_file_path.c_str());
load(conf_file);
add("conf_file_path", conf_file_path.c_str());
}
void config::load(std::istream &conf_file) {
std::string line;
while (getline(conf_file, line)) {
size_t pos;
if (line[0] != '#' && (pos = line.find('=')) != std::string::npos) {
std::string key(trim(line.substr(0, pos)));
std::string value(trim(line.substr(pos + 1)));
set(key, value);
}
}
std::string line;
while (getline(conf_file, line)) {
size_t pos;
if (line[0] != '#' && (pos = line.find('=')) != std::string::npos) {
std::string key(trim(line.substr(0, pos)));
std::string value(trim(line.substr(pos + 1)));
set(key, value);
}
}
}
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");
} else {
init();
load(conf_file);
}
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");
} else {
init();
load(conf_file);
}
}
std::string config::trim(const std::string str) {
size_t ltrim = str.find_first_not_of(" \t");
if (ltrim == std::string::npos) {
ltrim = 0;
}
size_t rtrim = str.find_last_not_of(" \t");
if (ltrim != 0 || rtrim != str.length() - 1) {
return str.substr(ltrim, rtrim - ltrim + 1);
}
return str;
size_t ltrim = str.find_first_not_of(" \t");
if (ltrim == std::string::npos) {
ltrim = 0;
}
size_t rtrim = str.find_last_not_of(" \t");
if (ltrim != 0 || rtrim != str.length() - 1) {
return str.substr(ltrim, rtrim - ltrim + 1);
}
return str;
}
+43 -40
View File
@@ -1,52 +1,55 @@
#ifndef OCELOT_CONFIG_H
#define OCELOT_CONFIG_H
#ifndef SRC_CONFIG_H_
#define SRC_CONFIG_H_
// Copyright [2017-2024] Orpheus
#include <string>
#include <map>
class confval {
private:
bool bool_val;
uint32_t uint_val;
std::string str_val;
enum {
CONF_NONEXISTENT,
CONF_BOOL,
CONF_UINT,
CONF_STR,
} val_type;
public:
confval();
confval(bool value);
confval(uint32_t value);
confval(const char * value);
uint32_t get_uint();
bool get_bool();
std::string get_str();
void set(const std::string &val);
private:
bool bool_val;
uint32_t uint_val;
std::string str_val;
enum {
CONF_NONEXISTENT,
CONF_BOOL,
CONF_UINT,
CONF_STR,
} val_type;
public:
confval();
confval(bool value);
confval(uint32_t value);
confval(const char * value);
uint32_t get_uint();
bool get_bool();
std::string get_str();
void set(const std::string &val);
};
class config {
private:
template <typename T> void add(const std::string &setting_name, T value);
std::string trim(const std::string str);
void init();
confval * get(const std::string &setting_name);
std::map<std::string, confval> settings;
confval * dummy_setting;
public:
config();
void load(std::istream &conf_file);
void load(const std::string &conf_file_path, std::istream &conf_file);
void reload();
bool get_bool(const std::string &setting_name);
uint32_t get_uint(const std::string &setting_name);
std::string get_str(const std::string &setting_name);
void set(const std::string &setting_name, const std::string &value);
private:
template <typename T> void add(const std::string &setting_name, T value);
std::string trim(const std::string str);
void init();
confval * get(const std::string &setting_name);
std::map<std::string, confval> settings;
confval * dummy_setting;
public:
config();
void load(std::istream &conf_file);
void load(const std::string &conf_file_path, std::istream &conf_file);
void reload();
bool get_bool(const std::string &setting_name);
uint32_t get_uint(const std::string &setting_name);
std::string get_str(const std::string &setting_name);
void set(const std::string &setting_name, const std::string &value);
};
template <typename T> void config::add(const std::string &setting_name, T value) {
confval setting(value);
settings[setting_name] = setting;
confval setting(value);
settings[setting_name] = setting;
}
#endif
#endif // SRC_CONFIG_H_
+558 -553
View File
File diff suppressed because it is too large. Load diff
+77 -63
View File
@@ -1,88 +1,102 @@
#ifndef OCELOT_DB_H
#define OCELOT_DB_H
#ifndef SRC_DB_H_
#define SRC_DB_H_
// Copyright [2017-2024] Orpheus
#pragma GCC visibility push(default)
#include <mysql++/mysql++.h>
#include <spdlog/spdlog.h>
#include <string>
#include <unordered_map>
#include <queue>
#include <mutex>
#include <memory>
#include <vector>
#include "config.h"
class mysql {
private:
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;
private:
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;
std::queue<std::string> user_queue;
std::queue<std::string> torrent_queue;
std::queue<std::string> peer_queue;
std::queue<std::string> snatch_queue;
std::queue<std::string> token_queue;
std::queue<std::string> user_queue;
std::queue<std::string> torrent_queue;
std::queue<std::string> peer_queue;
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;
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;
// 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::shared_ptr<spdlog::logger> logger;
std::shared_ptr<spdlog::logger> logger;
void load_config(config * conf);
void load_tokens(torrent_list &torrents);
mysqlpp::Connection create_connection();
void load_config(config * conf);
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_users();
void do_flush_torrents();
void do_flush_snatches();
void do_flush_peers();
void do_flush_tokens();
void flush_users();
void flush_torrents();
void flush_snatches();
void flush_peers();
void flush_tokens();
void clear_peer_data();
void flush_users();
void flush_torrents();
void flush_snatches();
void flush_peers();
void flush_tokens();
void clear_peer_data();
public:
bool verbose_flush;
public:
bool verbose_flush;
mysql(config * conf);
void reload_config(config * conf);
bool connected();
void load_torrents(torrent_list &torrents);
void load_users(user_list &users);
void load_whitelist(std::vector<std::string> &whitelist);
mysql(config * conf);
void reload_config(config * conf);
void flush();
bool connected();
bool all_clear();
void load_torrents(torrent_list &torrents);
void load_users(user_list &users);
void load_whitelist(std::vector<std::string> &whitelist);
void record_user(const std::string &record); // (id,uploaded_change,downloaded_change)
void record_torrent(const std::string &record); // (id,seeders,leechers,snatched_change,balance)
void record_snatch(const std::string &record, const std::string &ip); // (uid,fid,tstamp)
void record_peer(const std::string &record, const std::string &ip, const std::string &peer_id, const std::string &useragent); // (uid,fid,active,peerid,useragent,ip,uploaded,downloaded,upspeed,downspeed,left,timespent,announces,tstamp)
void record_peer(const std::string &record, const std::string &peer_id); // (fid,peerid,timespent,announces,tstamp)
void record_token(const std::string &record);
// (id,uploaded_change,downloaded_change)
void record_user(const std::string &record);
void flush();
// (id,seeders,leechers,snatched_change,balance)
void record_torrent(const std::string &record);
bool all_clear();
// (uid,fid,tstamp)
void record_snatch(const std::string &record, const std::string &ip);
std::mutex torrent_list_mutex;
std::mutex user_list_mutex;
std::mutex whitelist_mutex;
// (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
#endif
#endif // SRC_DB_H_
+171 -170
View File
@@ -1,5 +1,9 @@
#include <cerrno>
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <cerrno>
#include "ocelot.h"
#include "config.h"
#include "db.h"
@@ -10,235 +14,232 @@
// Define the connection mother (first half) and connection middlemen (second half)
//TODO Better errors
//---------- 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) {
logger = spdlog::get("logger");
logger = spdlog::get("logger");
// Handle config stuff first
load_config(conf);
// Handle config stuff first
load_config(conf);
listen_socket = create_listen_socket();
listen_socket = create_listen_socket();
// We failed to create the socket, so no point in running Ocelot
if (listen_socket == 0) {
logger->critical("Failed to create socket for Ocelot. Exiting.");
exit(1);
}
// We failed to create the socket, so no point in running Ocelot
if (listen_socket == 0) {
logger->critical("Failed to create socket for Ocelot. Exiting.");
exit(1);
}
listen_event.set<connection_mother, &connection_mother::handle_connect>(this);
listen_event.start(listen_socket, ev::READ);
// Create libev timer
schedule_event.set<schedule, &schedule::handle>(sched);
schedule_event.start(sched->schedule_interval, sched->schedule_interval); // After interval, every interval
listen_event.set<connection_mother, &connection_mother::handle_connect>(this);
listen_event.start(listen_socket, ev::READ);
// Create libev timer
schedule_event.set<schedule, &schedule::handle>(sched);
schedule_event.start(sched->schedule_interval, sched->schedule_interval); // After interval, every interval
}
void connection_mother::load_config(config * conf) {
listen_port = conf->get_uint("listen_port");
max_connections = conf->get_uint("max_connections");
max_middlemen = conf->get_uint("max_middlemen");
connection_timeout = conf->get_uint("connection_timeout");
keepalive_timeout = conf->get_uint("keepalive_timeout");
max_read_buffer = conf->get_uint("max_read_buffer");
max_request_size = conf->get_uint("max_request_size");
listen_port = conf->get_uint("listen_port");
max_connections = conf->get_uint("max_connections");
max_middlemen = conf->get_uint("max_middlemen");
connection_timeout = conf->get_uint("connection_timeout");
keepalive_timeout = conf->get_uint("keepalive_timeout");
max_read_buffer = conf->get_uint("max_read_buffer");
max_request_size = conf->get_uint("max_request_size");
}
void connection_mother::reload_config(config * conf) {
unsigned int old_listen_port = listen_port;
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));
int new_listen_socket = create_listen_socket();
if (new_listen_socket != 0) {
listen_event.stop();
listen_event.start(new_listen_socket, ev::READ);
close(listen_socket);
listen_socket = new_listen_socket;
} else {
logger->error("Couldn't create new listen socket when reloading config");
}
} else if (old_max_connections != max_connections) {
listen(listen_socket, max_connections);
}
unsigned int old_listen_port = listen_port;
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));
int new_listen_socket = create_listen_socket();
if (new_listen_socket != 0) {
listen_event.stop();
listen_event.start(new_listen_socket, ev::READ);
close(listen_socket);
listen_socket = new_listen_socket;
} else {
logger->error("Couldn't create new listen socket when reloading config");
}
} else if (old_max_connections != max_connections) {
listen(listen_socket, max_connections);
}
}
int connection_mother::create_listen_socket() {
sockaddr_in address;
memset(&address, 0, sizeof(address));
int new_listen_socket = socket(AF_INET, SOCK_STREAM, 0);
sockaddr_in address;
memset(&address, 0, sizeof(address));
int new_listen_socket = socket(AF_INET, SOCK_STREAM, 0);
// 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)));
return 0;
}
// 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)));
return 0;
}
// Get ready to bind
address.sin_family = AF_INET;
//address.sin_addr.s_addr = inet_addr(conf->host.c_str()); // htonl(INADDR_ANY)
address.sin_addr.s_addr = htonl(INADDR_ANY);
address.sin_port = htons(listen_port);
// Get ready to bind
address.sin_family = AF_INET;
address.sin_addr.s_addr = htonl(INADDR_ANY);
address.sin_port = htons(listen_port);
// Bind
if (bind(new_listen_socket, (sockaddr *) &address, sizeof(address)) == -1) {
logger->error("Bind failed: " + std::string(strerror(errno)));
return 0;
}
// Bind
if (bind(new_listen_socket, (sockaddr *) &address, sizeof(address)) == -1) {
logger->error("Bind failed: " + std::string(strerror(errno)));
return 0;
}
// Listen
if (listen(new_listen_socket, max_connections) == -1) {
logger->error("Listen failed: " + std::string(strerror(errno)));
return 0;
}
// Listen
if (listen(new_listen_socket, max_connections) == -1) {
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: " + std::string(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)));
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)));
return 0;
}
if (fcntl(new_listen_socket, F_SETFL, flags | O_NONBLOCK) == -1) {
logger->error("Could not set non-blocking: " + std::string(strerror(errno)));
return 0;
}
return new_listen_socket;
return new_listen_socket;
}
void connection_mother::run() {
logger->info("Sockets up on port " + std::to_string(listen_port) + ", starting event loop!");
ev_loop(ev_default_loop(0), 0);
logger->info("Sockets up on port " + std::to_string(listen_port) + ", starting event loop!");
ev_loop(ev_default_loop(0), 0);
}
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++;
new connection_middleman(listen_socket, work, this);
}
// Spawn a new middleman
if (stats.open_connections < max_middlemen) {
stats.opened_connections++;
stats.open_connections++;
new connection_middleman(listen_socket, work, this);
}
}
connection_mother::~connection_mother()
{
close(listen_socket);
close(listen_socket);
}
//---------- Connection middlemen - these little guys live until their connection is closed
connection_middleman::connection_middleman(int &listen_socket, worker * new_work, connection_mother * mother_arg) :
written(0), mother(mother_arg), work(new_work)
written(0), mother(mother_arg), work(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)));
delete this;
return;
}
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)));
delete this;
return;
}
// Set non-blocking
int flags = fcntl(connect_sock, F_GETFL);
if (flags == -1) {
logger->warn("Could not get connect socket flags");
}
if (fcntl(connect_sock, F_SETFL, flags | O_NONBLOCK) == -1) {
logger->warn("Could not set non-blocking");
}
// Set non-blocking
int flags = fcntl(connect_sock, F_GETFL);
if (flags == -1) {
logger->warn("Could not get connect socket flags");
}
if (fcntl(connect_sock, F_SETFL, flags | O_NONBLOCK) == -1) {
logger->warn("Could not set non-blocking");
}
// Get their info
request.reserve(mother->max_read_buffer);
written = 0;
// Get their info
request.reserve(mother->max_read_buffer);
written = 0;
read_event.set<connection_middleman, &connection_middleman::handle_read>(this);
read_event.start(connect_sock, ev::READ);
read_event.set<connection_middleman, &connection_middleman::handle_read>(this);
read_event.start(connect_sock, ev::READ);
// Let the socket timeout in timeout_interval seconds
timeout_event.set<connection_middleman, &connection_middleman::handle_timeout>(this);
timeout_event.set(mother->connection_timeout, mother->keepalive_timeout);
timeout_event.start();
// Let the socket timeout in timeout_interval seconds
timeout_event.set<connection_middleman, &connection_middleman::handle_timeout>(this);
timeout_event.set(mother->connection_timeout, mother->keepalive_timeout);
timeout_event.start();
}
connection_middleman::~connection_middleman() {
close(connect_sock);
stats.open_connections--;
close(connect_sock);
stats.open_connections--;
}
// Handler to read data from the socket, called by event loop when socket is readable
void connection_middleman::handle_read(ev::io &watcher, int events_flags) {
char buffer[mother->max_read_buffer + 1];
memset(buffer, 0, mother->max_read_buffer + 1);
int ret = recv(connect_sock, &buffer, mother->max_read_buffer, 0);
char buffer[mother->max_read_buffer + 1];
memset(buffer, 0, mother->max_read_buffer + 1);
int ret = recv(connect_sock, &buffer, mother->max_read_buffer, 0);
if (ret <= 0) {
delete this;
return;
}
stats.bytes_read += ret;
request.append(buffer, ret);
size_t request_size = request.size();
if (request_size > mother->max_request_size || (request_size >= 4 && request.compare(request_size - 4, std::string::npos, "\r\n\r\n") == 0)) {
stats.requests++;
read_event.stop();
client_opts.gzip = false;
client_opts.html = false;
client_opts.http_close = true;
if (ret <= 0) {
delete this;
return;
}
stats.bytes_read += ret;
request.append(buffer, ret);
size_t request_size = request.size();
if (request_size > mother->max_request_size || (request_size >= 4 && request.compare(request_size - 4, std::string::npos, "\r\n\r\n") == 0)) {
stats.requests++;
read_event.stop();
client_opts.gzip = false;
client_opts.html = false;
client_opts.http_close = true;
if (request_size > mother->max_request_size) {
shutdown(connect_sock, SHUT_RD);
response = error("GET string too long", client_opts);
} else {
char ip[INET_ADDRSTRLEN];
sockaddr_in client_addr;
socklen_t addr_len = sizeof(client_addr);
getpeername(connect_sock, (sockaddr *) &client_addr, &addr_len);
inet_ntop(AF_INET, &(client_addr.sin_addr), ip, INET_ADDRSTRLEN);
std::string ip_str = ip;
if (request_size > mother->max_request_size) {
shutdown(connect_sock, SHUT_RD);
response = error("GET string too long", client_opts);
} else {
char ip[INET_ADDRSTRLEN];
sockaddr_in client_addr;
socklen_t addr_len = sizeof(client_addr);
getpeername(connect_sock, (sockaddr *) &client_addr, &addr_len);
inet_ntop(AF_INET, &(client_addr.sin_addr), ip, INET_ADDRSTRLEN);
std::string ip_str = ip;
//--- CALL WORKER
response = work->work(request, ip_str, client_opts);
request.clear();
request_size = 0;
}
//--- CALL WORKER
response = work->work(request, ip_str, client_opts);
request.clear();
request_size = 0;
}
// Find out when the socket is writeable.
// The loop in connection_mother will call handle_write when it is.
write_event.set<connection_middleman, &connection_middleman::handle_write>(this);
write_event.start(connect_sock, ev::WRITE);
}
// Find out when the socket is writeable.
// The loop in connection_mother will call handle_write when it is.
write_event.set<connection_middleman, &connection_middleman::handle_write>(this);
write_event.start(connect_sock, ev::WRITE);
}
}
// Handler to write data to the socket, called by event loop when socket is writeable
void connection_middleman::handle_write(ev::io &watcher, int events_flags) {
int ret = send(connect_sock, response.c_str()+written, response.size()-written, MSG_NOSIGNAL);
if (ret == -1) {
return;
}
stats.bytes_written += ret;
written += ret;
if (written == response.size()) {
write_event.stop();
if (client_opts.http_close) {
timeout_event.stop();
delete this;
return;
}
timeout_event.again();
read_event.start();
response.clear();
written = 0;
}
int ret = send(connect_sock, response.c_str()+written, response.size()-written, MSG_NOSIGNAL);
if (ret == -1) {
return;
}
stats.bytes_written += ret;
written += ret;
if (written == response.size()) {
write_event.stop();
if (client_opts.http_close) {
timeout_event.stop();
delete this;
return;
}
timeout_event.again();
read_event.start();
response.clear();
written = 0;
}
}
// After a middleman has been alive for timout_interval seconds, this is called
void connection_middleman::handle_timeout(ev::timer &watcher, int events_flags) {
timeout_event.stop();
read_event.stop();
write_event.stop();
delete this;
timeout_event.stop();
read_event.stop();
write_event.stop();
delete this;
}
+64 -70
View File
@@ -1,3 +1,8 @@
#ifndef SRC_EVENTS_H_
#define SRC_EVENTS_H_
// Copyright [2017-2024] Orpheus
#ifdef EV_ERROR
#undef EV_ERROR
#endif
@@ -6,19 +11,20 @@
#define MSG_NOSIGNAL 0x2000 /* don't raise SIGPIPE */
#endif
#include <iostream>
#include <string>
#include <cstring>
// Sockets
#include <arpa/inet.h>
#include <fcntl.h>
#include <spdlog/spdlog.h>
#include <sys/socket.h>
#include <unistd.h>
// libev
#include <ev++.h>
// Sockets
#include <sys/socket.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <unistd.h>
#include <iostream>
#include <cstring>
#include <memory>
#include <string>
#include "config.h"
#include "worker.h"
@@ -26,88 +32,76 @@
#include "db.h"
#include "site_comm.h"
/*
TODO find out what these do
#include <netinet/in.h>
#include <netdb.h>
#include <sys/types.h>
*/
/*
We have three classes - the mother, the middlemen, and the worker
THE MOTHER
The mother is called when a client opens a connection to the server.
It creates a middleman for every new connection, which will be called
when its socket is ready for reading.
The mother is called when a client opens a connection to the server.
It creates a middleman for every new connection, which will be called
when its socket is ready for reading.
THE MIDDLEMEN
Each middleman hang around until data is written to its socket. It then
reads the data and sends it to the worker. When it gets the response, it
gets called to write its data back to the client.
Each middleman hang around until data is written to its socket. It then
reads the data and sends it to the worker. When it gets the response, it
gets called to write its data back to the client.
THE WORKER
The worker gets data from the middleman, and returns the response. It
doesn't concern itself with silly things like sockets.
The worker gets data from the middleman, and returns the response. It
doesn't concern itself with silly things like sockets.
see worker.h for the worker.
see worker.h for the worker.
*/
// THE MOTHER - Spawns connection middlemen
class connection_mother {
private:
void load_config(config * conf);
unsigned int listen_port;
unsigned int max_connections;
private:
void load_config(config * conf);
unsigned int listen_port;
unsigned int max_connections;
int listen_socket;
worker * work;
mysql * db;
ev::io listen_event;
ev::timer schedule_event;
std::shared_ptr<spdlog::logger> logger;
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();
void reload_config(config * conf);
int create_listen_socket();
void run();
void handle_connect(ev::io &watcher, int events_flags);
public:
connection_mother(config * conf, worker * worker_obj, mysql * db_obj, site_comm * sc_obj, schedule * sched_obj);
~connection_mother();
void reload_config(config * conf);
int create_listen_socket();
void run();
void handle_connect(ev::io &watcher, int events_flags);
unsigned int max_middlemen;
unsigned int connection_timeout;
unsigned int keepalive_timeout;
unsigned int max_read_buffer;
unsigned int max_request_size;
unsigned int max_middlemen;
unsigned int connection_timeout;
unsigned int keepalive_timeout;
unsigned int max_read_buffer;
unsigned int max_request_size;
};
// THE MIDDLEMAN
// Created by connection_mother
// Add their own watchers to see when sockets become readable
class connection_middleman {
private:
int connect_sock;
client_opts_t client_opts;
unsigned int written;
ev::io read_event;
ev::io write_event;
ev::timer timeout_event;
std::string request;
std::string response;
private:
int connect_sock;
client_opts_t client_opts;
unsigned int written;
ev::io read_event;
ev::io write_event;
ev::timer timeout_event;
std::string request;
std::string response;
connection_mother * mother;
worker * work;
connection_mother * mother;
worker * work;
public:
connection_middleman(int &listen_socket, worker* work, connection_mother * mother_arg);
~connection_middleman();
public:
connection_middleman(int &listen_socket, worker* work, connection_mother * mother_arg);
~connection_middleman();
void handle_read(ev::io &watcher, int events_flags);
void handle_write(ev::io &watcher, int events_flags);
void handle_timeout(ev::timer &watcher, int events_flags);
void handle_read(ev::io &watcher, int events_flags);
void handle_write(ev::io &watcher, int events_flags);
void handle_timeout(ev::timer &watcher, int events_flags);
};
#endif // SRC_EVENTS_H_
+66 -62
View File
@@ -1,81 +1,85 @@
#include <string>
// Copyright [2017-2024] Orpheus
#include <iostream>
#include <sstream>
#include <string>
#include "misc_functions.h"
int32_t strtoint32(const std::string& str) {
std::istringstream stream(str);
int32_t i = 0;
stream >> i;
return i;
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::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 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);
unsigned int in_length = in.length();
for (unsigned int i = 0; i < in_length; i++) {
unsigned char x = '0';
if (in[i] == '%' && (i + 2) < in_length) {
i++;
if (in[i] >= 'a' && in[i] <= 'f') {
x = static_cast<unsigned char>((in[i]-87) << 4);
} else if (in[i] >= 'A' && in[i] <= 'F') {
x = static_cast<unsigned char>((in[i]-55) << 4);
} else if (in[i] >= '0' && in[i] <= '9') {
x = static_cast<unsigned char>((in[i]-48) << 4);
}
std::string out;
out.reserve(20);
unsigned int in_length = in.length();
for (unsigned int i = 0; i < in_length; i++) {
unsigned char x = '0';
if (in[i] == '%' && (i + 2) < in_length) {
i++;
if (in[i] >= 'a' && in[i] <= 'f') {
x = static_cast<unsigned char>((in[i]-87) << 4);
} else if (in[i] >= 'A' && in[i] <= 'F') {
x = static_cast<unsigned char>((in[i]-55) << 4);
} else if (in[i] >= '0' && in[i] <= '9') {
x = static_cast<unsigned char>((in[i]-48) << 4);
}
i++;
if (in[i] >= 'a' && in[i] <= 'f') {
x += static_cast<unsigned char>(in[i]-87);
} else if (in[i] >= 'A' && in[i] <= 'F') {
x += static_cast<unsigned char>(in[i]-55);
} else if (in[i] >= '0' && in[i] <= '9') {
x += static_cast<unsigned char>(in[i]-48);
}
} else {
x = in[i];
}
out.push_back(x);
}
return out;
i++;
if (in[i] >= 'a' && in[i] <= 'f') {
x += static_cast<unsigned char>(in[i]-87);
} else if (in[i] >= 'A' && in[i] <= 'F') {
x += static_cast<unsigned char>(in[i]-55);
} else if (in[i] >= '0' && in[i] <= '9') {
x += static_cast<unsigned char>(in[i]-48);
}
} else {
x = in[i];
}
out.push_back(x);
}
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;
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;
}
+6 -3
View File
@@ -1,5 +1,8 @@
#ifndef MISC_FUNCTIONS__H
#define MISC_FUNCTIONS__H
#ifndef SRC_MISC_FUNCTIONS_H_
#define SRC_MISC_FUNCTIONS_H_
// Copyright [2017-2024] Orpheus
#include <string>
int32_t strtoint32(const std::string& str);
@@ -8,4 +11,4 @@ std::string inttostr(int i);
std::string hex_decode(const std::string &in);
std::string bintohex(const std::string &in);
#endif
#endif // SRC_MISC_FUNCTIONS_H_
+152 -155
View File
@@ -1,12 +1,16 @@
#include <fstream>
#include <iostream>
#include <csignal>
#include <thread>
// Copyright [2017-2024] Orpheus
#include <sys/stat.h>
#include <syslog.h>
#include <spdlog/spdlog.h>
#include <spdlog/sinks/daily_file_sink.h>
#include <spdlog/sinks/stdout_sinks.h>
#include <fstream>
#include <iostream>
#include <csignal>
#include <thread>
#include "ocelot.h"
#include "config.h"
#include "db.h"
@@ -22,190 +26,183 @@ static schedule *sched;
struct stats_t stats;
const char * version()
{
return "2.1.0";
const char * version() {
return "2.1.1";
}
static void create_daemon()
{
pid_t pid;
static void create_daemon() {
pid_t pid;
/* Fork off the parent process */
pid = fork();
/* Fork off the parent process */
pid = fork();
/* An error occurred */
if (pid < 0) {
exit(EXIT_FAILURE);
}
/* An error occurred */
if (pid < 0) {
exit(EXIT_FAILURE);
}
/* Success: Let the parent terminate */
if (pid > 0) {
exit(EXIT_SUCCESS);
}
/* Success: Let the parent terminate */
if (pid > 0) {
exit(EXIT_SUCCESS);
}
/* On success: The child process becomes session leader */
if (setsid() < 0)
exit(EXIT_FAILURE);
/* On success: The child process becomes session leader */
if (setsid() < 0)
exit(EXIT_FAILURE);
/* Set new file permissions */
umask(0);
/* Set new file permissions */
umask(0);
/* Change the working directory to the root directory */
/* or another appropriated directory */
if (chdir("/") < 0) {
exit(EXIT_FAILURE);
}
/* Change the working directory to the root directory */
/* or another appropriated directory */
if (chdir("/") < 0) {
exit(EXIT_FAILURE);
}
/* Close all open file descriptors */
int x;
for (x = static_cast<int>(sysconf(_SC_OPEN_MAX)); x >= 0; x--) {
close (x);
}
/* Close all open file descriptors */
int x;
for (x = static_cast<int>(sysconf(_SC_OPEN_MAX)); x >= 0; x--) {
close(x);
}
/* Open the log file */
openlog("ocelot", LOG_PID, LOG_DAEMON);
/* Open the log file */
openlog("ocelot", LOG_PID, LOG_DAEMON);
}
static void sig_handler(int sig) {
std::shared_ptr<spdlog::logger> logger = spdlog::get("logger");
if (sig == SIGINT || sig == SIGTERM) {
logger->info("Caught SIGINT/SIGTERM");
if (work->shutdown()) {
exit(0);
}
} else if (sig == SIGHUP) {
logger->info("Reloading config");
logger->flush();
conf->reload();
db->reload_config(conf);
mother->reload_config(conf);
sc->reload_config(conf);
sched->reload_config(conf);
work->reload_config(conf);
logger->info("Done reloading config");
} else if (sig == SIGUSR1) {
logger->info("Reloading from database");
std::thread w_thread(&worker::reload_lists, work);
w_thread.detach();
}
std::shared_ptr<spdlog::logger> logger = spdlog::get("logger");
if (sig == SIGINT || sig == SIGTERM) {
logger->info("Caught SIGINT/SIGTERM");
if (work->shutdown()) {
exit(0);
}
} else if (sig == SIGHUP) {
logger->info("Reloading config");
logger->flush();
conf->reload();
db->reload_config(conf);
mother->reload_config(conf);
sc->reload_config(conf);
sched->reload_config(conf);
work->reload_config(conf);
logger->info("Done reloading config");
} else if (sig == SIGUSR1) {
logger->info("Reloading from database");
std::thread w_thread(&worker::reload_lists, work);
w_thread.detach();
}
}
int main(int argc, char **argv) {
// we don't use printf so make cout/cerr a little bit faster
std::ios_base::sync_with_stdio(false);
// we don't use printf so make cout/cerr a little bit faster
std::ios_base::sync_with_stdio(false);
conf = new config();
conf = new config();
bool conf_arg = false;
bool daemonize = false;
bool verbose = false;
std::string conf_file_path("./ocelot.conf");
bool conf_arg = false;
bool daemonize = false;
bool verbose = false;
std::string conf_file_path("./ocelot.conf");
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-v") == 0) {
verbose = true;
}
else if (strcmp(argv[i], "-d") == 0 || strcmp(argv[i], "--daemonize") == 0) {
daemonize = true;
}
else if (strcmp(argv[i], "-c") == 0 && i < argc - 1) {
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;
return 0;
}
else {
std::cout << "Usage: " << argv[0] << " [-v] [-c configfile] [--daemonize]" << std::endl;
return 0;
}
}
for (int i = 1; i < argc; i++) {
if (strcmp(argv[i], "-v") == 0) {
verbose = true;
} else if (strcmp(argv[i], "-d") == 0 || strcmp(argv[i], "--daemonize") == 0) {
daemonize = true;
} else if (strcmp(argv[i], "-c") == 0 && i < argc - 1) {
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;
return 0;
} else {
std::cout << "Usage: " << argv[0] << " [-v] [-c configfile] [--daemonize]" << std::endl;
return 0;
}
}
std::ifstream conf_file(conf_file_path);
if (conf_file.fail()) {
std::cout << "Using default config because '" << conf_file_path << "' couldn't be opened" << std::endl;
if (!conf_arg) {
std::cout << "Start Ocelot with -c <path> to specify config file if necessary" << std::endl;
}
}
else {
conf->load(conf_file_path, conf_file);
}
std::ifstream conf_file(conf_file_path);
if (conf_file.fail()) {
std::cout << "Using default config because '" << conf_file_path << "' couldn't be opened" << std::endl;
if (!conf_arg) {
std::cout << "Start Ocelot with -c <path> to specify config file if necessary" << std::endl;
}
} else {
conf->load(conf_file_path, conf_file);
}
if (conf->get_bool("daemonize") || daemonize) {
create_daemon();
}
if (conf->get_bool("daemonize") || daemonize) {
create_daemon();
}
std::vector<spdlog::sink_ptr> sinks;
if (!conf->get_bool("daemonize") && !daemonize) {
sinks.push_back(std::make_shared<spdlog::sinks::stdout_sink_mt>());
}
if (conf->get_bool("log") && !conf->get_str("log_path").empty()) {
sinks.push_back(std::make_shared<spdlog::sinks::daily_file_sink_mt>(conf->get_str("log_path"), 23, 59));
}
std::vector<spdlog::sink_ptr> sinks;
if (!conf->get_bool("daemonize") && !daemonize) {
sinks.push_back(std::make_shared<spdlog::sinks::stdout_sink_mt>());
}
if (conf->get_bool("log") && !conf->get_str("log_path").empty()) {
sinks.push_back(std::make_shared<spdlog::sinks::daily_file_sink_mt>(conf->get_str("log_path"), 23, 59));
}
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);
spdlog::register_logger(combined_logger);
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);
spdlog::register_logger(combined_logger);
db = new mysql(conf);
db = new mysql(conf);
if (!db->connected()) {
combined_logger->critical("Could not connect to DB. Exiting!");
return 0;
}
db->verbose_flush = verbose;
if (!db->connected()) {
combined_logger->critical("Could not connect to DB. Exiting!");
return 0;
}
db->verbose_flush = verbose;
sc = new site_comm(conf);
sc->verbose_flush = verbose;
sc = new site_comm(conf);
sc->verbose_flush = verbose;
user_list users_list;
torrent_list torrents_list;
std::vector<std::string> whitelist;
db->load_users(users_list);
db->load_torrents(torrents_list);
db->load_whitelist(whitelist);
user_list users_list;
torrent_list torrents_list;
std::vector<std::string> whitelist;
db->load_users(users_list);
db->load_torrents(torrents_list);
db->load_whitelist(whitelist);
stats.open_connections = 0;
stats.opened_connections = 0;
stats.connection_rate = 0;
stats.requests = 0;
stats.request_rate = 0;
stats.leechers = 0;
stats.seeders = 0;
stats.announcements = 0;
stats.succ_announcements = 0;
stats.scrapes = 0;
stats.bytes_read = 0;
stats.bytes_written = 0;
stats.start_time = time(NULL);
stats.open_connections = 0;
stats.opened_connections = 0;
stats.connection_rate = 0;
stats.requests = 0;
stats.request_rate = 0;
stats.leechers = 0;
stats.seeders = 0;
stats.announcements = 0;
stats.succ_announcements = 0;
stats.scrapes = 0;
stats.bytes_read = 0;
stats.bytes_written = 0;
stats.start_time = time(NULL);
// Create worker object, which handles announces and scrapes and all that jazz
work = new worker(conf, torrents_list, users_list, whitelist, db, sc);
// Create worker object, which handles announces and scrapes and all that jazz
work = new worker(conf, torrents_list, users_list, whitelist, db, sc);
// Create schedule object
sched = new schedule(conf, work, db, sc);
// Create schedule object
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);
// Create connection mother, which binds to its socket and handles the event stuff
mother = new connection_mother(conf, work, db, sc, sched);
// Add signal handlers now that all objects have been created
struct sigaction handler{}, ignore{};
ignore.sa_handler = SIG_IGN;
handler.sa_handler = sig_handler;
sigemptyset(&handler.sa_mask);
handler.sa_flags = 0;
// Add signal handlers now that all objects have been created
struct sigaction handler{}, ignore{};
ignore.sa_handler = SIG_IGN;
handler.sa_handler = sig_handler;
sigemptyset(&handler.sa_mask);
handler.sa_flags = 0;
sigaction(SIGINT, &handler, NULL);
sigaction(SIGTERM, &handler, NULL);
sigaction(SIGHUP, &handler, NULL);
sigaction(SIGUSR1, &handler, NULL);
sigaction(SIGUSR2, &ignore, NULL);
sigaction(SIGINT, &handler, NULL);
sigaction(SIGTERM, &handler, NULL);
sigaction(SIGHUP, &handler, NULL);
sigaction(SIGUSR1, &handler, NULL);
sigaction(SIGUSR2, &ignore, NULL);
mother->run();
mother->run();
return 0;
return 0;
}
+75 -72
View File
@@ -1,5 +1,9 @@
#ifndef OCELOT_H
#define OCELOT_H
#ifndef SRC_OCELOT_H_
#define SRC_OCELOT_H_
// Copyright [2017-2024] Orpheus
#include <time.h>
#include <string>
#include <map>
@@ -8,7 +12,6 @@
#include <set>
#include <memory>
#include <atomic>
#include <time.h>
typedef uint32_t torid_t;
typedef uint32_t userid_t;
@@ -17,19 +20,19 @@ class user;
typedef std::shared_ptr<user> user_ptr;
typedef struct {
int64_t uploaded;
int64_t downloaded;
int64_t corrupt;
int64_t left;
time_t last_announced;
time_t first_announced;
uint32_t announces;
uint16_t port;
bool visible;
bool invalid_ip;
user_ptr user;
std::string ip_port;
std::string ip;
int64_t uploaded;
int64_t downloaded;
int64_t corrupt;
int64_t left;
time_t last_announced;
time_t first_announced;
uint32_t announces;
uint16_t port;
bool visible;
bool invalid_ip;
user_ptr user;
std::string ip_port;
std::string ip;
} peer;
typedef std::map<std::string, peer> peer_list;
@@ -37,52 +40,52 @@ typedef std::map<std::string, peer> peer_list;
enum freetype { NORMAL, FREE, NEUTRAL };
typedef struct {
torid_t id;
uint32_t completed;
int64_t balance;
freetype free_torrent;
time_t last_flushed;
peer_list seeders;
peer_list leechers;
std::string last_selected_seeder;
std::set<userid_t> tokened_users;
torid_t id;
uint32_t completed;
int64_t balance;
freetype free_torrent;
time_t last_flushed;
peer_list seeders;
peer_list leechers;
std::string last_selected_seeder;
std::set<userid_t> tokened_users;
} torrent;
enum {
DUPE, // 0
TRUMP, // 1
BAD_FILE_NAMES, // 2
BAD_FOLDER_NAMES, // 3
BAD_TAGS, // 4
BAD_FORMAT, // 5
DISCS_MISSING, // 6
DISCOGRAPHY,// 7
EDITED_LOG,// 8
INACCURATE_BITRATE, // 9
LOW_BITRATE, // 10
MUTT_RIP,// 11
BAD_SOURCE,// 12
ENCODE_ERRORS,// 13
BANNED, // 14
TRACKS_MISSING,// 15
TRANSCODE, // 16
CASSETTE, // 17
UNSPLIT_ALBUM, // 18
USER_COMPILATION, // 19
WRONG_FORMAT, // 20
WRONG_MEDIA, // 21
AUDIENCE // 22
DUPE, // 0
TRUMP, // 1
BAD_FILE_NAMES, // 2
BAD_FOLDER_NAMES, // 3
BAD_TAGS, // 4
BAD_FORMAT, // 5
DISCS_MISSING, // 6
DISCOGRAPHY, // 7
EDITED_LOG, // 8
INACCURATE_BITRATE, // 9
LOW_BITRATE, // 10
MUTT_RIP, // 11
BAD_SOURCE, // 12
ENCODE_ERRORS, // 13
BANNED, // 14
TRACKS_MISSING, // 15
TRANSCODE, // 16
CASSETTE, // 17
UNSPLIT_ALBUM, // 18
USER_COMPILATION, // 19
WRONG_FORMAT, // 20
WRONG_MEDIA, // 21
AUDIENCE // 22
};
typedef struct {
int reason;
time_t time;
int reason;
time_t time;
} del_message;
typedef struct {
bool gzip;
bool html;
bool http_close;
bool gzip;
bool html;
bool http_close;
} client_opts_t;
typedef std::unordered_map<std::string, torrent> torrent_list;
@@ -90,26 +93,26 @@ typedef std::unordered_map<std::string, user_ptr> user_list;
typedef std::unordered_map<std::string, std::string> params_type;
struct stats_t {
std::atomic<uint32_t> open_connections;
std::atomic<uint64_t> opened_connections;
std::atomic<uint64_t> connection_rate;
std::atomic<uint32_t> leechers;
std::atomic<uint32_t> seeders;
std::atomic<uint32_t> user_queue_size;
std::atomic<uint32_t> torrent_queue_size;
std::atomic<uint32_t> peer_queue_size;
std::atomic<uint32_t> snatch_queue_size;
std::atomic<uint32_t> token_queue_size;
std::atomic<uint64_t> requests;
std::atomic<uint64_t> request_rate;
std::atomic<uint64_t> announcements;
std::atomic<uint64_t> succ_announcements;
std::atomic<uint64_t> scrapes;
std::atomic<uint64_t> bytes_read;
std::atomic<uint64_t> bytes_written;
time_t start_time;
std::atomic<uint32_t> open_connections;
std::atomic<uint64_t> opened_connections;
std::atomic<uint64_t> connection_rate;
std::atomic<uint32_t> leechers;
std::atomic<uint32_t> seeders;
std::atomic<uint32_t> user_queue_size;
std::atomic<uint32_t> torrent_queue_size;
std::atomic<uint32_t> peer_queue_size;
std::atomic<uint32_t> snatch_queue_size;
std::atomic<uint32_t> token_queue_size;
std::atomic<uint64_t> requests;
std::atomic<uint64_t> request_rate;
std::atomic<uint64_t> announcements;
std::atomic<uint64_t> succ_announcements;
std::atomic<uint64_t> scrapes;
std::atomic<uint64_t> bytes_read;
std::atomic<uint64_t> bytes_written;
time_t start_time;
};
extern struct stats_t stats;
extern const char *version();
#endif
#endif // SRC_OCELOT_H_
+79 -76
View File
@@ -1,6 +1,9 @@
// Copyright [2017-2024] Orpheus
#include <iostream>
#include <map>
#include <sstream>
#include "ocelot.h"
#include "misc_functions.h"
#include "report.h"
@@ -8,81 +11,81 @@
#include "user.h"
std::string report(params_type &params, 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;
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;
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";
}
output << "success";
return response(output.str(), client_opts);
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";
}
output << "success";
return response(output.str(), client_opts);
}
+13 -6
View File
@@ -1,9 +1,16 @@
#include "ocelot.h"
#ifndef SRC_REPORT_H_
#define SRC_REPORT_H_
// Copyright [2017-2024] Orpheus
#include <string>
std::string report(
params_type &params,
user_list &users_list,
client_opts_t &client_opts,
unsigned int announce_interval,
unsigned int announce_jitter
params_type &params,
user_list &users_list,
client_opts_t &client_opts,
unsigned int announce_interval,
unsigned int announce_jitter
);
#endif // SRC_REPORT_H_
+39 -34
View File
@@ -1,51 +1,56 @@
// Copyright [2017-2024] Orpheus
#include <algorithm>
#include <sstream>
#include <boost/iostreams/filtering_streambuf.hpp>
#include <boost/iostreams/copy.hpp>
#include <boost/iostreams/filter/gzip.hpp>
#include "response.h"
#include "misc_functions.h"
#include "response.h"
const std::string response(const std::string &body, client_opts_t &client_opts) {
std::string out;
bool processed = false;
if (client_opts.html) {
out = "<html><head><meta name=\"robots\" content=\"noindex, nofollow\" /></head><body>" + body + "</body></html>";
processed = true;
}
if (client_opts.gzip) {
std::stringstream ss, zss;
ss << body;
boost::iostreams::filtering_streambuf<boost::iostreams::input> in;
in.push(boost::iostreams::gzip_compressor());
in.push(ss);
boost::iostreams::copy(in, zss);
out = zss.str();
processed = true;
}
if (processed) {
return response_head(out.length(), client_opts) + out;
}
return response_head(body.length(), client_opts) + body;
std::string out;
bool processed = false;
if (client_opts.html) {
out = "<html><head><meta name=\"robots\" content=\"noindex, nofollow\" /></head><body>" + body + "</body></html>";
processed = true;
}
if (client_opts.gzip) {
std::stringstream ss, zss;
ss << body;
boost::iostreams::filtering_streambuf<boost::iostreams::input> in;
in.push(boost::iostreams::gzip_compressor());
in.push(ss);
boost::iostreams::copy(in, zss);
out = zss.str();
processed = true;
}
if (processed) {
return response_head(out.length(), client_opts) + out;
}
return response_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 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);
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;
return "15:warning message" + inttostr(msg.length()) + ':' + msg;
}
+7 -3
View File
@@ -1,11 +1,15 @@
#ifndef RESPONSE_H
#define RESPONSE_H
#ifndef SRC_RESPONSE_H_
#define SRC_RESPONSE_H_
// Copyright [2017-2024] Orpheus
#include <string>
#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 error(const std::string &err, client_opts_t &client_opts);
const std::string warning(const std::string &msg);
#endif
#endif // SRC_RESPONSE_H_
+38 -35
View File
@@ -1,6 +1,9 @@
#include <iostream>
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <iostream>
#include "ocelot.h"
#include "config.h"
#include "db.h"
@@ -8,53 +11,53 @@
#include "schedule.h"
schedule::schedule(config * conf, worker * worker_obj, mysql * db_obj, site_comm * sc_obj) : work(worker_obj), db(db_obj), sc(sc_obj) {
load_config(conf);
counter = 0;
last_opened_connections = 0;
last_request_count = 0;
next_reap_peers = reap_peers_interval;
logger = spdlog::get("logger");
load_config(conf);
counter = 0;
last_opened_connections = 0;
last_request_count = 0;
next_reap_peers = reap_peers_interval;
logger = spdlog::get("logger");
}
void schedule::load_config(config * conf) {
reap_peers_interval = conf->get_uint("reap_peers_interval");
schedule_interval = conf->get_uint("schedule_interval");
reap_peers_interval = conf->get_uint("reap_peers_interval");
schedule_interval = conf->get_uint("schedule_interval");
}
void schedule::reload_config(config * conf) {
load_config(conf);
load_config(conf);
}
//---------- Schedule - gets called every schedule_interval seconds
void schedule::handle(ev::timer &watcher, int events_flags) {
unsigned int cur_schedule_interval = watcher.repeat;
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)");
}
unsigned int cur_schedule_interval = watcher.repeat;
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)");
}
if (work->get_status() == CLOSING && db->all_clear() && sc->all_clear()) {
logger->info("all clear, shutting down");
exit(0);
}
if (work->get_status() == CLOSING && db->all_clear() && sc->all_clear()) {
logger->info("all clear, shutting down");
exit(0);
}
last_opened_connections = stats.opened_connections;
last_request_count = stats.requests;
last_opened_connections = stats.opened_connections;
last_request_count = stats.requests;
db->flush();
sc->flush_tokens();
db->flush();
sc->flush_tokens();
next_reap_peers -= cur_schedule_interval;
if (next_reap_peers <= 0) {
work->start_reaper();
next_reap_peers = reap_peers_interval;
}
next_reap_peers -= cur_schedule_interval;
if (next_reap_peers <= 0) {
work->start_reaper();
next_reap_peers = reap_peers_interval;
}
counter++;
if (schedule_interval != cur_schedule_interval) {
watcher.set(schedule_interval, schedule_interval);
}
counter++;
if (schedule_interval != cur_schedule_interval) {
watcher.set(schedule_interval, schedule_interval);
}
}
+24 -19
View File
@@ -1,29 +1,34 @@
#ifndef SCHEDULE_H
#define SCHEDULE_H
#ifndef SRC_SCHEDULE_H_
#define SRC_SCHEDULE_H_
// Copyright [2017-2024] Orpheus
#ifdef EV_ERROR
#undef EV_ERROR
#endif
#include <ev++.h>
class schedule {
private:
void load_config(config * conf);
private:
void load_config(config * conf);
unsigned int reap_peers_interval;
worker * work;
mysql * db;
site_comm * sc;
uint64_t last_opened_connections;
uint64_t last_request_count;
unsigned int counter;
int next_reap_peers;
std::shared_ptr<spdlog::logger> logger;
public:
schedule(config * conf, worker * worker_obj, mysql * db_obj, site_comm * sc_obj);
void reload_config(config * conf);
void handle(ev::timer &watcher, int events_flags);
unsigned int reap_peers_interval;
worker * work;
mysql * db;
site_comm * sc;
uint64_t last_opened_connections;
uint64_t last_request_count;
unsigned int counter;
int next_reap_peers;
std::shared_ptr<spdlog::logger> logger;
unsigned int schedule_interval;
public:
schedule(config * conf, worker * worker_obj, mysql * db_obj, site_comm * sc_obj);
void reload_config(config * conf);
void handle(ev::timer &watcher, int events_flags);
unsigned int schedule_interval;
};
#endif
#endif // SRC_SCHEDULE_H_
+98 -94
View File
@@ -1,3 +1,7 @@
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <iostream>
#include <istream>
#include <ostream>
@@ -5,9 +9,9 @@
#include <sstream>
#include <queue>
#include <mutex>
#include <boost/asio.hpp>
#include <thread>
#include <spdlog/spdlog.h>
#include <boost/asio.hpp>
#include "config.h"
#include "site_comm.h"
@@ -15,126 +19,126 @@
using boost::asio::ip::tcp;
site_comm::site_comm(config * conf) : t_active(false) {
logger = spdlog::get("logger");
load_config(conf);
logger = spdlog::get("logger");
load_config(conf);
}
void site_comm::load_config(config * conf) {
site_host = conf->get_str("site_host");
site_service = conf->get_str("site_service");
site_path = conf->get_str("site_path");
site_password = conf->get_str("site_password");
readonly = conf->get_bool("readonly");
site_host = conf->get_str("site_host");
site_service = conf->get_str("site_service");
site_path = conf->get_str("site_path");
site_password = conf->get_str("site_password");
readonly = conf->get_bool("readonly");
}
void site_comm::reload_config(config * conf) {
load_config(conf);
load_config(conf);
}
bool site_comm::all_clear() {
return (token_queue.size() == 0);
return (token_queue.size() == 0);
}
void site_comm::expire_token(int torrent, int user) {
std::stringstream token_pair;
token_pair << user << ':' << torrent;
if (!expire_token_buffer.empty()) {
expire_token_buffer += ",";
}
expire_token_buffer += token_pair.str();
if (expire_token_buffer.length() > 350) {
logger->info("Flushing overloaded token buffer");
if (!readonly) {
std::lock_guard<std::mutex> lock(expire_queue_lock);
token_queue.push(expire_token_buffer);
}
expire_token_buffer.clear();
}
std::stringstream token_pair;
token_pair << user << ':' << torrent;
if (!expire_token_buffer.empty()) {
expire_token_buffer += ",";
}
expire_token_buffer += token_pair.str();
if (expire_token_buffer.length() > 350) {
logger->info("Flushing overloaded token buffer");
if (!readonly) {
std::lock_guard<std::mutex> lock(expire_queue_lock);
token_queue.push(expire_token_buffer);
}
expire_token_buffer.clear();
}
}
void site_comm::flush_tokens()
{
if (readonly) {
expire_token_buffer.clear();
return;
}
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));
}
if (expire_token_buffer.empty()) {
return;
}
token_queue.push(expire_token_buffer);
expire_token_buffer.clear();
if (!t_active) {
std::thread thread(&site_comm::do_flush_tokens, this);
thread.detach();
}
if (readonly) {
expire_token_buffer.clear();
return;
}
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));
}
if (expire_token_buffer.empty()) {
return;
}
token_queue.push(expire_token_buffer);
expire_token_buffer.clear();
if (!t_active) {
std::thread thread(&site_comm::do_flush_tokens, this);
thread.detach();
}
}
void site_comm::do_flush_tokens()
{
t_active = true;
try {
while (token_queue.size() > 0) {
boost::asio::io_service io_service;
t_active = true;
try {
while (token_queue.size() > 0) {
boost::asio::io_service io_service;
tcp::resolver resolver(io_service);
tcp::resolver::query query(site_host, site_service);
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
tcp::resolver::iterator end;
tcp::resolver resolver(io_service);
tcp::resolver::query query(site_host, site_service);
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
tcp::resolver::iterator end;
tcp::socket socket(io_service);
boost::system::error_code error = boost::asio::error::host_not_found;
while (error && endpoint_iterator != end) {
socket.close();
socket.connect(*endpoint_iterator++, error);
}
if (error) {
throw boost::system::system_error(error);
}
tcp::socket socket(io_service);
boost::system::error_code error = boost::asio::error::host_not_found;
while (error && endpoint_iterator != end) {
socket.close();
socket.connect(*endpoint_iterator++, error);
}
if (error) {
throw boost::system::system_error(error);
}
boost::asio::streambuf request;
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";
boost::asio::streambuf request;
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";
boost::asio::write(socket, request);
boost::asio::write(socket, request);
boost::asio::streambuf response;
boost::asio::read_until(socket, response, "\r\n");
boost::asio::streambuf response;
boost::asio::read_until(socket, response, "\r\n");
std::istream response_stream(&response);
std::string http_version;
response_stream >> http_version;
unsigned int status_code;
response_stream >> status_code;
std::string status_message;
std::getline(response_stream, status_message);
std::istream response_stream(&response);
std::string http_version;
response_stream >> http_version;
unsigned int status_code;
response_stream >> status_code;
std::string status_message;
std::getline(response_stream, status_message);
if (!response_stream || http_version.substr(0, 5) != "HTTP/") {
logger->info("Invalid response");
continue;
}
if (!response_stream || http_version.substr(0, 5) != "HTTP/") {
logger->info("Invalid response");
continue;
}
if (status_code == 200) {
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!");
// TODO: add exponential backoff
std::this_thread::sleep_for(std::chrono::seconds(5));
}
}
} catch (std::exception &er) {
logger->error("Exception: " + std::string(er.what()));
}
t_active = false;
if (status_code == 200) {
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!");
// TODO: add exponential backoff
std::this_thread::sleep_for(std::chrono::seconds(5));
}
}
} catch (std::exception &er) {
logger->error("Exception: " + std::string(er.what()));
}
t_active = false;
}
site_comm::~site_comm()
+36 -28
View File
@@ -1,36 +1,44 @@
#ifndef OCELOT_SITE_COMM_H
#define OCELOT_SITE_COMM_H
#include <string>
#include <boost/asio.hpp>
#include <queue>
#include <mutex>
#ifndef SRC_SITE_COMM_H_
#define SRC_SITE_COMM_H_
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <string>
#include <queue>
#include <memory>
#include <mutex>
#include <boost/asio.hpp>
#include "config.h"
using boost::asio::ip::tcp;
class site_comm {
private:
std::string site_host;
std::string site_service;
std::string site_path;
std::string site_password;
std::mutex expire_queue_lock;
std::string expire_token_buffer;
std::queue<std::string> token_queue;
std::shared_ptr<spdlog::logger> logger;
bool readonly;
bool t_active;
void load_config(config * conf);
void do_flush_tokens();
private:
std::string site_host;
std::string site_service;
std::string site_path;
std::string site_password;
std::mutex expire_queue_lock;
std::string expire_token_buffer;
std::queue<std::string> token_queue;
std::shared_ptr<spdlog::logger> logger;
bool readonly;
bool t_active;
void load_config(config * conf);
void do_flush_tokens();
public:
bool verbose_flush;
site_comm(config * conf);
void reload_config(config * conf);
bool all_clear();
void expire_token(int torrent, int user);
void flush_tokens();
~site_comm();
public:
bool verbose_flush;
site_comm(config * conf);
void reload_config(config * conf);
bool all_clear();
void expire_token(int torrent, int user);
void flush_tokens();
~site_comm();
};
#endif
#endif // SRC_SITE_COMM_H_
+4 -2
View File
@@ -1,6 +1,8 @@
// Copyright [2017-2024] Orpheus
#include "user.h"
user::user(userid_t uid, bool leech, bool protect) : id(uid), deleted(false), leechstatus(leech), protect_ip(protect) {
stats.leeching = 0;
stats.seeding = 0;
stats.leeching = 0;
stats.seeding = 0;
}
+32 -27
View File
@@ -1,33 +1,38 @@
#ifndef USER_H
#define USER_H
#ifndef SRC_USER_H_
#define SRC_USER_H_
// Copyright [2017-2024] Orpheus
#include <atomic>
#include "ocelot.h"
class user {
private:
userid_t id;
bool deleted;
bool leechstatus;
bool protect_ip;
struct {
std::atomic<uint32_t> leeching;
std::atomic<uint32_t> seeding;
} stats;
public:
user(userid_t uid, bool leech, bool protect);
userid_t get_id() { return id; }
bool is_deleted() { return deleted; }
void set_deleted(bool status) { deleted = status; }
bool is_protected() { return protect_ip; }
void set_protected(bool status) { protect_ip = status; }
bool can_leech() { return leechstatus; }
void set_leechstatus(bool status) { leechstatus = status; }
void decr_leeching() { --stats.leeching; }
void decr_seeding() { --stats.seeding; }
void incr_leeching() { ++stats.leeching; }
void incr_seeding() { ++stats.seeding; }
uint32_t get_leeching() { return stats.leeching; }
uint32_t get_seeding() { return stats.seeding; }
private:
userid_t id;
bool deleted;
bool leechstatus;
bool protect_ip;
struct {
std::atomic<uint32_t> leeching;
std::atomic<uint32_t> seeding;
} stats;
public:
user(userid_t uid, bool leech, bool protect);
userid_t get_id() { return id; }
bool is_deleted() { return deleted; }
void set_deleted(bool status) { deleted = status; }
bool is_protected() { return protect_ip; }
void set_protected(bool status) { protect_ip = status; }
bool can_leech() { return leechstatus; }
void set_leechstatus(bool status) { leechstatus = status; }
void decr_leeching() { --stats.leeching; }
void decr_seeding() { --stats.seeding; }
void incr_leeching() { ++stats.leeching; }
void incr_seeding() { ++stats.seeding; }
uint32_t get_leeching() { return stats.leeching; }
uint32_t get_seeding() { return stats.seeding; }
};
#endif
#endif // SRC_USER_H_
+1055 -1053
View File
File diff suppressed because it is too large. Load diff
+50 -46
View File
@@ -1,67 +1,71 @@
#ifndef WORKER_H
#define WORKER_H
#ifndef SRC_WORKER_H_
#define SRC_WORKER_H_
// Copyright [2017-2024] Orpheus
#include <spdlog/spdlog.h>
#include <string>
#include <vector>
#include <list>
#include <unordered_map>
#include <iostream>
#include <memory>
#include <mutex>
#include <ctime>
#include <spdlog/spdlog.h>
#include <random>
#include "site_comm.h"
#include "ocelot.h"
enum tracker_status { OPEN, PAUSED, CLOSING }; // tracker status
enum tracker_status { OPEN, PAUSED, CLOSING }; // tracker status
class worker {
private:
config * conf;
mysql * db;
site_comm * s_comm;
torrent_list &torrents_list;
user_list &users_list;
std::vector<std::string> &whitelist;
std::unordered_map<std::string, del_message> del_reasons;
tracker_status status;
bool reaper_active;
time_t cur_time;
std::shared_ptr<spdlog::logger> logger;
std::mt19937 randgen;
std::uniform_int_distribution<int> jitter;
private:
config * conf;
mysql * db;
site_comm * s_comm;
torrent_list &torrents_list;
user_list &users_list;
std::vector<std::string> &whitelist;
std::unordered_map<std::string, del_message> del_reasons;
tracker_status status;
bool reaper_active;
time_t cur_time;
std::shared_ptr<spdlog::logger> logger;
std::mt19937 randgen;
std::uniform_int_distribution<int> jitter;
unsigned int announce_interval;
unsigned int del_reason_lifetime;
unsigned int peers_timeout;
unsigned int numwant_limit;
bool keepalive_enabled;
std::string site_password;
std::string report_password;
unsigned int announce_interval;
unsigned int del_reason_lifetime;
unsigned int peers_timeout;
unsigned int numwant_limit;
bool keepalive_enabled;
std::string site_password;
std::string report_password;
std::mutex del_reasons_lock;
void load_config(config * conf);
void do_start_reaper();
void reap_peers();
void reap_del_reasons();
std::string get_del_reason(int code);
peer_list::iterator add_peer(peer_list &peer_list, const std::string &peer_id);
inline bool peer_is_visible(user_ptr &u, peer *p);
std::mutex del_reasons_lock;
void load_config(config * conf);
void do_start_reaper();
void reap_peers();
void reap_del_reasons();
std::string get_del_reason(int code);
peer_list::iterator add_peer(peer_list &peer_list, const std::string &peer_id);
inline bool peer_is_visible(user_ptr &u, peer *p);
public:
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 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);
public:
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 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);
void reload_lists();
bool shutdown();
void reload_lists();
bool shutdown();
tracker_status get_status() { return status; }
tracker_status get_status() { return status; }
void start_reaper();
void start_reaper();
};
#endif
#endif // SRC_WORKER_H_