test4_main.c

/*   
 *   (c) Copyright 2008 Philipp Skadorov (philipp_s@users.sourceforge.net)
 *
 *   This file is part of FREESECS.
 *
 *   FREESECS is free software: you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation, either version 3 of the License, or
 *   (at your option) any later version.
 *
 *   FREESECS is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with FREESECS, see COPYING.
 *   If not, see <http://www.gnu.org/licenses/>.
 */
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>
#include "hsms_msgs.hpp"
#include "hsmsd_cli.h"

static void active_hsms_msg_handler(hsmsd_msg_t* msg);
static void passive_hsms_msg_handler(hsmsd_msg_t* msg);
static void cnx_state_handler(hsmsd_cnx_state_t state);
static void cnx_state_handler(hsmsd_cnx_state_t state);
static void prepare_dict(void);
static void start_threads(void*);
static void* thread_func0(void*);
static void* thread_func1(void*);
static void* thread_func2(void*);
static void* thread_func3(void*);
static void* thread_funcx(int num, void* data);

static pthread_cond_t cnx_ready;
static pthread_mutex_t cnx_mx = PTHREAD_MUTEX_INITIALIZER;

static struct
{
    sem_t sent_sem;
    sem_t recv_sem;
    sem_t err_sem;
    unsigned long us_timeout;
    unsigned int transaction_id;
    hsmsd_msg_t* primary;
    hsmsd_msg_t* secondary;
}msg_dict[4] = 
{
    {.us_timeout = 1, .transaction_id = 0 << 24, 
     .primary = (hsmsd_msg_t*)&s1f13_msg,   .secondary = (hsmsd_msg_t*)&s1f14_msg},
    {.us_timeout = 1, .transaction_id = 1 << 24, 
     .primary = (hsmsd_msg_t*)&s3f17_msg,   .secondary = (hsmsd_msg_t*)&s3f18_msg},
    {.us_timeout = 1, .transaction_id = 2 << 24, 
     .primary = (hsmsd_msg_t*)&s16f15_msg,  .secondary = (hsmsd_msg_t*)&s16f16_msg},
    {.us_timeout = 1, .transaction_id = 3 << 24, 
     .primary = (hsmsd_msg_t*)&s14f9_msg,   .secondary = (hsmsd_msg_t*)&s14f10_msg}
};


static hsmsd_handle_t active_cnx, passive_cnx;

