<commit_before>#include "net_stats.h"
#include <pthread.h>
#include "pthread_util.h"
#include <netinet/in.h>
#include <linux/tcp.h>
#include <sys/types.h>
#include "timeops.h"
#include <cmath>
#include <map>
using std::pair; using std::make_pair;

class InvalidEstimateException {};

NetStats::StatsCache NetStats::stats_cache;
pthread_rwlock_t NetStats::stats_cache_lock;
NetStats::static_initializer NetStats::init;

NetStats::IROBIfaceMap NetStats::irob_iface_map;
IntSet NetStats::striped_irobs;
pthread_mutex_t NetStats::irob_iface_map_lock = PTHREAD_MUTEX_INITIALIZER;


NetStats::static_initializer::static_initializer()
{
    pthread_rwlock_init(&NetStats::stats_cache_lock, NULL);
}


static struct timeval time_to_send(size_t msg_size, u_long bw_estimate)
{
    double bw_seconds = ((double)msg_size) / bw_estimate;
    useconds_t bw_time_usecs = (useconds_t)(bw_seconds * 1000000);
    struct timeval bw_time = convert_to_timeval(bw_time_usecs);
    return bw_time;
}


NetStats::NetStats(struct net_interface local_iface, 
                   struct net_interface remote_iface)
    : local_addr(local_iface.ip_addr), remote_addr(remote_iface.ip_addr)
{
    pthread_rwlock_init(&my_lock, NULL);
    last_RTT.tv_sec = last_srv_time.tv_sec = -1;
    last_RTT.tv_usec = last_srv_time.tv_usec = 0;
    last_req_size = 0;
    last_irob = -1;

    cache_restore();

    u_long init_bandwidth = iface_bandwidth(local_iface, remote_iface);
    u_long init_latency = iface_RTT(local_iface, remote_iface) / 2;
    if (init_bandwidth > 0) {
        dbgprintf("Adding initial bandwidth observation: %lu bytes/sec\n",
                  init_bandwidth);
        net_estimates.estimates[NET_STATS_BW_UP].add_observation(init_bandwidth);
    }
    if (init_latency > 0) {
        dbgprintf("Adding initial latency observation: %lu ms\n",
                  init_latency);
        net_estimates.estimates[NET_STATS_LATENCY].add_observation(init_latency);
    }
    cache_save();
}

void
NetStats::update(struct net_interface local_iface,
                 struct net_interface remote_iface)
{
    {
        PthreadScopedRWLock lock(&my_lock, true);

        u_long spot_bandwidth = iface_bandwidth(local_iface, remote_iface);
        u_long spot_latency = iface_RTT(local_iface, remote_iface) / 2;
        if (spot_bandwidth > 0) {
            dbgprintf("Adding bandwidth observation from scout: %lu bytes/sec\n",
                      spot_bandwidth);
            //net_estimates.estimates[NET_STATS_BW_UP].add_observation(spot_bandwidth);
            net_estimates.estimates[NET_STATS_BW_UP].reset(spot_bandwidth);
        }
        if (spot_latency > 0) {
            dbgprintf("Adding latency observation from scout: %lu ms\n",
                      spot_latency);
            //net_estimates.estimates[NET_STATS_LATENCY].add_observation(spot_latency);
            net_estimates.estimates[NET_STATS_LATENCY].reset(spot_latency);
        }
    }
    cache_save();
}


NetStats::~NetStats()
{
    cache_save();
}

void
NetStats::cache_save()
{
    PthreadScopedRWLock wrlock(&stats_cache_lock, true);
    PthreadScopedRWLock rd_self_lock(&my_lock, false);

    struct net_interface local_iface, remote_iface;
    local_iface.ip_addr = local_addr;
    remote_iface.ip_addr = remote_addr;
    StatsCache::key_type key = make_pair(local_iface, remote_iface);
    for (size_t i = 0; i < NUM_ESTIMATES; ++i) {
        stats_cache[key].estimates[i] = net_estimates.estimates[i];
    }
}

void<commit_msg>Replacing network estimates with the oracle-scout is kinda bogus, so instead we'll only do that if the CSocket is new (by removing the cache_restore() call until I can do it based on BSSID.)<commit_after>#include "net_stats.h"
#include <pthread.h>
#include "pthread_util.h"
#include <netinet/in.h>
#include <linux/tcp.h>
#include <sys/types.h>
#include "timeops.h"
#include <cmath>
#include <map>
using std::pair; using std::make_pair;

