ap_net.tests.c

/** \file ap_net/ap_net.tests.c
 * \brief Testing facility for ap_net module
 */
#include "ap_net.h"
#include "../ap_utils.h"
#include "../ap_log.h"
#include <assert.h>
#include <fcntl.h>
#include <stdio.h>
#include <sys/stat.h>
#include <time.h>
#include <unistd.h>


/* comment out if you want to perform tcp testing only */
#define TEST_UDP
#define pool_of_pools_size 1000

#define max_clients 10
#define max_tests_per_client 100
#define CLIENT_POOL_SIZE 20
#define CLIENT_MAX_LIFETIME max_tests_per_client

#define CONNECTION_TIMEOUT 3000
#define TCP_POLLER_DEBUG 0
#define UDP_POLLER_DEBUG 0

#define SERVER_DEBUG_LEVEL 0
#define CLIENT_DEBUG_LEVEL 0

const char *log_file_name = "ap_net.tests.log";

const char *localhost_str = "127.0.0.1";
const int tcp_port = 22222, udp_port = 22223; /* server listener */

/* prototypes of client-server main functions */
void go_client(void);

/* prototypes of callbacks */
int server_callback(struct ap_net_connection_t *conn, int signal_type);
void generate_sequences(void);

/* the test sequences are generated by randomly picking the elements of this string
 * so you can place more symbols of one type here to make this kind of test to be more frequent
 * For explanation of the meaning of each symbol see below.
 */
const char *test_letters = "+-sc"; /* all test frequencies are equal */
/*const char *test_letters = "++++++++++--sssssccccc";*/
const char *test_message = "cs_sequence is used as testing plan for each of clients."
        " every char in string is a option for each new connection:"
        " '+' - normal connect/send/recv/close"
        " '-' - drop on accept (TCP only)"
        " 's' - drop on server's answer timeout"
        " 'c' - drop on client's message timeout"
        " Sequence lengths can be different."
        " Test quality report is produced by the server."
        " Per-client lists are generated automatically at random."
        " TCP and UDP protocols test sequences are equal";

int test_message_len;

/* a little helper to determine pool's type */
#define POOL_IS_TCP(p) ((p)->flags & AP_NET_POOL_FLAGS_TCP ? 1 : 0)
#define TEST_IS_TCP 1
#define TEST_IS_UDP 0

/** \brief The Testing Plan */
struct
{
    char *plan;
    char *results;
    int count, done, good, bad;
} tests[2][max_clients];

#define TCP_TEST_RESULT tests[TEST_IS_TCP][ud->client_id].results[ud->test_index]
#define UDP_TEST_RESULT tests[TEST_IS_UDP][ud->client_id].results[ud->test_index]

const char *state_text[] = {
#define STATE_CONNECTED    0
        "Connected",
#define STATE_TEST_ID_SENT 1
        "ID sent",
#define STATE_MSG_SENT     2
        "Message sent to server",
#define STATE_GOT_MESSAGE  3
        "Client's message received",
#define STATE_GOT_ANSWER   4
        "Answer received from server",
#define STATE_ANSWER_SENT  5
        "Answer sent to client"
};

/** \brief user-data record that will be attached to connection->user_data on server side of test
*/
typedef struct
{
    int client_id;
    int is_control; /* true if this is control connection */
    int test_index;
    char test_type;
    int state; /* see STATE_* */
} server_userdata;

/** \brief user-data record that will be attached to connection->user_data on client side of test
*/
typedef struct
{
    int test_index;
    char test_type;
    int state; /* see STATE_* */
    char send_buf[50]; /* buffer for preparing and sending data */
    int send_len;
} client_userdata;

const char *id_packet_marker = "bIDp";
/** \brief current test settings that client send to server via control channel after new connection was made
*/
typedef struct
{
    char marker[5]; /* == *id_packet_market */
    int is_tcp;
    in_port_t port;
    char test_type;
    int test_index;
    int client_index;
} binary_id_packet;

/* we receive id packets out-of-sync with new connections, so some part of them is stalled for a while */
#define ID_STACK_MAX max_clients * max_tests_per_client
binary_id_packet id_stack[ID_STACK_MAX];

const char *udp_dummy_msg = "UDP_dummy"; /* used as first packet on UDP channel to make connection up */
int client_index; /* global, so any forked client's functions will know it instantly */

/* Teh very first of client's TCP connections is used as control stream to synchronize current test data with server
  This marker is the signal for server side to treat connection that sent it as control */
const char *control_conn_marker = "DEBUG";
struct ap_net_connection_t *control_conns[max_clients]; /* direct links to clients control channels */

/* Those pool will be used for client-server mesaging tests */
struct ap_net_conn_pool_t *tcp_pool;
struct ap_net_conn_pool_t *udp_pool;

