<commit_before>#include <stdio.h>
#include <string.h>

#include "channel.h"
#include "protocol.h"


int msg_channel_id(void *p) {
    struct msg_channel *c = (struct msg_channel *) p;
    return c->epid;
}


struct ep_tree *msg_server_self(struct msg_server *s) {
    return &s->hosts[0].shared_tree;
}


int msg_server_init_host(struct msg_server *s) {
    struct ep_host host;
    ep_host_init_self(&host);


    // find own address and hostname
    size_t host_max = 32;
    s->hosts = malloc(sizeof(struct msg_host) * host_max);
    s->host_count = 0;
    msg_server_add_host(s, host.addr, host.hostname);
    for (int i = 0; i < host_max; ++i) {
        msg_tree_init(&s->hosts[i].shared_tree);
    }
}


int msg_server_add_host(struct msg_server *s, const char *addr, const char *name) {
    int hostid = s->host_count++;
    struct msg_host *h = &s->hosts[hostid];
    h->active_id = 0;
    strcpy(h->addr, addr);
    strcpy(h->hostname, name);
    return hostid;
}


void msg_server_printsub(struct msg_server *s) {
    printf("%d sub keys\n", s->node_to_sub.keys.elem_count);
    printf("%d sub values\n", s->node_to_sub.value_count);

    // find all the keys
    for (int i = 0; i < s->node_to_sub.keys.elem_count; ++i) {
        struct ep_subarray *sa = ep_multimap_get_index(&s->node_to_sub, i);
        printf("key %d (%d to %d)\n", sa->key, sa->begin, sa->end);

        // print values
        for (int i = sa->begin; i < sa->end; ++i) {
            size_t addr = sizeof(struct msg_subscriber) * i;
            struct msg_subscriber *sub = (struct msg_subscriber *) &s->node_to_sub.values[addr];
            printf("\tsub %d => %d\n", sub->subid, sub->epid);
        }
    }
}


int msg_node_of_host(struct msg_server *s, int epid) {
    int *nodelist = ep_multimap_get(&s->host_to_tree, epid, 0);
    if (nodelist) {
        return nodelist[0];
    }
    else {
        return 0;
    }
}


void msg_server_add_client(struct msg_server *s, int epid, int nodeid) {
    printf("connect epid %d => node %d\n", epid, nodeid);
    int index = ep_multimap_insert(&s->host_to_tree, epid, 1);
    int *nodelist = ep_multimap_get(&s->host_to_tree, epid, index);
    nodelist[0] = nodeid;
}


void msg_server_rm_client(struct msg_server *s, int i) {

}


void msg_server_init_channel(struct msg_server *s, int epid) {
    struct msg_channel ch;
    ch.epid = epid;
    ch.type = 0;
    ch.subs = 0;
    ep_map_insert(&s->socket_type, &ch);
}


void msg_server_subscribe(struct msg_server *s, int sendnode, int epid, int subid) {
    printf("subscribe updates on node id %d to epid %d\n", sendnode, epid);
    int index = ep_multimap_insert(&s->node_to_sub, sendnode, 1);
    struct msg_subscriber *sub = ep_multimap_get(&s->node_to_sub, sendnode, index);
    if (sub) {
        sub->epid = epid;
        sub->subid = subid;
    }
    else {
        printf("id %d not found\n", sendnode);
    }
}


void msg_server_read_data(struct msg_server *serv, struct ep_buffer *buf) {
    struct msg_node_update ud;
    ep_buffer_peek(buf, (char *) &ud, 0, sizeof(int) * 2);

    //
    ud.len = 0;

    // pass to all subscribed processes
    // and all peers with at least one subscriber
    struct ep_subarray *sa = ep_multimap_get_key(&serv->node_to_sub, ud.nodeid);
    for (int i = sa->begin; i < sa->end; ++i) {
        struct msg_subscriber *sub = ep_multimap_get_value(&serv->node_to_sub, i);
        msg_server_reply_data(serv, sub->epid, &ud);

    }
}


void msg_server_reply_data(struct msg_server *serv, int epid, struct msg_node_update *u) {
    struct ep_table_entry *e = ep_map_get(&serv->tb.entries, epid);
    struct ep_channel *ch = &e->data.ch;
    msg_send_block(&ch->write_buf, u->nodeid, u->hdlid, u->buf, u->len);
    printf("write length: %d\n", ch->write_buf.size);
    ep_channel_flush(ch);
}


void msg_server_init(struct msg_server *s, const char *sockpath) {

    // alloc structures
    ep_map_alloc(&s->socket_type, msg_channel_id, sizeof(struct msg_channel), 1024);
    ep_multimap_init(&s->host_to_tree, sizeof(int), 1024);
    ep_multimap_init(&s->node_to_sub, sizeof(struct msg_subscriber), 1024);
    ep_table_init(&s->tb, 256);
    msg_server_init_host(s);

    // start threads
    ep_thread_pool_create(&s->pool, &s->tb, 4, 0);

    // create an acceptor
    struct ep_acceptor acc;
    ep_listen_remote(&acc.addr, 2204);
    ep_init_acceptor(&acc, msg_server_accept, s);
    ep_add_acceptor(&s->tb, &acc);
}


