<commit_before>/*
 * q2pc_server.c
 *
 *  Created on: Apr 9, 2014
 *      Author: mgrosvenor
 */

 //#LINKFLAGS=-lpthread

#include <stdlib.h>
#include <pthread.h>
#include <signal.h>
#include <sys/time.h>

#include "q2pc_server.h"
#include "../transport/q2pc_transport.h"
#include "../errors/errors.h"
#include "../protocol/q2pc_protocol.h"


typedef struct{
    i64 lo;
    i64 hi;
    i64 count;
    i64 thread_id;
} thread_params_t;

void* run_thread( void* p);

//Server wide global
static CH_ARRAY(TRANS_CONN)* cons       = NULL;
static volatile bool stop_signal        = false;
static volatile bool pause_signal       = false;
static pthread_t* threads               = NULL;
static i64 real_thread_count            = 0;
static volatile i64* votes_scoreboard   = NULL;
static volatile i64* votes_count        = NULL;
static q2pc_trans* trans;


void cleanup()
{
    stop_signal = true;
    __sync_synchronize(); //Full fence

    if(threads){
        for(int i = 0; i < real_thread_count; i++){
            pthread_join(threads[i],NULL);
        }
    }

    if(trans){
        trans->delete(trans);
    }

}


//Signal handler to terminate early
void term(int signo)
{
    ch_log_info("Terminating...\n");
    (void)signo;
    cleanup();

    ch_log_info("Terminating... Done.\n");
    exit(0);
}


//Pause/unpause worker threads
void dopause_all(){ pause_signal = true;  __sync_synchronize(); }
void unpause_all(){ pause_signal = false; __sync_synchronize(); }

//Wait for all clients to connect
void do_connectall(i64 client_count)
{
    cons = CH_ARRAY_NEW(TRANS_CONN,client_count,NULL);
    if(!cons){ ch_log_fatal("Cannot allocate connections array\n"); }

    for(int i = 0; i < client_count; i++){
        trans->connect(trans, cons->off(cons,i));
    }

}



void server_init(const i64 thread_count, const i64 client_count , const transport_s* transport)
{

    //Signal handling for the main thread
    signal(SIGHUP,  term);
    signal(SIGKILL, term);
    signal(SIGTERM, term);
    signal(SIGINT, term);

    //Set up and init the voting scoreboard
    votes_scoreboard = aligned_alloc(sizeof(i64), sizeof(i64) * client_count);
    if(!votes_scoreboard){
        ch_log_fatal("Could not allocate memory for votes scoreboard\n");
    }
    bzero((void*)votes_scoreboard,sizeof(i64) * client_count);


    //Set up all the connections
    ch_log_info("Waiting for clients to connect...\n\r");
    trans = trans_factory(transport);
    do_connectall(client_count);
    ch_log_info("Waiting for clients to connect... Done.\n");


    //Calculate the connection to thread mappings
    i64 cons_per_thread = MAX(client_count / thread_count, 1);
    real_thread_count  = MIN(thread_count, client_count);
    i64 lo = 0;
    i64 hi = lo + cons_per_thread;

    votes_count = aligned_alloc(sizeof(i64), sizeof(i64) * real_thread_count);
    if(!votes_count){
        ch_log_fatal("Could not allocate memory for votes counter\n");
    }
    bzero((void*)votes_count,sizeof(i64) * real_thread_count);


    //Fire up the threads<commit_msg>Fixing build issues for clang3.3<commit_after>/*
 * q2pc_server.c
 *
 *  Created on: Apr 9, 2014
 *      Author: mgrosvenor
 */

 //#LINKFLAGS=-lpthread

#include <stdlib.h>
#include <pthread.h>
#include <signal.h>
#include <sys/time.h>

#include "q2pc_server.h"
#include "../transport/q2pc_transport.h"
#include "../errors/errors.h"
#include "../protocol/q2pc_protocol.h"


typedef struct{
    i64 lo;
    i64 hi;
    i64 count;
    i64 thread_id;
} thread_params_t;

void* run_thread( void* p);

//Server wide global
static CH_ARRAY(TRANS_CONN)* cons       = NULL;
static volatile bool stop_signal        = false;
static volatile bool pause_signal       = false;
static pthread_t* threads               = NULL;
static i64 real_thread_count            = 0;
static volatile i64* votes_scoreboard   = NULL;
static volatile i64* votes_count        = NULL;
static q2pc_trans* trans;


void cleanup()
{
    stop_signal = true;
    __sync_synchronize(); //Full fence

    if(threads){
        for(int i = 0; i < real_thread_count; i++){
            pthread_join(threads[i],NULL);
        }
    }

    if(trans){
        trans->delete(trans);
    }

}


//Signal handler to terminate early
void term(int signo)
{
    ch_log_info("Terminating...\n");
    (void)signo;
    cleanup();

    ch_log_info("Terminating... Done.\n");
    exit(0);
}


//Pause/unpause worker threads
void dopause_all(){ pause_signal = true;  __sync_synchronize(); }
void unpause_all(){ pause_signal = false; __sync_synchronize(); }

//Wait for all clients to connect
void do_connectall(i64 client_count)
{
    cons = CH_ARRAY_NEW(TRANS_CONN,client_count,NULL);
    if(!cons){ ch_log_fatal("Cannot allocate connections array\n"); }

    for(int i = 0; i < client_count; i++){
        trans->connect(trans, cons->off(cons,i));
    }

}



void server_init(const i64 thread_count, const i64 client_count , const transport_s* transport)
{

    //Signal handling for the main thread
    signal(SIGHUP,  term);
    signal(SIGKILL, term);
    signal(SIGTERM, term);
    signal(SIGINT, term);

    //Set up and init the voting scoreboard
    votes_scoreboard = (volatile i64*)aligned_alloc(sizeof(i64), sizeof(i64) * client_count);
    if(!votes_scoreboard){
        ch_log_fatal("Could not allocate memory for votes scoreboard\n");
    }
    bzero((void*)votes_scoreboard,sizeof(i64) * client_count);


    //Set up all the connections
    ch_log_info("Waiting for clients to connect...\n\r");
    trans = trans_factory(transport);
    do_connectall(client_count);
    ch_log_info("Waiting for clients to connect... Done.\n");


    //Calculate the connection to thread mappings
    i64 cons_per_thread = MAX(client_count / thread_count, 1);
    real_thread_count  = MIN(thread_count, client_count);
    i64 lo = 0;
    i64 hi = lo + cons_per_thread;

    votes_count = (volatile i64*)aligned_alloc(sizeof(i64), sizeof(i64) * real_thread_count);
    if(!votes_count){
        ch_log_fatal("Could not allocate memory for votes counter\n");
    }
    bzero((void*)votes_count,sizeof(i64) * real_thread_count);


    //Fire up the threads
{'fraction_non_alphanumeric': 0.12388083998046558, 'fraction_numerical': 0.018557707960279994, 'mean_word_length': 3.302521008403361, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 17, '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': '2326683', 'n_tokens_mistral': 2158, 'n_tokens_neox': 2037, 'n_words': 505}