test3_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 <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 active_cnx_state_handler(hsmsd_cnx_state_t state);
static void passive_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 sem_t cnx_ready_sem;


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];

    sem_init(&cnx_ready_sem, 0, 0);

    r = hsmsd_alloc_handle(&active_cnx, "hsms_local_active");
    printf("alloc hsms_local_active returned %d\n", r);
    if(r) exit(-1);

    r = hsmsd_alloc_handle(&passive_cnx, "hsms_local_passive");
    printf("alloc hsms_local_passive returned %d\n", r);
    if(r) exit(-2);

    r = hsmsd_subscribe_for_msgs(active_cnx,    (hsmsd_msg_handler_t)active_hsms_msg_handler);
    fprintf(stdout, "hsmsd_subscribe_for_msgs(active) returned %d\n", r);
    if(r) exit(-3);

    r = hsmsd_subscribe_for_msgs(passive_cnx,   (hsmsd_msg_handler_t)passive_hsms_msg_handler);
    fprintf(stdout, "hsmsd_subscribe_for_msgs(passive) returned %d\n", r);
    if(r) exit(-4);

    r = hsmsd_subscribe_for_cnx_state(active_cnx, (hsmsd_cnx_state_handler_t)active_cnx_state_handler);
    fprintf(stdout, "hsmsd_subscribe_for_cnx_state(active) returned %d\n", r);
    if(r) exit(-5);

    r = hsmsd_subscribe_for_cnx_state(passive_cnx, (hsmsd_cnx_state_handler_t)passive_cnx_state_handler);
    fprintf(stdout, "hsmsd_subscribe_for_cnx_state(passive) returned %d\n", r);
    if(r) exit(-6);

    r = hsmsd_cnx_start(passive_cnx);
    fprintf(stdout, "hsmsd_cnx_start(passive) returned %d\n", r);
    if(r) exit(-8);
    
    //sleep(1);

    r = hsmsd_cnx_start(active_cnx);
    fprintf(stdout, "hsmsd_cnx_start(active) returned %d\n", r);
    if(r) exit(-7);

    sem_wait(&cnx_ready_sem);

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

    start_threads((void*)&active_cnx);
    fprintf(stdout, "threads started\n");
    
    while(1)
    {
        sleep(2);

        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);
    }

    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"};

void active_cnx_state_handler(hsmsd_cnx_state_t state)
{
    if(CNX_SELECTED == state)
    {
        sem_post(&cnx_ready_sem);
    }
    fprintf(stdout, "active_cnx_state_handler: %s\n", str_state[state]);

}

void passive_cnx_state_handler(hsmsd_cnx_state_t state)
{
    fprintf(stdout, "passive_cnx_state_handler: %s\n", str_state[state]);
}

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)
{
    fprintf(stdout, "thread function %d\n", num);
    hsmsd_handle_t h = *((hsmsd_handle_t*)data);
    while(1)
    {
        usleep(msg_dict[num].us_timeout);
        msg_dict[num].primary->sysbytes += 1;
        msg_dict[num].secondary->sysbytes += 1;
        if(0 == hsmsd_cnx_send_msg(h, msg_dict[num].primary))
        {
            sem_post(&msg_dict[num].sent_sem);
        }
        else
        {
            fprintf(stdout, "thread function %d: exit...\n", num);
            sem_post(&msg_dict[num].err_sem);
            pthread_exit(NULL);
        }
    }
    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);
}

---