int msg_server_connect(struct msg_server *serv, const char *addr) {
    struct ep_channel ch;
    ep_connect_remote(&ch.addr, addr, 2204);
    int err = ep_init_channel(&ch);
    if (err == -1) {<commit_msg>debug msg<commit_after>#include <stdio.h>
#include <string.h>

#include "channel.h"
#include "protocol.h"


int msg_channel_id(void *p) {
    struct msg_channel *c = (struct msg_channel *) p;
    return c->epid;
}


struct ep_tree *msg_server_self(struct msg_server *s) {
    return &s->hosts[0].shared_tree;
}


int msg_server_init_host(struct msg_server *s) {
    struct ep_host host;
    ep_host_init_self(&host);


    // find own address and hostname
    size_t host_max = 32;
    s->hosts = malloc(sizeof(struct msg_host) * host_max);
    s->host_count = 0;
    msg_server_add_host(s, host.addr, host.hostname);
    for (int i = 0; i < host_max; ++i) {
        msg_tree_init(&s->hosts[i].shared_tree);
    }
}


int msg_server_add_host(struct msg_server *s, const char *addr, const char *name) {
    int hostid = s->host_count++;
    struct msg_host *h = &s->hosts[hostid];
    h->active_id = 0;
    strcpy(h->addr, addr);
    strcpy(h->hostname, name);
    return hostid;
}


void msg_server_printsub(struct msg_server *s) {
    printf("%d sub keys\n", s->node_to_sub.keys.elem_count);
    printf("%d sub values\n", s->node_to_sub.value_count);

    // find all the keys
    for (int i = 0; i < s->node_to_sub.keys.elem_count; ++i) {
        struct ep_subarray *sa = ep_multimap_get_index(&s->node_to_sub, i);
        printf("key %d (%d to %d)\n", sa->key, sa->begin, sa->end);

        // print values
        for (int i = sa->begin; i < sa->end; ++i) {
            size_t addr = sizeof(struct msg_subscriber) * i;
            struct msg_subscriber *sub = (struct msg_subscriber *) &s->node_to_sub.values[addr];
            printf("\tsub %d => %d\n", sub->subid, sub->epid);
        }
    }
}


int msg_node_of_host(struct msg_server *s, int epid) {
    int *nodelist = ep_multimap_get(&s->host_to_tree, epid, 0);
    if (nodelist) {
        return nodelist[0];
    }
    else {
        return 0;
    }
}


void msg_server_add_client(struct msg_server *s, int epid, int nodeid) {
    printf("connect epid %d => node %d\n", epid, nodeid);
    int index = ep_multimap_insert(&s->host_to_tree, epid, 1);
    int *nodelist = ep_multimap_get(&s->host_to_tree, epid, index);
    nodelist[0] = nodeid;
}


void msg_server_rm_client(struct msg_server *s, int i) {

}


void msg_server_init_channel(struct msg_server *s, int epid) {
    struct msg_channel ch;
    ch.epid = epid;
    ch.type = 0;
    ch.subs = 0;
    ep_map_insert(&s->socket_type, &ch);
}


void msg_server_subscribe(struct msg_server *s, int sendnode, int epid, int subid) {
    printf("subscribe updates on node id %d to epid %d\n", sendnode, epid);
    int index = ep_multimap_insert(&s->node_to_sub, sendnode, 1);
    struct msg_subscriber *sub = ep_multimap_get(&s->node_to_sub, sendnode, index);
    if (sub) {
        sub->epid = epid;
        sub->subid = subid;
    }
    else {
        printf("id %d not found\n", sendnode);
    }
}


void msg_server_read_data(struct msg_server *serv, struct ep_buffer *buf) {
    struct msg_node_update ud;
    ep_buffer_peek(buf, (char *) &ud, 0, sizeof(int) * 2);

    //
    ud.len = 0;

    // pass to all subscribed processes
    // and all peers with at least one subscriber
    struct ep_subarray *sa = ep_multimap_get_key(&serv->node_to_sub, ud.nodeid);
    printf("recv data (node %d, handle %d, range %d to %d)", ud.nodeid, ud.hdlid, sa->begin, sa->end);
    for (int i = sa->begin; i < sa->end; ++i) {
        struct msg_subscriber *sub = ep_multimap_get_value(&serv->node_to_sub, i);
        msg_server_reply_data(serv, sub->epid, &ud);

    }
}


void msg_server_reply_data(struct msg_server *serv, int epid, struct msg_node_update *u) {
    struct ep_table_entry *e = ep_map_get(&serv->tb.entries, epid);
    struct ep_channel *ch = &e->data.ch;
    msg_send_block(&ch->write_buf, u->nodeid, u->hdlid, u->buf, u->len);
    printf("[%d] write %d bytes\n", epid, ch->write_buf.size);
    ep_channel_flush(ch);
}


void msg_server_init(struct msg_server *s, const char *sockpath) {

    // alloc structures
    ep_map_alloc(&s->socket_type, msg_channel_id, sizeof(struct msg_channel), 1024);
    ep_multimap_init(&s->host_to_tree, sizeof(int), 1024);
    ep_multimap_init(&s->node_to_sub, sizeof(struct msg_subscriber), 1024);
    ep_table_init(&s->tb, 256);
    msg_server_init_host(s);

    // start threads
    ep_thread_pool_create(&s->pool, &s->tb, 4, 0);

    // create an acceptor
    struct ep_acceptor acc;
    ep_listen_remote(&acc.addr, 2204);
    ep_init_acceptor(&acc, msg_server_accept, s);
    ep_add_acceptor(&s->tb, &acc);
}


int msg_server_connect(struct msg_server *serv, const char *addr) {
    struct ep_channel ch;
    ep_connect_remote(&ch.addr, addr, 2204);
    int err = ep_init_channel(&ch);
    if (err == -1) {
{'fraction_non_alphanumeric': 0.1429481733220051, 'fraction_numerical': 0.009345794392523364, 'mean_word_length': 3.3236914600550964, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 113, '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': '572312', 'n_tokens_mistral': 3643, 'n_tokens_neox': 3487, 'n_words': 813}