/* ******************************************************* */
int main(void)
{
    struct ap_net_conn_pool_t **pool_of_pools;
    int i;
    int n;
    int log_file_handle;
    time_t start_time, last_event_time;
    ap_net_connection_t *conn;
    server_userdata *ud;


    ap_log_debug_to_tty = 1; /* we like to see immediately if some trouble happens */

    test_message_len = strlen(test_message);

    for(i = 0; i < ID_STACK_MAX; ++i ) /* clearing */
        id_stack[i].client_index = -1;

    /* *********************************************************** */
    /* *********************************************************** */
    /* *********************************************************** */
    printf("test: create and destroy 1000 pools with full init\n");
    fflush(stdout);
    /* *********************************************************** */
    /* *********************************************************** */
    /* *********************************************************** */

    pool_of_pools = malloc( pool_of_pools_size * sizeof(void *));
    assert(pool_of_pools != NULL);

    for( i = 0; i < pool_of_pools_size; ++i )
    {
         /* basics and not memory hungry settings. just for testing of the create/destroy functions */
         pool_of_pools[i] = ap_net_conn_pool_create(AP_NET_POOL_FLAGS_TCP, 10, 5000, 512, NULL);
         assert(pool_of_pools[i] != NULL);
    }

    /* now randomly changing the max number of connections for pools to test set_max_connections() */
    for( i = 0; i < pool_of_pools_size; ++i )
    {
        n = rand() % 19 + 1;

        if ( ! ap_net_conn_pool_set_max_connections(pool_of_pools[i], n, 512))
        {
            printf("!ERROR: pool[%d] resize to %d failed:\n%s\n", i, n, ap_error_get_string());
            exit(1);
        }
    }

    for( i = 0; i < pool_of_pools_size; ++i )
        ap_net_conn_pool_destroy(pool_of_pools[i], 1);

    /* *********************************************************** */
    /* *********************************************************** */
    /* *********************************************************** */
    printf("test: Server + %d tcp/udp clients, up to %d tests\n", max_clients, max_clients * max_tests_per_client);
    fflush(stdout);

    generate_sequences(); /* this is where the sequentce of tests is generated for each client */

    for ( i = 0; i < max_clients; ++i )
        control_conns[i] = NULL;

    /* tcp pool create and init */
    tcp_pool = ap_net_conn_pool_create(AP_NET_POOL_FLAGS_TCP, max_clients * max_tests_per_client,
                                       CONNECTION_TIMEOUT, strlen(test_message) * 2, server_callback);

    if ( tcp_pool == NULL
        || ! ap_net_conn_pool_set_str_addr(tcp_pool, (char *)localhost_str, tcp_port)
        || -1 == ap_net_conn_pool_listener_create(tcp_pool, 1, 1) )
    {
        printf("* !ERROR: tcp_pool: %s\n", ap_error_get_string());
        exit(1);
    }

    tcp_pool->poller->debug = TCP_POLLER_DEBUG;

    /* udp pool create and init */
    udp_pool = ap_net_conn_pool_create(0, max_clients * max_tests_per_client, CONNECTION_TIMEOUT, strlen(test_message) * 2, server_callback);

    if ( udp_pool == NULL
        || ! ap_net_conn_pool_set_ip4_addr(udp_pool, INADDR_LOOPBACK, udp_port)
        || -1 == ap_net_conn_pool_listener_create(udp_pool, 1, 1) )
    {
        printf("* !ERROR: udp_pool: %s\n", ap_error_get_string());
        exit(1);
    }

    udp_pool->poller->debug = UDP_POLLER_DEBUG;

    log_file_handle = open(log_file_name, O_WRONLY | O_CREAT | O_TRUNC, S_IRUSR | S_IWUSR);

    if ( log_file_handle <= 0 )
    {
        printf("Failed to open log file: %s %s\n", log_file_name, strerror(errno));
        exit(1);
    }

    if ( ! ap_log_add_debug_handle(log_file_handle) )
    {
        printf("Failed to register log file handle for debug output: %s\n", ap_error_get_string());
        exit(1);
    }

    /* ------------------------------------------------------------------
       Spawning clients
    */
    for ( client_index = 0; client_index < max_clients; ++client_index )
    {
        if ( 0 == fork() )
        {
            /* we'll reuse those varables for client's operations. also another test of correct re-use */
            ap_net_conn_pool_destroy(tcp_pool, 1);
            ap_net_conn_pool_destroy(udp_pool, 1);

            go_client();

            exit(0);
        }
    }

    ap_log_debug_level = SERVER_DEBUG_LEVEL;

/* ###############################################################################################################
 * Main loop
 * ###############################################################################################################
 */
    start_time = last_event_time = time(NULL);

    while ( (time(NULL) - start_time < 5) || tcp_pool->used_slots > 0 || udp_pool->used_slots > 0 )
    {
        if (time(NULL) - last_event_time > 5 )
        {
            ap_log_debug_log("! Finishing: last event was long ago. tcp slots used: %d, udp slots: %d\n", tcp_pool->used_slots, udp_pool->used_slots);

            break;
        }

        usleep(10000);

        if ( ! ap_net_conn_pool_poll(tcp_pool) )
        {
            ap_log_debug_log("* !ERROR: tcp_pool: %s\n", ap_error_get_string());
            exit(1);
        }

        for (i = 0; i < tcp_pool->max_connections; ++i ) /* checking for unassigned Ids */
        {
            conn = &tcp_pool->conns[i];
            ud = conn->user_data;

            if( ! (bit_is_set(conn->state, AP_NET_ST_CONNECTED)) || ud->client_id != -1 )
                continue;

            for( n = 0; n < ID_STACK_MAX; ++n )
            {
                if( id_stack[n].is_tcp && id_stack[n].port == conn->remote.addr4.sin_port )
                    break;
            }

            if ( n == ID_STACK_MAX ) /* not in stack, skipping */
                continue;

            ud->test_index = id_stack[n].test_index;
            ud->test_type  = id_stack[n].test_type;
            ud->client_id  = id_stack[n].client_index;
            id_stack[n].client_index = -1; /* free the stack slot */

            ap_log_debug_log("%c", ud->test_type);
        }

        if ( tcp_pool->poller->events_count )
            last_event_time = time(NULL);

#ifdef TEST_UDP
        if ( ! ap_net_conn_pool_poll(udp_pool) )
        {
            ap_log_debug_log("* !ERROR: udp_pool: %s\n", ap_error_get_string());
            exit(1);
        }

        for (i = 0; i < udp_pool->max_connections; ++i ) /* checking for unassigned Ids */
        {
            conn = &udp_pool->conns[i];
            ud = conn->user_data;

            if( ! (bit_is_set(conn->state, AP_NET_ST_CONNECTED)) || ud->client_id != -1 )
                continue;

            for( n = 0; n < ID_STACK_MAX; ++n )
            {
                if ( ! id_stack[n].is_tcp && id_stack[n].port == conn->remote.addr4.sin_port )
                    break;
            }

            if ( n == ID_STACK_MAX ) /* not in stack, skipping */
                continue;

            ud->test_index = id_stack[n].test_index;
            ud->test_type  = id_stack[n].test_type;
            ud->client_id  = id_stack[n].client_index;
            id_stack[n].client_index = -1; /* free the stack slot */

            ap_log_debug_log("%c", ud->test_type);
        }

        if ( udp_pool->poller->events_count )
            last_event_time = time(NULL);
#endif
    } /* while ( has activity ) */

    /* *********************************************************** */
    /* *********************************************************** */
    /* *********************************************************** */
    /* DONE HERE. REPORTING STATS */
    /* *********************************************************** */
    /* *********************************************************** */
    /* *********************************************************** */

    ap_log_debug_log("\n");

    for (i = 0; i < max_clients; ++i)
    {
        ap_log_debug_log("Client #%4d: %s\n         TCP: %s\n", i, tests[TEST_IS_TCP][i].plan, tests[TEST_IS_TCP][i].results);
        ap_log_debug_log("\t\tTotal: %d, Good %d, Bad: %d\n", tests[TEST_IS_TCP][i].count, tests[TEST_IS_TCP][i].good, tests[TEST_IS_TCP][i].bad);
#ifdef TEST_UDP
        ap_log_debug_log("         UDP: %s\n", tests[TEST_IS_UDP][i].results);
#endif
        ap_log_debug_log("\t\tTotal: %d, Good %d, Bad: %d\n", tests[TEST_IS_UDP][i].count, tests[TEST_IS_UDP][i].good, tests[TEST_IS_UDP][i].bad);
    }

    for( n = i = 0; i < ID_STACK_MAX; ++i)
        if ( id_stack[i].client_index != -1 && id_stack[i].test_type != '-' )
        {
            ++n;
            ap_log_debug_log("%02d) Client #%d %s test index: %d, '%c'\n", n, id_stack[i].client_index, id_stack[i].is_tcp ? "TCP" : "UDP", id_stack[i].test_index, id_stack[i].test_type);
        }

    if(n)
        ap_log_debug_log("! ERROR: %d id packets still in stack\n", n);

    if ( tcp_pool->used_slots > max_clients )
    {
        ap_log_debug_log ("TCP pool stale connections:\n");

        for (i = 0; i < tcp_pool->max_connections; ++i)
        {
            if ( ! (tcp_pool->conns[i].state & AP_NET_ST_CONNECTED) )
                continue;

            conn = &tcp_pool->conns[i];
            ud = conn->user_data;

            if ( ud->is_control )
                continue;

            ap_log_debug_log("%d) Client %d, expire in: %ldms, State: ", conn->idx, ud->client_id, ap_utils_timespec_elapsed(NULL, &conn->expire, NULL));

            if ( bit_is_set(conn->state, AP_NET_ST_BUSY) )
                ap_log_debug_log(" Busy,");

            if ( bit_is_set(conn->state, AP_NET_ST_DISCONNECTION) )
                ap_log_debug_log(" Disc,");

            if ( bit_is_set(conn->state, AP_NET_ST_ERROR) )
                ap_log_debug_log(" Error,");

            if ( bit_is_set(conn->state, AP_NET_ST_EXPIRED) )
                ap_log_debug_log(" Expired,");

            if ( bit_is_set(conn->state, AP_NET_ST_IN) )
                ap_log_debug_log(" DataIn,");

            if ( bit_is_set(conn->state, AP_NET_ST_OUT) )
                ap_log_debug_log(" DataOut,");

            ap_log_debug_log(" buf p: %d, f: %d, s: %d, Test #%d '%c'\n", conn->bufpos, conn->buffill, conn->bufsize, ud->test_index, ud->test_type);
        }
    }

    ap_log_debug_log("test: done. Elapsed %d seconds\n", (int)(time(NULL) - start_time));

    ap_net_conn_pool_destroy(tcp_pool,1);
    ap_net_conn_pool_destroy(udp_pool,1);

    /* *********************************************************** */
    /* *********************************************************** */
    /* *********************************************************** */
    ap_log_remove_debug_handle(log_file_handle);
    close(log_file_handle);

    exit(0);
}

