<commit_before>#include "twister.h"

#include "twister_utils.h"

#include "main.h"
#include "init.h"
#include "bitcoinrpc.h"

using namespace json_spirit;
using namespace std;

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>

#include <time.h>

twister::twister()
{
}

// ===================== LIBTORRENT & DHT ===========================

#include "libtorrent/config.hpp"
#include "libtorrent/entry.hpp"
#include "libtorrent/bencode.hpp"
#include "libtorrent/session.hpp"
#include "libtorrent/alert_types.hpp"

#define TORRENT_DISABLE_GEO_IP
#include "libtorrent/aux_/session_impl.hpp"

#define DEBUG_ACCEPT_POST 1

using namespace libtorrent;
static session *ses = NULL;
static int num_outstanding_resume_data;

static CCriticalSection cs_dhtgetMap;
static map<sha1_hash, alert_manager*> m_dhtgetMap;

static CCriticalSection cs_twister;
static map<std::string, bool> m_specialResources;
static map<std::string, torrent_handle> m_userTorrent;

static std::string m_preferredSpamLang = "[en]";
static std::string m_receivedSpamMsgStr;
static std::string m_receivedSpamUserStr;
static std::map<std::string,UserData> m_users;

sha1_hash dhtTargetHash(std::string const &username, std::string const &resource, std::string const &type)
{
    entry target;
    target["n"] = username;
    target["r"] = resource;
    target["t"] = type;

    std::vector<char> buf;
    bencode(std::back_inserter(buf), target);
    return hasher(buf.data(), buf.size()).final();
}

torrent_handle startTorrentUser(std::string const &username)
{
    LOCK(cs_twister);
    if( !m_userTorrent.count(username) ) {
        sha1_hash ih = dhtTargetHash(username, "tracker", "m");

        printf("adding torrent for [%s,tracker]\n", username.c_str());
        add_torrent_params tparams;
        tparams.info_hash = ih;
        tparams.name = username;
        boost::filesystem::path torrentPath = GetDataDir() / "swarm";
        tparams.save_path= torrentPath.string();

        error_code ec;
        create_directory(tparams.save_path, ec);

        std::string filename = combine_path(tparams.save_path, to_hex(ih.to_string()) + ".resume");
        load_file(filename.c_str(), tparams.resume_data);

        m_userTorrent[username] = ses->add_torrent(tparams);
        m_userTorrent[username].force_dht_announce();
        torrent_status status = m_userTorrent[username].status();
    }
    return m_userTorrent[username];
}

int lastPostKfromTorrent(std::string const &username)
{
    if( !m_userTorrent.count(username) )
        return -1;

    torrent_status status = m_userTorrent[username].status();
    return status.last_have;
}

void ThreadWaitExtIP()
{
    RenameThread("wait-extip");

    std::string ipStr;

    // wait up to 5 seconds for bitcoin to get the external IP
    for( int i = 0; i < 10; i++ ) {
        const CNetAddr paddrPeer("8.8.8.8");
        CAddress addr( GetLocalAddress(&paddrPeer) );
        if( addr.IsValid() ) {
            ipStr = addr.ToStringIP();
            break;
        }
        MilliSleep(500);
    }

    error_code ec;
    int listen_port = GetListenPort() + LIBTORRENT_PORT_OFFSET;
    std::string bind_to_interface = "";

    printf("Creating new libtorrent session ext_ip=%s port=%d\n", ipStr.c_str(), listen_port);

    ses = new session(fingerprint("TW", LIBTORRENT_VERSION_MAJOR, LIBTORRENT_VERSION_MINOR, 0, 0)
            , session::add_default_plugins
            , alert::dht_notification
            , ipStr.size() ? ipStr.c_str() : NULL
            , std::make_pair(listen_port, listen_port));

    std::vector<char> in;
    boost::filesystem::path sesStatePath = GetDataDir() / "ses_state";
    if (load_file(sesStatePath.string(), in) == 0)
    {
            lazy_entry e;
            if (lazy_bdecode(&in[0], &in[0] + in.size(), e, ec) == 0)
                    ses->load_state(e);
    }

    ses->start_upnp();
    ses->start_natpmp();

    ses->listen_on(std::make_pair(listen_port, listen_port)
            , ec, bind_to_interface.c_str());
    if (ec)
    {
            fprintf(stderr, "failed to listen%s%s on ports %d-%d: %s\n"
                    , bind_to_interface.empty() ? "" : " on ", bind_to_interface.c_str()
                    , listen_port, listen_port+1, ec.message().c_str());
    }

    dht_settings dhts;
    // settings to test local connections
    dhts.restrict_routing_ips = false;
    dhts.restrict_search_ips = false;
    ses->set_dht_settings(dhts);
    ses->start_dht();

    session_settings settings;
    // settings to test local connections
    settings.allow_multiple_connections_per_ip = true;
    //settings.enable_outgoing_utp = false; // (false to see connections in netstat)
    //settings.dht_announce_interval = 60; // test
    //settings.min_announce_interval = 60; // test
    settings.anonymous_mode = false; // (false => send peer_id, avoid connecting to itself)
    ses->set_settings(settings);


    boost::filesystem::path userDataPath = GetDataDir() / "user_data";
    loadUserData(userDataPath.string(), m_users);
    printf("loaded user_data for %zd users\n", m_users.size());

    // now restart the user torrents (all m_following)
    std::map<std::string,UserData>::const_iterator i;
    for (i = m_users.begin(); i != m_users.end(); ++i) {
        UserData const &data = i->second;
        BOOST_FOREACH(string username, data.m_following) {
            startTorrentUser(username);
        }
    }

    printf("libtorrent + dht started\n");
}