int main(int argc, char **argv)
{
    int r, sent[4], recv[4], err[4];
    int is_active = 0;

    hsmsd_handle_t              *pcnx;
    hsmsd_cnx_state_t           state;
    hsmsd_msg_handler_t         msg_handler;
    const char                  *cnx_name;
    int                         close_timeout, open_timeout;
    int                         counter = 0;


    if(argc < 2)
    {
        printf("usage: hsmsd_test4 -[a,p]\n");
        exit(-1);
    }
    if(0 == strcmp(argv[1], "-a"))
    {
        is_active = 1;
    }
    else if(0 == strcmp(argv[1], "-p"))
    {
        is_active = 0;
    }
    else
    {
        printf("usage: hsmsd_test4 -[a,p]\n");
        exit(-1);
    }

    //sem_init(&cnx_ready_sem, 0, 0);
    pthread_cond_init(&cnx_ready, NULL);

    if(is_active)
    {
        pcnx            = &active_cnx;
        msg_handler     = active_hsms_msg_handler;
        cnx_name        = "hsms_test4_active";
        close_timeout   = 3;
        open_timeout    = 3;
    }
    else
    {
        pcnx            = &passive_cnx;
        msg_handler     = passive_hsms_msg_handler;
        cnx_name        = "hsms_test4_passive";
        close_timeout   = 10;
        open_timeout    = 10;
    }

    r = hsmsd_alloc_handle(pcnx, cnx_name);
    printf("alloc cnx returned %d\n", r);
    if(r) exit(-1);

    r = hsmsd_subscribe_for_msgs(*pcnx,msg_handler);
    fprintf(stdout, "hsmsd_subscribe_for_msgs returned %d\n", r);
    if(r) exit(-3);

    r = hsmsd_subscribe_for_cnx_state(*pcnx, cnx_state_handler);
    fprintf(stdout, "hsmsd_subscribe_for_cnx_state returned %d\n", r);
    if(r) exit(-5);

    prepare_dict();
    fprintf(stdout, "dict prepared\n");

    start_threads((void*)pcnx);
    fprintf(stdout, "threads started\n");
    
    r = hsmsd_cnx_start(*pcnx);
    fprintf(stdout, "hsmsd_cnx_start returned %d\n", r);
    if(r) exit(-8);
    
    pthread_mutex_lock(&cnx_mx);
    pthread_cond_wait(&cnx_ready, &cnx_mx);
    pthread_mutex_unlock(&cnx_mx);

    while(1)
    {
        counter++;
        sleep(1);

        sem_getvalue(&msg_dict[0].sent_sem, &sent[0]);
        sem_getvalue(&msg_dict[1].sent_sem, &sent[1]);
        sem_getvalue(&msg_dict[2].sent_sem, &sent[2]);
        sem_getvalue(&msg_dict[3].sent_sem, &sent[3]);

        sem_getvalue(&msg_dict[0].recv_sem, &recv[0]);
        sem_getvalue(&msg_dict[1].recv_sem, &recv[1]);
        sem_getvalue(&msg_dict[2].recv_sem, &recv[2]);
        sem_getvalue(&msg_dict[3].recv_sem, &recv[3]);

        sem_getvalue(&msg_dict[0].err_sem, &err[0]);
        sem_getvalue(&msg_dict[1].err_sem, &err[1]);
        sem_getvalue(&msg_dict[2].err_sem, &err[2]);
        sem_getvalue(&msg_dict[3].err_sem, &err[3]);
        
        fprintf(stdout, "sent:\t%d\t|\t%d\t|\t%d\t|\t%d\n", 
                                 sent[0], sent[1], sent[2], sent[3]);
        fprintf(stdout, "recv:\t%d\t|\t%d\t|\t%d\t|\t%d\n", 
                                 recv[0], recv[1], recv[2], recv[3]);
        fprintf(stdout, "err:\t%d\t|\t%d\t|\t%d\t|\t%d\n", 
                                 err[0], err[1], err[2], err[3]);
        fflush(stdout);

        if(close_timeout > 0)
        {
            printf("main: get_cnx_state -->\n");
            hsmsd_cnx_get_state(*pcnx, &state);
            printf("main: get_cnx_state: %d <--\n", state);
            printf("main: counter: %d\n", counter);
            if(CNX_SELECTED == state && counter >= close_timeout)
            {
                counter = 0;
                printf("stopping %s...\n", cnx_name);
                fflush(stdout);
                hsmsd_cnx_stop(*pcnx);
            }
            else if(CNX_NOT_CONNECTED == state && counter >= open_timeout)
            {
                counter = 0;
                printf("starting %s...\n", cnx_name);
                fflush(stdout);
                hsmsd_cnx_start(*pcnx);

                fprintf(stdout, "main thread: waiting for cnx ready sem...\n");
                fflush(stdout);
                pthread_mutex_lock(&cnx_mx);
                pthread_cond_wait(&cnx_ready, &cnx_mx);
                pthread_mutex_unlock(&cnx_mx);
                fprintf(stdout, "main thread: cnx ready sem UP!\n");
                fflush(stdout);
            }
        }
    }

    return 0;
}

void active_hsms_msg_handler(hsmsd_msg_t* msg)
{
    int pool_num = msg->sysbytes >> 24;
    if(4 > pool_num)
    {
        sem_post(&msg_dict[pool_num].recv_sem);
    }
    else
    {
        fprintf(stdout, ">>>!!!!!unknown id: %d!!!!!<<<\n",pool_num);
    }
    free(msg);

}

