<commit_before>#include <apr_hash.h>
#include <apr_queue.h>
#include <apr_thread_cond.h>

#include "c4-internal.h"
#include "net/network.h"
#include "operator/operator.h"
#include "parser/parser.h"
#include "planner/installer.h"
#include "planner/planner.h"
#include "router.h"
#include "storage/sqlite.h"
#include "storage/table.h"
#include "types/catalog.h"
#include "util/dump_table.h"
#include "util/list.h"
#include "util/strbuf.h"
#include "util/tuple_buf.h"

struct C4Router
{
    C4Runtime *c4;
    apr_pool_t *pool;

    /* Map from table name => OpChainList */
    apr_hash_t *op_chain_tbl;

    /* Queue of to-be-performed external actions inserted by other threads */
    apr_queue_t *queue;

    /* Derived tuples computed within the current fixpoint; to-be-routed */
    TupleBuf *route_buf;
    /* Pending network output tuples computed within current fixpoint */
    TupleBuf *net_buf;

    int ntuple_routed;
};

typedef enum WorkItemKind
{
    WI_TUPLE,
    WI_PROGRAM,
    WI_DUMP_TABLE,
    WI_SHUTDOWN
} WorkItemKind;

typedef struct WorkItem
{
    WorkItemKind kind;

    /* WI_TUPLE: */
    Tuple *tuple;
    TableDef *tbl_def;

    /* WI_PROGRAM: */
    char *program_src;

    /* WI_DUMP_TABLE: */
    char *tbl_name;
    apr_thread_cond_t *cond;
    apr_thread_mutex_t *lock;
    StrBuf *buf;
    bool is_done;
} WorkItem;

static void router_enqueue(C4Router *router, WorkItem *wi);
static bool drain_queue(C4Router *router);

C4Router *
router_make(C4Runtime *c4)
{
    C4Router *router;
    apr_status_t s;

    router = apr_pcalloc(c4->pool, sizeof(*router));
    router->c4 = c4;
    router->pool = c4->pool;
    router->op_chain_tbl = apr_hash_make(router->pool);
    router->route_buf = tuple_buf_make(4096, router->pool);
    router->net_buf = tuple_buf_make(512, router->pool);
    router->ntuple_routed = 0;
    s = apr_queue_create(&router->queue, 512, router->pool);
    if (s != APR_SUCCESS)
        FAIL_APR(s);

    return router;
}

/*
 * Route a new tuple that belongs to the given table. If "check_remote" is true
 * and the tuple is remote (i.e. the tuple's location specifier denotes an
 * address other than the local node address), then we enqueue the tuple in the
 * network buffer.  Otherwise, we insert the tuple into the specified table, and
 * then enqueue it to be routed in the next fixpoint.
 *
 * XXX: "check_remote" is a hack. The problem is that a node might have many
 * different addresses (due to multi-homing, IPv4 vs. IPv6, DNS aliases,
 * etc.). If we receive a network tuple that doesn't exactly match our local
 * node address, we can easily get into an infinite loop.
 */
void
router_install_tuple(C4Router *router, Tuple *tuple, TableDef *tbl_def,
                     bool check_remote)
{
#if 1
    c4_log(router->c4, "%s: %s (=> %s)",
           __func__, log_tuple(router->c4, tuple), tbl_def->name);
#endif
    if (check_remote && tuple_is_remote(tuple, tbl_def, router->c4))
    {
        router_enqueue_net(router, tuple, tbl_def);
        return;
    }

    /* If the tuple is a duplicate, no need to route it */
    if (tbl_def->table->insert(tbl_def->table, tuple) == false)
        return;

    router_enqueue_internal(router, tuple, tbl_def);
}