void ThreadMaintainDHTNodes()
{
    RenameThread("maintain-dht-nodes");

    while(1) {
        MilliSleep(15000);

        if( ses ) {
            vector<CAddress> vAddr = addrman.GetAddr();
            session_status ss = ses->status();
            if( ss.dht_nodes < (int)(vNodes.size() + vAddr.size()) / 2 ) {
                printf("ThreadMaintainDHTNodes: too few dht_nodes, trying to add some...\n");
                LOCK(cs_vNodes);
                BOOST_FOREACH(CNode* pnode, vNodes) {

                    // if !fInbound we created this connection so ip is reachable
                    if( !pnode->fInbound ) {
                        std::string addr = pnode->addr.ToStringIP();
                        int port = pnode->addr.GetPort() + LIBTORRENT_PORT_OFFSET;

                        printf("Adding dht node (outbound) %s:%d\n", addr.c_str(), port);
                        ses->add_dht_node(std::pair<std::string, int>(addr, port));
                    }
                }
                BOOST_FOREACH(const CAddress &a, vAddr) {<commit_msg>trying (once more) to improve ThreadMaintainDHTNodes (dht nodes from registration network nodes)<commit_after>#include "twister.h"

#include "twister_utils.h"

#include "main.h"
#include "init.h"
#include "bitcoinrpc.h"

using namespace json_spirit;
using namespace std;

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>
#include <boost/lexical_cast.hpp>

#include <time.h>

twister::twister()
{
}

// ===================== LIBTORRENT & DHT ===========================

#include "libtorrent/config.hpp"
#include "libtorrent/entry.hpp"
#include "libtorrent/bencode.hpp"
#include "libtorrent/session.hpp"
#include "libtorrent/alert_types.hpp"

#define TORRENT_DISABLE_GEO_IP
#include "libtorrent/aux_/session_impl.hpp"

#define DEBUG_ACCEPT_POST 1

using namespace libtorrent;
static session *ses = NULL;
static int num_outstanding_resume_data;

static CCriticalSection cs_dhtgetMap;
static map<sha1_hash, alert_manager*> m_dhtgetMap;

static CCriticalSection cs_twister;
static map<std::string, bool> m_specialResources;
static map<std::string, torrent_handle> m_userTorrent;

static std::string m_preferredSpamLang = "[en]";
static std::string m_receivedSpamMsgStr;
static std::string m_receivedSpamUserStr;
static std::map<std::string,UserData> m_users;

sha1_hash dhtTargetHash(std::string const &username, std::string const &resource, std::string const &type)
{
    entry target;
    target["n"] = username;
    target["r"] = resource;
    target["t"] = type;

    std::vector<char> buf;
    bencode(std::back_inserter(buf), target);
    return hasher(buf.data(), buf.size()).final();
}

torrent_handle startTorrentUser(std::string const &username)
{
    LOCK(cs_twister);
    if( !m_userTorrent.count(username) ) {
        sha1_hash ih = dhtTargetHash(username, "tracker", "m");

        printf("adding torrent for [%s,tracker]\n", username.c_str());
        add_torrent_params tparams;
        tparams.info_hash = ih;
        tparams.name = username;
        boost::filesystem::path torrentPath = GetDataDir() / "swarm";
        tparams.save_path= torrentPath.string();

        error_code ec;
        create_directory(tparams.save_path, ec);

        std::string filename = combine_path(tparams.save_path, to_hex(ih.to_string()) + ".resume");
        load_file(filename.c_str(), tparams.resume_data);

        m_userTorrent[username] = ses->add_torrent(tparams);
        m_userTorrent[username].force_dht_announce();
        torrent_status status = m_userTorrent[username].status();
    }
    return m_userTorrent[username];
}

int lastPostKfromTorrent(std::string const &username)
{
    if( !m_userTorrent.count(username) )
        return -1;

    torrent_status status = m_userTorrent[username].status();
    return status.last_have;
}

void ThreadWaitExtIP()
{
    RenameThread("wait-extip");

    std::string ipStr;

    // wait up to 5 seconds for bitcoin to get the external IP
    for( int i = 0; i < 10; i++ ) {
        const CNetAddr paddrPeer("8.8.8.8");
        CAddress addr( GetLocalAddress(&paddrPeer) );
        if( addr.IsValid() ) {
            ipStr = addr.ToStringIP();
            break;
        }
        MilliSleep(500);
    }

    error_code ec;
    int listen_port = GetListenPort() + LIBTORRENT_PORT_OFFSET;
    std::string bind_to_interface = "";

    printf("Creating new libtorrent session ext_ip=%s port=%d\n", ipStr.c_str(), listen_port);

    ses = new session(fingerprint("TW", LIBTORRENT_VERSION_MAJOR, LIBTORRENT_VERSION_MINOR, 0, 0)
            , session::add_default_plugins
            , alert::dht_notification
            , ipStr.size() ? ipStr.c_str() : NULL
            , std::make_pair(listen_port, listen_port));

    std::vector<char> in;
    boost::filesystem::path sesStatePath = GetDataDir() / "ses_state";
    if (load_file(sesStatePath.string(), in) == 0)
    {
            lazy_entry e;
            if (lazy_bdecode(&in[0], &in[0] + in.size(), e, ec) == 0)
                    ses->load_state(e);
    }

    ses->start_upnp();
    ses->start_natpmp();

    ses->listen_on(std::make_pair(listen_port, listen_port)
            , ec, bind_to_interface.c_str());
    if (ec)
    {
            fprintf(stderr, "failed to listen%s%s on ports %d-%d: %s\n"
                    , bind_to_interface.empty() ? "" : " on ", bind_to_interface.c_str()
                    , listen_port, listen_port+1, ec.message().c_str());
    }

    dht_settings dhts;
    // settings to test local connections
    dhts.restrict_routing_ips = false;
    dhts.restrict_search_ips = false;
    ses->set_dht_settings(dhts);
    ses->start_dht();

    session_settings settings;
    // settings to test local connections
    settings.allow_multiple_connections_per_ip = true;
    //settings.enable_outgoing_utp = false; // (false to see connections in netstat)
    //settings.dht_announce_interval = 60; // test
    //settings.min_announce_interval = 60; // test
    settings.anonymous_mode = false; // (false => send peer_id, avoid connecting to itself)
    ses->set_settings(settings);


    boost::filesystem::path userDataPath = GetDataDir() / "user_data";
    loadUserData(userDataPath.string(), m_users);
    printf("loaded user_data for %zd users\n", m_users.size());

    // now restart the user torrents (all m_following)
    std::map<std::string,UserData>::const_iterator i;
    for (i = m_users.begin(); i != m_users.end(); ++i) {
        UserData const &data = i->second;
        BOOST_FOREACH(string username, data.m_following) {
            startTorrentUser(username);
        }
    }

    printf("libtorrent + dht started\n");
}

void ThreadMaintainDHTNodes()
{
    RenameThread("maintain-dht-nodes");

    while(1) {
        MilliSleep(15000);

        if( ses ) {
            LOCK(cs_vNodes);
            vector<CAddress> vAddr = addrman.GetAddr();
            session_status ss = ses->status();
            int totalNodesCandidates = (int)(vNodes.size() + vAddr.size());
            if( (!ss.dht_nodes && totalNodesCandidates) ||
                ss.dht_nodes < totalNodesCandidates / 2 ) {
                printf("ThreadMaintainDHTNodes: too few dht_nodes, trying to add some...\n");
                BOOST_FOREACH(CNode* pnode, vNodes) {

                    // if !fInbound we created this connection so ip is reachable
                    if( !pnode->fInbound ) {
                        std::string addr = pnode->addr.ToStringIP();
                        int port = pnode->addr.GetPort() + LIBTORRENT_PORT_OFFSET;

                        printf("Adding dht node (outbound) %s:%d\n", addr.c_str(), port);
                        ses->add_dht_node(std::pair<std::string, int>(addr, port));
                    } else if( !ss.dht_nodes /* last resort! */ ) {
                        // can't use port number of inbound connectio, try standard port
                        std::string addr = pnode->addr.ToStringIP();
                        int port = Params().GetDefaultPort() + LIBTORRENT_PORT_OFFSET;

                        printf("Adding dht node (inbound) %s:%d\n", addr.c_str(), port);
                        ses->add_dht_node(std::pair<std::string, int>(addr, port));
                    }
                }
                BOOST_FOREACH(const CAddress &a, vAddr) {
{'fraction_non_alphanumeric': 0.13563868348557342, 'fraction_numerical': 0.004929679570827896, 'mean_word_length': 2.9549885321100917, 'pattern_counts': {'":': 0, '<': 34, '<?xml version=': 0, '>': 62, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4391294', 'n_tokens_mistral': 4597, 'n_tokens_neox': 4405, 'n_words': 900}