/* ******************************************************** */
void generate_sequences(void)
{
    int client_idx, seq_len, test_idx, test_letters_len;
    char c;


    test_letters_len = strlen(test_letters);

    for (client_idx = 0; client_idx < max_clients; ++client_idx)
    {
        seq_len = rand() % (max_tests_per_client / 2) + (max_tests_per_client / 2);
        tests[TEST_IS_TCP][client_idx].count = seq_len;
        tests[TEST_IS_TCP][client_idx].good = 0;
        tests[TEST_IS_TCP][client_idx].bad = 0;
        tests[TEST_IS_TCP][client_idx].plan = malloc(seq_len + 1);
        tests[TEST_IS_TCP][client_idx].results = malloc(seq_len + 1);

        assert(tests[TEST_IS_TCP][client_idx].plan != NULL);
        assert(tests[TEST_IS_TCP][client_idx].results != NULL);

        memset(tests[TEST_IS_TCP][client_idx].results, '?', seq_len);

        tests[TEST_IS_UDP][client_idx].count = seq_len;
        tests[TEST_IS_UDP][client_idx].good = 0;
        tests[TEST_IS_UDP][client_idx].bad = 0;
        tests[TEST_IS_UDP][client_idx].plan = malloc(seq_len + 1);
        tests[TEST_IS_UDP][client_idx].results = malloc(seq_len + 1);

        assert(tests[TEST_IS_UDP][client_idx].plan != NULL);
        assert(tests[TEST_IS_UDP][client_idx].results != NULL);

        memset(tests[TEST_IS_UDP][client_idx].results, '?', seq_len);

        for (test_idx = 0; test_idx < seq_len; ++test_idx)
        {
            c = test_letters[rand() % test_letters_len];

            tests[TEST_IS_TCP][client_idx].plan[test_idx] = c;

            tests[TEST_IS_UDP][client_idx].plan[test_idx] = (c == '-' || c == 'c' ? '+' : c);
        }

        tests[TEST_IS_TCP][client_idx].plan[seq_len] = '\0';
        tests[TEST_IS_TCP][client_idx].results[seq_len] = '\0';
        tests[TEST_IS_UDP][client_idx].plan[seq_len] = '\0';
        tests[TEST_IS_UDP][client_idx].results[seq_len] = '\0';
    }
}

