GDP

/* vim: set ai sw=4 sts=4 ts=4 : */

#include <stdio.h>

#include <stdarg.h>

#include <time.h>

#include <unistd.h>

#include <string.h>

#include <ep/ep_app.h>

#include <ep/ep_dbg.h>

#include <gdp/gdp.h>

#include <pthread.h>

#if __APPLE__

#define SET_THREAD_NAME(n)        pthread_setname_np(n)

#else

#define SET_THREAD_NAME(n)        pthread_setname_np(pthread_self(), n)

#endif

pthread_mutex_t lock;

time_t start;

struct ArgStruct

{

    char *logname;

    int sleepBefore;

    int numRecords;

    char *threadName;

};

void

localPrint(const char *s, ...)

{

    char tmpStr[100];

    va_list fmt;

    va_start(fmt, s);

    pthread_t cur_thread = pthread_self();

    pthread_getname_np(cur_thread, tmpStr, 100);

    pthread_mutex_lock(&lock);

    printf("%9.6f ## %s] ", (double)(time(NULL)-start), tmpStr);

    vprintf(s, fmt);

    fflush(stdout);

    pthread_mutex_unlock(&lock);

    va_end(fmt);

}

void

*appendThread(void *arguments)

{

    // do this, because we can only pass one argument to pthread_create

    struct ArgStruct *args = (struct ArgStruct *)arguments;

    char *logname = args->logname;

    int sleepBefore = args->sleepBefore;

    int numRecords = args->numRecords;

    char *threadName = args->threadName;

    // rest of the logic

    int i;

    char data[100];

    gdp_gcl_t *gcl;

    gdp_name_t gcliname;

    EP_STAT estat;

    SET_THREAD_NAME(threadName);

    // open the GCL

    gdp_parse_name(logname, gcliname);

    // no error checks for now...

    estat = gdp_gcl_open(gcliname, GDP_MODE_AO, NULL, &gcl);

        ep_app_message(estat, "%s: gdp_gcl_open", threadName);

    gdp_datum_t *datum = gdp_datum_new();

    sleep(sleepBefore);

    for (i=0; i<numRecords; i++)

    {

        sleep(1);

        sprintf(data, "\"%s-%s-%d\"", logname, threadName, i);

        localPrint("Appending %s to %s\n", data, logname);

        // do the actual append

        gdp_buf_write(gdp_datum_getbuf(datum), data, strlen(data));

        estat = gdp_gcl_append(gcl, datum);

                ep_app_message(estat, "%s: gdp_gcl_append", threadName);

    }

    gdp_datum_free(datum);

    return NULL;

}

void

*subscribeThread(void *arguments)

{

    struct ArgStruct *args = arguments;

    char *logname = args->logname;

    char data[100];

    gdp_gcl_t *gcl;

    gdp_name_t gcliname;

    EP_STAT estat;

    logname = args->logname;

    SET_THREAD_NAME(args->threadName);

    // open the GCL

    gdp_parse_name(logname, gcliname);

    // no error checks for now...

    estat = gdp_gcl_open(gcliname, GDP_MODE_RO, NULL, &gcl);

        ep_app_message(estat, "%s: gdp_gcl_open", args->threadName);

    estat = gdp_gcl_subscribe(gcl, 0, 0, NULL, NULL, NULL);

        ep_app_message(estat, "%s: gdp_gcl_subscribe", args->threadName);

    for (;;)

    {

        // get next event

        gdp_event_t *gev = gdp_event_next(gcl, 0);

        gdp_datum_t *datum = gdp_event_getdatum(gev);

        gdp_buf_read(gdp_datum_getbuf(datum), data, 100);        

        localPrint("Got data: %s on %s\n", data, logname);

        gdp_event_free(gev);

    }

    return NULL;

}

int

main(int argc, char *argv[])

{

        int opt;

        while ((opt = getopt(argc, argv, "D:")) > 0)

        {

                switch (opt)

                {

                case 'D':

                        ep_dbg_set(optarg);

                        break;

                }

        }

        argc -= optind;

        argv += optind;

    if (argc != 4)

    {

        printf("Usage: %s [-Ddbg] numA numS delay numRec\n", argv[0]);

        return -1;

    }

    int i, numA, numS, delay, numR;

    //char tmpStr[100];

    numA = (int) strtol(argv[0], NULL, 10);

    numS = (int) strtol(argv[1], NULL, 10);

    delay = (int) strtol(argv[2], NULL, 10);

    numR = (int) strtol(argv[3], NULL, 10);

    pthread_t *appendTIDs = (pthread_t *)malloc(sizeof(pthread_t)*numA);

    gdp_init(NULL);

    start = time(NULL);

    SET_THREAD_NAME("MainThread");

    // create append threads

    for (i=0; i<numA; i++)

    {

        localPrint("Starting AThread%d\n", i+1);

        char *logname = malloc(sizeof(char)*100);

        sprintf(logname, "log%d", i+1);

        char *threadName = malloc(sizeof(char)*100);

        sprintf(threadName, "AThread%d", i+1);

        // create a structure to hold the arguments

        struct ArgStruct *args = malloc(sizeof (struct ArgStruct));

        args->logname = logname;

        args->sleepBefore = delay*(i+1);

        args->numRecords = numR;

                args->threadName = threadName;

        // do the actual thread creation

        pthread_create(&appendTIDs[i], NULL, appendThread, args);

    }

    // create subscription threads

    for (i=0; i<numS; i++)

    {

        pthread_t t;

        localPrint("Starting SThread%d\n", i+1);

        char *logname = malloc(sizeof(char)*100);

        sprintf(logname, "log%d", i+1);

        char *threadName = malloc(sizeof(char)*100);

        sprintf(threadName, "SThread%d", i+1);

        // create a structure to hold the arguments

        struct ArgStruct *args = malloc(sizeof (struct ArgStruct));

        args->logname = logname;

                args->threadName = threadName;

        pthread_create(&t, NULL, subscribeThread, args);

    }

    // join on append threads

    for (i=0; i<numA; i++)

    {

        pthread_join(appendTIDs[i], NULL);

    }

    return 0;

}