void passive_hsms_msg_handler(hsmsd_msg_t* msg)
{
    int r;
    switch(msg->function)
    {
        case 9:
            r = hsmsd_cnx_send_msg(passive_cnx, (hsmsd_msg_t*)&s14f10_msg);
        break;
        case 13:
            r = hsmsd_cnx_send_msg(passive_cnx, (hsmsd_msg_t*)&s1f14_msg);
        break;
        case 15:
            r = hsmsd_cnx_send_msg(passive_cnx, (hsmsd_msg_t*)&s16f16_msg);
        break;
        case 17:
            r = hsmsd_cnx_send_msg(passive_cnx, (hsmsd_msg_t*)&s3f18_msg);
        break;
        default:
        break;
    }
    free(msg);
}

static char str_state[][32] = {"NOT CONNECTED", "NOT SELECTED", "SELECTED"};
static hsmsd_cnx_state_t cached_state;
void cnx_state_handler(hsmsd_cnx_state_t state)
{
    cached_state = state;
    if(CNX_SELECTED == state)
    {
        pthread_mutex_lock(&cnx_mx);
        pthread_cond_broadcast(&cnx_ready);
        pthread_mutex_unlock(&cnx_mx);
    }
    fprintf(stdout, "cnx_state_handler: %s\n", str_state[state]);
    fflush(stdout);
}


void prepare_dict(void)
{
    sem_init(&msg_dict[0].sent_sem, 0, 0);
    sem_init(&msg_dict[1].sent_sem, 0, 0);
    sem_init(&msg_dict[2].sent_sem, 0, 0);
    sem_init(&msg_dict[3].sent_sem, 0, 0);

    sem_init(&msg_dict[0].recv_sem, 0, 0);
    sem_init(&msg_dict[1].recv_sem, 0, 0);
    sem_init(&msg_dict[2].recv_sem, 0, 0);
    sem_init(&msg_dict[3].recv_sem, 0, 0);

    sem_init(&(msg_dict[0].err_sem), 0, 0);
    sem_init(&(msg_dict[1].err_sem), 0, 0);
    sem_init(&(msg_dict[2].err_sem), 0, 0);
    sem_init(&(msg_dict[3].err_sem), 0, 0);
}

void start_threads(void* pdata)
{
    pthread_t id0, id1, id2, id3;
    pthread_create(&id0, NULL, thread_func0, pdata);
    pthread_create(&id1, NULL, thread_func1, pdata);
    pthread_create(&id2, NULL, thread_func2, pdata);
    pthread_create(&id3, NULL, thread_func3, pdata);
}

void* thread_funcx(int num, void* data)
{
    int send_failure = 0;
    hsmsd_cnx_state_t state;

    hsmsd_handle_t h = *((hsmsd_handle_t*)data);

    fprintf(stdout, "thread function %d\n", num);

    for(;;)
    {
        hsmsd_cnx_get_state(h, &state);
        if(CNX_SELECTED != state)
        {
            fprintf(stdout, "thread function %d: waiting for cnx ready sem...\n", num);
            pthread_mutex_lock(&cnx_mx);
            pthread_cond_wait(&cnx_ready, &cnx_mx);
            pthread_mutex_unlock(&cnx_mx);
            fprintf(stdout, "thread function %d: cnx ready sem UP!\n", num);
        }
        for(;;)
        {
            usleep(msg_dict[num].us_timeout);
            msg_dict[num].primary->sysbytes += 1;
            msg_dict[num].secondary->sysbytes += 1;

            send_failure = hsmsd_cnx_send_msg(h, msg_dict[num].primary);

            if(0 == send_failure)
            {

                sem_post(&msg_dict[num].sent_sem);
            }
            else
            {
                fprintf(stdout, "thread function %d: error sending msg s%df%d - %s\n", 
                            num, msg_dict[num].primary->stream, 
                            msg_dict[num].primary->function,
                            strerror(send_failure));
                sem_post(&msg_dict[num].err_sem);
                //pthread_exit(NULL);
                break;//go through cnx state check
            }
        }
    }
    fprintf(stdout, "thread function %d: exit...\n", num);
    return NULL;
}

void* thread_func0(void* data)
{
    return thread_funcx(0, data);
}

void* thread_func1(void* data)
{
    return thread_funcx(1, data);
}

void* thread_func2(void* data)
{
    return thread_funcx(2, data);
}

void* thread_func3(void* data)
{
    return thread_funcx(3, data);
}

---