/* ******************************************************** */
int server_callback(struct ap_net_connection_t *conn, int signal_type)
{
    int i, n;
    server_userdata *ud;
    binary_id_packet *id;


    ud = conn->user_data;

    switch (signal_type)
    {
        case AP_NET_SIGNAL_CONN_CREATED:
            if( conn->user_data == NULL )
            {
                conn->user_data = malloc(sizeof(server_userdata));
                if ( conn->user_data == NULL )
                {
                    ap_log_debug_log("* !ERROR: Server oom\n");
                    exit(1);
                }
            }

            ud = conn->user_data;
            ud->is_control = 0;
            ud->test_type = '\0';
            ud->client_id = -1;
            ud->test_index = -1;
            ud->state = -1;

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_DESTROYING:
            free(conn->user_data);

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_CONNECTED:
            /* UDP pool receives this. as it does reverse connect() on incoming data */
            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_ACCEPTED:
            ud->state = STATE_CONNECTED;
            ud->is_control = 0;
            ud->test_type = '\0';
            ud->client_id = -1;
            ud->test_index = -1;

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_CLOSING:
            for ( i = 0; i < max_clients; ++i ) /* finding is this a control channel? */
                if ( conn == control_conns[i] ) /* yep */
                {
                    control_conns[i] = NULL;
                    return 1;
                }

            if( ud->client_id == -1 )
            {
                ap_log_debug_log("* !ERROR: Conn %d close (%s) with unset ID. Buf fill %d, pos: %d\n",
                        conn->idx, ap_error_get_string(), conn->buffill, conn->bufpos);
                return 1;
            }

            /* first trying to recover the last data and check validity */
            if ( ud->test_type == 's' && ud->state != STATE_GOT_MESSAGE && conn->buffill == test_message_len )
                ud->state = STATE_GOT_MESSAGE;

            if ( (ud->test_type == '+' && ud->state != STATE_ANSWER_SENT)
                    || (ud->test_type == '-' && ud->state != STATE_CONNECTED)
                    || (ud->test_type == 'c' && ud->state != STATE_CONNECTED)
                    || (ud->test_type == 's' && ud->state != STATE_GOT_MESSAGE)
                )
            {
                ap_log_debug_log("* !ERROR: Conn %d close (%s) with invalid state '%s'. client %d test #%d '%c'. Buf fill %d, pos: %d\n",
                        conn->idx, ap_error_get_string(), state_text[ud->state], ud->client_id, ud->test_index, ud->test_type, conn->buffill, conn->bufpos);

                tests[POOL_IS_TCP(conn->parent)][ud->client_id].results[ud->test_index] = '-';
                tests[POOL_IS_TCP(conn->parent)][ud->client_id].done++;
                tests[POOL_IS_TCP(conn->parent)][ud->client_id].bad++;

                break;
            }

            tests[POOL_IS_TCP(conn->parent)][ud->client_id].results[ud->test_index] = '+';
            tests[POOL_IS_TCP(conn->parent)][ud->client_id].done++;
            tests[POOL_IS_TCP(conn->parent)][ud->client_id].good++;

            ap_log_debug_log_raw(POOL_IS_TCP(conn->parent) ? "." : "*", 1);

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_MOVED_FROM:
        case AP_NET_SIGNAL_CONN_MOVED_TO:
            ap_log_debug_log("\nMOVED???? Unimplemented.\n");

            exit(1);

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_DATA_IN:
        case AP_NET_SIGNAL_CONN_DATA_LEFT:
            for ( i = 0; i < max_clients; ++i ) /* finding is this a message from control channel? */
            {
                if ( conn == control_conns[i] ) /* yep */
                {
                    /* buffer can contain many IDs, so we looping through them and pushing to the id stack for retrieval after poll call end */
                    for(; (conn->buffill - conn->bufpos) >= sizeof(binary_id_packet); conn->bufpos += sizeof(binary_id_packet))
                    {
                        id = (binary_id_packet *)(conn->buf + conn->bufpos);

                        if( 0 != strncmp((const char *)id->marker, id_packet_marker, strlen(id_packet_marker)) ) /* signature check */
                        {
                            ap_log_debug_log("* !ERROR: non-ID packet on control connection, client #%d/%d\n", ud->client_id, i);
                            ap_log_mem_dump_to_fd(fileno(stdout), id, conn->buffill - conn->bufpos);

                            return 1;
                        }

                        for( i = 0; i < ID_STACK_MAX; ++i )
                            if( id_stack[i].client_index == -1 ) /* pushing to stack and do the next loop on received data */
                            {
                                memcpy(&id_stack[i], id, sizeof(binary_id_packet));

                                break;
                            }

                        if( i == ID_STACK_MAX )
                        {
                            ap_log_debug_log("! ERROR: ID stack is full!\n");

                            exit(1);
                        }
                    }

                    return 1;
                }
            } /* end of control connection message processing */

            /* check if it is new control connection */
            n = strlen(control_conn_marker);

            if ( conn->bufpos == 0 && 0 == strncmp(conn->buf, control_conn_marker, n))
            {
                for ( i = 0; i < max_clients; ++i )
                    if (control_conns[i] == NULL )
                    {
                        control_conns[i] = conn;
                        break;
                    }

                ud->client_id = conn->buf[n] - '0';
                ud->is_control = 1;
                ap_log_debug_log("d#%s/%d\n", conn->buf + n, i);
                ap_utils_timespec_clear(&conn->expire); /* no expiration on control */

                conn->bufpos += n + 2; /* there can be already some ID packets in buffer, so we move position instead of resetting the buffer */

                return 1;
            }

            /* end of control connections section
             * ***********************************************
             * begin of ordinary connection section
             */

            /*ap_log_debug_log("%c", bit)is_set(conn->parent->flags, AP_NET_POOL_FLAGS_TCP) ? 'T' : 'U');*/

            if ( ud->client_id == -1 )
                return 1;

            if ( ! ( ud->state == STATE_GOT_MESSAGE )
                    && ((ud->test_type == '+' && ud->state != STATE_CONNECTED)
                            || ud->test_type == '-'
                            || ud->test_type == 'c'
                            || (ud->test_type == 's' && ud->state != STATE_CONNECTED)
                    ))
            {
                ap_log_debug_log("* !ERROR: Incoming data with invalid state '%s' on client %d test #%d-'%c'\n", state_text[ud->state], client_index, ud->test_index, ud->test_type);

                tests[POOL_IS_TCP(conn->parent)][ud->client_id].results[ud->test_index] = '-';
                tests[POOL_IS_TCP(conn->parent)][ud->client_id].done++;
                tests[POOL_IS_TCP(conn->parent)][ud->client_id].bad++;
            }

            ud->state = STATE_GOT_MESSAGE;

            if ( ud->test_type != '+' )
            {
                conn->buffill = 0; /* discarding data */

                break;
            }

            ap_net_conn_pool_send(conn->parent, conn->idx, (char *)test_message, test_message_len);
            ud->state = STATE_ANSWER_SENT;
            ap_net_conn_pool_close_connection(conn->parent, conn->idx);

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_CAN_SEND:
            /* !NOTE: this is for async operations. unused now */
            if ( ud->test_type != '+' || ud->state != STATE_GOT_MESSAGE )
                break;

            /* for testing purposes we should try here to send async, by storing sent bytes count in user_data */
            ap_net_conn_pool_send(conn->parent, conn->idx, (char *)test_message, test_message_len);
            ud->state = STATE_ANSWER_SENT;
            ap_net_conn_pool_close_connection(conn->parent, conn->idx);

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_TIMED_OUT:
            /*ap_log_debug_log("* got %s conn #%d timeout\n", bit_is_set(conn->parent->flags, AP_NET_POOL_FLAGS_TCP) ? "TCP" : "UDP", conn->idx);*/
            if( (ud->test_type == '+' && ud->state != STATE_ANSWER_SENT && conn->buffill < test_message_len)
                    || (ud->test_type == 's' && ud->state != STATE_GOT_MESSAGE))
            {
                if ( ud->client_id != -1 )
                    ap_log_debug_log("* !ERROR: Client %d: test #%d '%c' timed out in state '%s'. fill: %d\n", ud->client_id, ud->test_index, ud->test_type, state_text[ud->state], conn->buffill);
            }

            break;

        /* ======================================================== */
        default:
            ap_log_debug_log("* !ERROR: Server callback: unknown signal: %d\n", signal_type);

            exit(1);
    }

    return 1;
}