class InvalidEstimateException {};

NetStats::StatsCache NetStats::stats_cache;
pthread_rwlock_t NetStats::stats_cache_lock;
NetStats::static_initializer NetStats::init;

NetStats::IROBIfaceMap NetStats::irob_iface_map;
IntSet NetStats::striped_irobs;
pthread_mutex_t NetStats::irob_iface_map_lock = PTHREAD_MUTEX_INITIALIZER;


NetStats::static_initializer::static_initializer()
{
    pthread_rwlock_init(&NetStats::stats_cache_lock, NULL);
}


static struct timeval time_to_send(size_t msg_size, u_long bw_estimate)
{
    double bw_seconds = ((double)msg_size) / bw_estimate;
    useconds_t bw_time_usecs = (useconds_t)(bw_seconds * 1000000);
    struct timeval bw_time = convert_to_timeval(bw_time_usecs);
    return bw_time;
}


NetStats::NetStats(struct net_interface local_iface, 
                   struct net_interface remote_iface)
    : local_addr(local_iface.ip_addr), remote_addr(remote_iface.ip_addr)
{
    pthread_rwlock_init(&my_lock, NULL);
    last_RTT.tv_sec = last_srv_time.tv_sec = -1;
    last_RTT.tv_usec = last_srv_time.tv_usec = 0;
    last_req_size = 0;
    last_irob = -1;

    // don't do this until cache-by-BSSID is implemented.
    //  the last WiFi estimates don't predict the next ones.
    // anyway, the 3G stats stay as long as that CSocket does,
    //  which should be the entire trace.
    //cache_restore();

    u_long init_bandwidth = iface_bandwidth(local_iface, remote_iface);
    u_long init_latency = iface_RTT(local_iface, remote_iface) / 2;
    if (init_bandwidth > 0) {
        dbgprintf("Adding initial bandwidth observation: %lu bytes/sec\n",
                  init_bandwidth);
        net_estimates.estimates[NET_STATS_BW_UP].add_observation(init_bandwidth);
    }
    if (init_latency > 0) {
        dbgprintf("Adding initial latency observation: %lu ms\n",
                  init_latency);
        net_estimates.estimates[NET_STATS_LATENCY].add_observation(init_latency);
    }
    cache_save();
}

void
NetStats::update(struct net_interface local_iface,
                 struct net_interface remote_iface)
{
    {
        PthreadScopedRWLock lock(&my_lock, true);

        u_long spot_bandwidth = iface_bandwidth(local_iface, remote_iface);
        u_long spot_latency = iface_RTT(local_iface, remote_iface) / 2;
        if (spot_bandwidth > 0) {
            dbgprintf("Adding bandwidth observation from scout: %lu bytes/sec\n",
                      spot_bandwidth);
            net_estimates.estimates[NET_STATS_BW_UP].add_observation(spot_bandwidth);
            //net_estimates.estimates[NET_STATS_BW_UP].reset(spot_bandwidth);
        }
        if (spot_latency > 0) {
            dbgprintf("Adding latency observation from scout: %lu ms\n",
                      spot_latency);
            net_estimates.estimates[NET_STATS_LATENCY].add_observation(spot_latency);
            //net_estimates.estimates[NET_STATS_LATENCY].reset(spot_latency);
        }
    }
    cache_save();
}


NetStats::~NetStats()
{
    cache_save();
}

void
NetStats::cache_save()
{
    PthreadScopedRWLock wrlock(&stats_cache_lock, true);
    PthreadScopedRWLock rd_self_lock(&my_lock, false);

    struct net_interface local_iface, remote_iface;
    local_iface.ip_addr = local_addr;
    remote_iface.ip_addr = remote_addr;
    StatsCache::key_type key = make_pair(local_iface, remote_iface);
    for (size_t i = 0; i < NUM_ESTIMATES; ++i) {
        stats_cache[key].estimates[i] = net_estimates.estimates[i];
    }
}

void
{'fraction_non_alphanumeric': 0.1008801624915369, 'fraction_numerical': 0.005010155721056195, 'mean_word_length': 3.7867790019442644, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 23, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, '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': '2169021', 'n_tokens_mistral': 2668, 'n_tokens_neox': 2578, 'n_words': 448}