void
router_do_fixpoint(C4Router *router)
{
    TupleBuf *route_buf = router->route_buf;
    TupleBuf *net_buf = router->net_buf;
    bool benchmark_done = false;

    ASSERT(tuple_buf_is_empty(net_buf));

    /*
     * NB: We route tuples from route_buf in a FIFO manner, but that
     * is not necessarily the only choice.
     */
    while (!tuple_buf_is_empty(route_buf))
    {
        Tuple *tuple;
        TableDef *tbl_def;
        OpChain *op_chain;

        tuple_buf_shift(route_buf, &tuple, &tbl_def);

        op_chain = tbl_def->op_chain_list->head;
        while (op_chain != NULL)<commit_msg>Tweak comment.<commit_after>#include <apr_hash.h>
#include <apr_queue.h>
#include <apr_thread_cond.h>

#include "c4-internal.h"
#include "net/network.h"
#include "operator/operator.h"
#include "parser/parser.h"
#include "planner/installer.h"
#include "planner/planner.h"
#include "router.h"
#include "storage/sqlite.h"
#include "storage/table.h"
#include "types/catalog.h"
#include "util/dump_table.h"
#include "util/list.h"
#include "util/strbuf.h"
#include "util/tuple_buf.h"

struct C4Router
{
    C4Runtime *c4;
    apr_pool_t *pool;

    /* Map from table name => OpChainList */
    apr_hash_t *op_chain_tbl;

    /* Queue of to-be-performed external actions inserted by other threads */
    apr_queue_t *queue;

    /* Derived tuples computed within the current fixpoint; to-be-routed */
    TupleBuf *route_buf;
    /* Pending network output tuples computed within current fixpoint */
    TupleBuf *net_buf;

    int ntuple_routed;
};

typedef enum WorkItemKind
{
    WI_TUPLE,
    WI_PROGRAM,
    WI_DUMP_TABLE,
    WI_SHUTDOWN
} WorkItemKind;

typedef struct WorkItem
{
    WorkItemKind kind;

    /* WI_TUPLE: */
    Tuple *tuple;
    TableDef *tbl_def;

    /* WI_PROGRAM: */
    char *program_src;

    /* WI_DUMP_TABLE: */
    char *tbl_name;
    apr_thread_cond_t *cond;
    apr_thread_mutex_t *lock;
    StrBuf *buf;
    bool is_done;
} WorkItem;

static void router_enqueue(C4Router *router, WorkItem *wi);
static bool drain_queue(C4Router *router);

C4Router *
router_make(C4Runtime *c4)
{
    C4Router *router;
    apr_status_t s;

    router = apr_pcalloc(c4->pool, sizeof(*router));
    router->c4 = c4;
    router->pool = c4->pool;
    router->op_chain_tbl = apr_hash_make(router->pool);
    router->route_buf = tuple_buf_make(4096, router->pool);
    router->net_buf = tuple_buf_make(512, router->pool);
    router->ntuple_routed = 0;
    s = apr_queue_create(&router->queue, 512, router->pool);
    if (s != APR_SUCCESS)
        FAIL_APR(s);

    return router;
}

/*
 * Route a new tuple that belongs to the given table. If "check_remote" is true
 * and the tuple is remote (i.e. the tuple's location specifier denotes an
 * address other than the local node address), then we enqueue the tuple in the
 * network buffer.  Otherwise, we insert the tuple into the specified table, and
 * then enqueue it to be routed in the next fixpoint.
 *
 * XXX: "check_remote" is a hack. The problem is that a node might have many
 * different addresses (due to multi-homing, IPv4 vs. IPv6, DNS aliases,
 * etc.). If we receive a network tuple that doesn't exactly match our local
 * node address, we can easily get into an infinite loop.
 */
void
router_install_tuple(C4Router *router, Tuple *tuple, TableDef *tbl_def,
                     bool check_remote)
{
#if 1
    c4_log(router->c4, "%s: %s (=> %s)",
           __func__, log_tuple(router->c4, tuple), tbl_def->name);
#endif
    if (check_remote && tuple_is_remote(tuple, tbl_def, router->c4))
    {
        router_enqueue_net(router, tuple, tbl_def);
        return;
    }

    /* If the tuple is a duplicate, no need to route it */
    if (tbl_def->table->insert(tbl_def->table, tuple) == false)
        return;

    router_enqueue_internal(router, tuple, tbl_def);
}

void
router_do_fixpoint(C4Router *router)
{
    TupleBuf *route_buf = router->route_buf;
    TupleBuf *net_buf = router->net_buf;
    bool benchmark_done = false;

    ASSERT(tuple_buf_is_empty(net_buf));

    /*
     * We route tuples from route_buf in a FIFO manner, but that is not
     * necessarily the only choice.
     */
    while (!tuple_buf_is_empty(route_buf))
    {
        Tuple *tuple;
        TableDef *tbl_def;
        OpChain *op_chain;

        tuple_buf_shift(route_buf, &tuple, &tbl_def);

        op_chain = tbl_def->op_chain_list->head;
        while (op_chain != NULL)
{'fraction_non_alphanumeric': 0.11301057027273914, 'fraction_numerical': 0.008873809213101918, 'mean_word_length': 3.30561797752809, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 61, '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': '1157651', 'n_tokens_mistral': 2748, 'n_tokens_neox': 2647, 'n_words': 721}