/* ******************************************************** */
int client_callback(struct ap_net_connection_t *conn, int signal_type)
{
    client_userdata *ud;


    ud = conn->user_data;

    switch (signal_type)
    {
        case AP_NET_SIGNAL_CONN_CREATED:
            if( conn->user_data == NULL )
            {
                conn->user_data = malloc(sizeof(client_userdata));

                if ( conn->user_data == NULL )
                {
                    ap_log_debug_log("\t- !ERROR: Client %d oom\n", client_index);
                    exit(1);
                }
            }

            ud = conn->user_data;
            ud->test_type = '\0';
            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_DESTROYING:
            free(conn->user_data);
            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_CONNECTED:
            ud->state = STATE_CONNECTED;
            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_ACCEPTED:
            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_CLOSING:
            return 1;
 
            if ( (! bit_is_set(conn->state, AP_NET_ST_EXPIRED)) && ( /* problems with wrong state also reported on timeout signal, so we don't do it here */
            		(ud->test_type == '+' && ud->state != STATE_GOT_ANSWER)
                    || (ud->test_type == '-' && ud->state != STATE_TEST_ID_SENT)
                    || (ud->test_type == 'c' && ud->state != STATE_TEST_ID_SENT)
                    || (ud->test_type == 's' && ud->state != STATE_MSG_SENT)
                ))
            {
                ap_log_debug_log("\t- !ERROR: Client %d: test #%d '%c' failed in state '%s'\n", client_index, ud->test_index, ud->test_type, state_text[ud->state]);

                exit(1);
            }

            ud->test_type = '\0';

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_MOVED_FROM:
        case AP_NET_SIGNAL_CONN_MOVED_TO:
            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_DATA_IN:
        case AP_NET_SIGNAL_CONN_DATA_LEFT:
            ud = conn->user_data;

            if ( ud->state == STATE_MSG_SENT )
            {
                if(conn->buffill == test_message_len)
                {
                    ud->state = STATE_GOT_ANSWER;
                    ap_net_conn_pool_close_connection(conn->parent, conn->idx);
                }
            }

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_CAN_SEND:
            /* ! NOTE: we do not do async sends now, so this block is not functional */
            if ( ud->state == STATE_TEST_ID_SENT )
            {
                if ( ud->test_type == 'c' ) /* let it expire */
                    return 1;

                if ( test_message_len != ap_net_conn_pool_send(conn->parent, conn->idx, (void *)test_message, test_message_len))
                {
                    ap_log_debug_log("\t- !ERROR: Client %d: test #%d '%c' failed in state '%s': %s\n", client_index, ud->test_index, ud->test_type, state_text[ud->state], ap_error_get_string());
                    exit(1);
                }

            }

            else if ( ud->state == STATE_MSG_SENT )
            {

            }

            else if ( ud->state == STATE_GOT_ANSWER )
            {
                ap_net_conn_pool_close_connection(conn->parent, conn->idx);
            }

            break;

        /* ======================================================== */
        case AP_NET_SIGNAL_CONN_TIMED_OUT:
            /*
            if( ud->test_type == '+' )
                ap_log_debug_log("\t- !ERROR: Client %d: test #%d '%c' timed out in state '%s'\n", client_index, ud->test_index, ud->test_type, state_text[ud->state]);
            */
            break;

        /* ======================================================== */
        default:
            ap_log_debug_log("\t- !ERROR: Client %d unknown signal: %d\n", client_index, signal_type);

            exit(1);
    }

    return 1;
}

/* ******************************************************** */
void go_client(void)
{
    struct ap_net_connection_t *conn, *control_conn;
    char *tests_tcp, *tests_udp;
    int tcp_test_idx, udp_test_idx;
    int tcp_test_count, udp_test_count;
    int n;
    time_t start_time;
    client_userdata *ud;
    binary_id_packet *id_packet;


    tests_tcp = tests[TEST_IS_TCP][client_index].plan;
    tests_udp = tests[TEST_IS_UDP][client_index].plan;
    tcp_test_count = tests[TEST_IS_TCP][client_index].count;
    udp_test_count = tests[TEST_IS_UDP][client_index].count;

    tests_udp = tests[TEST_IS_UDP][client_index].plan;

    /* making tcp pool */
    if ( NULL == (tcp_pool = ap_net_conn_pool_create(AP_NET_POOL_FLAGS_TCP, CLIENT_POOL_SIZE, 1000, 256, client_callback))
            || ! ap_net_conn_pool_poller_create(tcp_pool))
    {
        ap_log_debug_log("\t- !ERROR: Client %d tcp_pool create: %s\n", client_index, ap_error_get_string());

        exit(1);
    }

    /* making udp pool */
    if ( NULL == (udp_pool = ap_net_conn_pool_create(0, CLIENT_POOL_SIZE, 1000, 256, client_callback))
            || ! ap_net_conn_pool_poller_create(udp_pool))
    {
        ap_log_debug_log("\t- !ERROR: Client %d udp_pool create: %s\n", client_index, ap_error_get_string());

        exit(1);
    }

    ap_log_debug_level = CLIENT_DEBUG_LEVEL;

    /* creating connection for debug output */
    if ( NULL == (control_conn = ap_net_conn_pool_connect_straddr(tcp_pool, 0, localhost_str, AF_INET, tcp_port, 0)))
    {
        ap_log_debug_log("\t- !ERROR: Client %d debug conn create: %s\n", client_index, ap_error_get_string());

        exit(1);
    }

    ud = control_conn->user_data;

    n = 1 + sprintf(ud->send_buf, "%s%d", control_conn_marker, client_index);

    if ( n != ap_net_conn_pool_send(tcp_pool, 0, (void *)(ud->send_buf), n) )
    {
        ap_log_debug_log("\t- !ERROR: Client %d debug conn marker send: %s\n", client_index, ap_error_get_string());

        exit(1);
    }

    /*ap_log_debug_log("\t- Client %d debug conn created\n", client_index);*/
    sleep(1);

    tcp_test_idx = udp_test_idx = 0;

#ifndef TEST_UDP
    udp_test_idx = udp_test_count;
#endif

    start_time = time(NULL);

    /* -------------------------------------------------------------------- */
    while ( tests_tcp[tcp_test_idx] != '\0' || tests_udp[udp_test_idx] != '\0' )
    {
        if ( time(NULL) - start_time >= CLIENT_MAX_LIFETIME )
        {
            ap_log_debug_log("\n\n\t!ERROR: Client #%d: Max lifetime exceeded. Tests done: TCP: %d, UDP: %d\n",
                    client_index, tcp_test_count - tcp_test_idx, udp_test_count - udp_test_idx);

            exit(1);
        }

        if ( ! ap_net_conn_pool_poll(tcp_pool) )
        {
            ap_log_debug_log("\t- !ERROR: Client #%d: tcp_pool: %s\n", client_index, ap_error_get_string());

            exit(1);
        }

#ifdef TEST_UDP
        if ( ! ap_net_conn_pool_poll(udp_pool) )
        {
            ap_log_debug_log("\t- !ERROR: Client #%d: udp_pool: %s\n", client_index, ap_error_get_string());

            exit(1);
        }
#endif

        if ( tcp_pool->used_slots == tcp_pool->max_connections
#ifdef TEST_UDP
                && udp_pool->used_slots == udp_pool->max_connections
#endif
                )
        {
            usleep(5000);

            continue;
        }

        if ( tests_tcp[tcp_test_idx] != '\0' && tcp_pool->used_slots < tcp_pool->max_connections ) /* adding new tcp connection */
        {
            if( NULL == (conn = ap_net_conn_pool_connect_straddr(tcp_pool, 0, localhost_str, AF_INET, tcp_port, CONNECTION_TIMEOUT)))
            {
                ap_log_debug_log("\t- !ERROR: Client %d test %d tcp conn create: %s\n", client_index, tcp_test_idx, ap_error_get_string());
                exit(1);
            }

            ud = conn->user_data;
            ud->test_index = tcp_test_idx;
            ud->test_type = tests_tcp[tcp_test_idx];

            id_packet = (binary_id_packet *)(ud->send_buf);
            strcpy(id_packet->marker, id_packet_marker);
            id_packet->is_tcp = 1;
            id_packet->port = conn->local.addr4.sin_port;
            id_packet->test_type = ud->test_type;
            id_packet->test_index = ud->test_index;
            id_packet->client_index = client_index;

            /*if(client_index == 0)
                ap_log_debug_log("\t Client %d test %d tcp port: %d\n", client_index, tcp_test_idx, id_packet->port);
                */

            if ( -1 == ap_net_conn_pool_send(tcp_pool, 0, id_packet, sizeof(binary_id_packet)))
            {
                ap_log_debug_log("\t- !ERROR: Client %d test %d tcp id send: %s\n", client_index, tcp_test_idx, ap_error_get_string());
                exit(1);
            }

            ++tcp_test_idx;
            ud->state = STATE_TEST_ID_SENT;

            if(ud->test_type == '+' || ud->test_type == 's')
            {
                usleep(100000);

                if ( -1 == ap_net_conn_pool_send(conn->parent, conn->idx, (char *)test_message, test_message_len))
                {
                    ap_log_debug_log("\t- !ERROR: Client %d test %d message send: %s\n", client_index, tcp_test_idx, ap_error_get_string());
                    exit(1);
                }

                ud->state = STATE_MSG_SENT;
            }
        } /* if ( tcp_pool->used_slots < tcp_pool->max_connections ) */

#ifdef TEST_UDP
        if ( tests_udp[udp_test_idx] != '\0' && udp_pool->used_slots < udp_pool->max_connections ) /* adding new udp connection */
        {
            conn = ap_net_conn_pool_connect_ip4(udp_pool, 0, INADDR_LOOPBACK, udp_port, tests_udp[udp_test_idx] == 'c' ? 1 : CONNECTION_TIMEOUT);
            if( conn == NULL )
            {
                ap_log_debug_log("\t- !ERROR: Client %d test %d udp conn create: %s\n", client_index, udp_test_idx, ap_error_get_string());
                exit(1);
            }

            ud = conn->user_data;
            ud->test_index = udp_test_idx;
            ud->test_type = tests_udp[udp_test_idx];

            id_packet = (binary_id_packet *)ud->send_buf;
            strcpy(id_packet->marker, id_packet_marker);
            id_packet->is_tcp = 0;
            id_packet->port = conn->local.addr4.sin_port;
            id_packet->test_type = ud->test_type;
            id_packet->test_index = ud->test_index;
            id_packet->client_index = client_index;

            if ( -1 == ap_net_conn_pool_send(tcp_pool, 0, id_packet, sizeof(binary_id_packet)))
            {
                ap_log_debug_log("\t- !ERROR: Client %d test %d udp id send: %s\n", client_index, udp_test_idx, ap_error_get_string());
                exit(1);
            }

            ++udp_test_idx;
            ud->state = STATE_TEST_ID_SENT;

            if(ud->test_type == '+' || ud->test_type == 's')
            {
                usleep(100000);

                if ( -1 == ap_net_conn_pool_send(conn->parent, conn->idx, (char *)test_message, test_message_len))
                {
                    ap_log_debug_log("\t- !ERROR: Client %d test %d message send: %s\n", client_index, udp_test_idx, ap_error_get_string());
                    exit(1);
                }

                ud->state = STATE_MSG_SENT;
            }
        } /* if ( udp_pool->used_slots < udp_pool->max_connections ) */
#endif
    } /* while ( time(NULL) - start_time < CLIENT_MAX_LIFETIME && (tests_tcp[tcp_test_idx] != '\0' || tests_udp[udp_test_idx] != '\0'))  ... */

    ap_log_debug_log("\n\n\t---- Client #%d done\n\n", client_index);

    exit(0);
}