pmix/test/cli_stages.c

/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil -*- */
/*
 * Copyright (c) 2015-2019 Intel, Inc.  All rights reserved.
 * Copyright (c) 2015      Research Organization for Information Science
 *                         and Technology (RIST). All rights reserved.
 * Copyright (c) 2015-2019 Mellanox Technologies, Inc.
 *                         All rights reserved.
 * $COPYRIGHT$
 *
 * Additional copyrights may follow
 *
 * $HEADER$
 */

#include "cli_stages.h"

cli_info_t *cli_info = NULL;
int cli_info_cnt = 0;
bool test_abort = false;

int cli_rank(cli_info_t *cli)
{
    int i;
    for(i=0; i < cli_info_cnt; i++){
        if( cli == &cli_info[i] ){
            return cli->rank;
        }
    }
    return -1;
}

void cli_init(int nprocs)
{
    int n, i;
    cli_state_t order[CLI_TERM+1];

    cli_info = malloc( sizeof(cli_info_t) * nprocs);
    cli_info_cnt = nprocs;

    order[CLI_UNINIT] = CLI_FORKED;
    order[CLI_FORKED] = CLI_FIN;
    order[CLI_CONNECTED] = CLI_UNDEF;
    order[CLI_FIN] = CLI_TERM;
    order[CLI_DISCONN] = CLI_UNDEF;
    order[CLI_TERM] = CLI_UNDEF;

    for (n=0; n < nprocs; n++) {
        cli_info[n].sd = -1;
        cli_info[n].ev = NULL;
        cli_info[n].pid = -1;
        cli_info[n].state = CLI_UNINIT;
        PMIX_CONSTRUCT(&(cli_info[n].modex), pmix_list_t);
        for (i = 0; i < CLI_TERM+1; i++) {
            cli_info[n].next_state[i] = order[i];
        }
        cli_info[n].rank = -1;
        cli_info[n].ns = NULL;
    }
}

void cli_connect(cli_info_t *cli, int sd, struct event_base * ebase, event_callback_fn callback)
{
    if( CLI_CONNECTED != cli->next_state[cli->state] ){
        TEST_ERROR(("Rank %d has bad next state: expect %d have %d!",
                     cli_rank(cli), CLI_CONNECTED, cli->next_state[cli->state]));
        test_abort = true;
        return;
    }

    cli->sd = sd;
    cli->ev = event_new(ebase, sd,
                      EV_READ|EV_PERSIST, callback, cli);
    event_add(cli->ev,NULL);
    pmix_ptl_base_set_nonblocking(sd);
    TEST_VERBOSE(("Connection accepted from rank %d", cli_rank(cli) ));
    cli->state = CLI_CONNECTED;
}

void cli_finalize(cli_info_t *cli)
{
    if( CLI_FIN != cli->next_state[cli->state] ){
        TEST_ERROR(("rank %d: bad client next state: expect %d have %d!",
                     cli_rank(cli), CLI_FIN, cli->next_state[cli->state]));
        test_abort = true;
    }

    cli->state = CLI_FIN;
}

void cli_disconnect(cli_info_t *cli)
{
    if( CLI_DISCONN != cli->next_state[cli->state] ){
        TEST_ERROR(("rank %d: bad client next state: expect %d have %d!",
                     cli_rank(cli), CLI_DISCONN, cli->next_state[cli->state]));
        test_abort = true;
    }

    if( 0 > cli->sd ){
        TEST_ERROR(("Bad sd = %d of rank = %d ", cli->sd, cli_rank(cli)));
        test_abort = true;
    } else {
        TEST_VERBOSE(("close sd = %d for rank = %d", cli->sd, cli_rank(cli)));
        close(cli->sd);
        cli->sd = -1;
    }

    if( NULL == cli->ev ){
        TEST_ERROR(("Bad ev = NULL of rank = %d ", cli_rank(cli)));
        test_abort = true;
    } else {
        TEST_VERBOSE(("remove event of rank %d from event queue", cli_rank(cli)));
        event_del(cli->ev);
        event_free(cli->ev);
        cli->ev = NULL;
    }

    TEST_VERBOSE(("Destruct modex list for the rank %d", cli_rank(cli)));
    PMIX_LIST_DESTRUCT(&(cli->modex));

    cli->state = CLI_DISCONN;
}

void cli_terminate(cli_info_t *cli)
{
    if( CLI_TERM != cli->next_state[cli->state] ){
        TEST_ERROR(("rank %d: bad client next state: expect %d have %d!",
                     cli_rank(cli), CLI_TERM, cli->next_state[cli->state]));
        test_abort = true;
    }
    cli->pid = -1;
    TEST_VERBOSE(("Client rank = %d terminated", cli_rank(cli)));
    cli->state = CLI_TERM;
    if (NULL != cli->ns) {
        free(cli->ns);
    }
}

void cli_cleanup(cli_info_t *cli)
{
    if (CLI_TERM < cli->state) {
        TEST_ERROR(("Bad rank %d state %d", cli_rank(cli), cli->state));
        test_abort = true;
        return;
    }
    switch( cli->next_state[cli->state] ){
    case CLI_FORKED:
        break;
    case CLI_CONNECTED:
        /* error - means that process terminated w/o calling finalize */
        if (!test_abort) {
            TEST_ERROR(("rank %d with state %d unexpectedly terminated.", cli_rank(cli), cli->state));
        }
        cli->state = CLI_TERM;
        test_abort = true;
        break;
    case CLI_FIN:
        /* error - means that process terminated w/o calling finalize */
        if (!test_abort) {
            TEST_ERROR(("rank %d with state %d unexpectedly terminated.", cli_rank(cli), cli->state));
        }
        cli_finalize(cli);
        cli_cleanup(cli);
        test_abort = true;
        break;
    case CLI_DISCONN:
        cli_disconnect(cli);
        cli_cleanup(cli);
        break;
    case CLI_TERM:
        cli_terminate(cli);
        break;
    default:
        TEST_ERROR(("Bad rank %d next state %d", cli_rank(cli), cli->next_state[cli->state]));
        test_abort = true;
        return;
    }
}


bool test_terminated(void)
{
    bool ret = true;
    int i;

    // All clients should disconnect
    for(i=0; i < cli_info_cnt; i++){
        ret = ret && (CLI_TERM <= cli_info[i].state);
    }
    return (ret || test_abort);
}

void cli_wait_all(double timeout)
{
    struct timeval tv;
    double start_time, cur_time;

    gettimeofday(&tv, NULL);
    start_time = tv.tv_sec + 1E-6*tv.tv_usec;
    cur_time = start_time;

    //TEST_VERBOSE(("Wait for all children to terminate"))

    // Wait for all children to cleanup after the test.
    while( !test_terminated() && ( timeout >= (cur_time - start_time) ) ){
        struct timespec ts;
        int status, i;
        pid_t pid;
        while( 0 < (pid = waitpid(-1, &status, WNOHANG) ) ){
            TEST_VERBOSE(("waitpid = %d", pid));
            for(i=0; i < cli_info_cnt; i++){
                if( cli_info[i].pid == pid ){
                    TEST_VERBOSE(("the child with pid = %d has rank = %d, ns = %s\n"
                                "\t\texited = %d, signalled = %d", pid,
                                  cli_info[i].rank, cli_info[i].ns,
                                WIFEXITED(status), WIFSIGNALED(status) ));
                    if( WIFEXITED(status) || WIFSIGNALED(status) ){
                        cli_cleanup(&cli_info[i]);
                    }
                }
            }
        }
        if( pid < 0 ){
            if( errno == ECHILD ){
                TEST_VERBOSE(("No more children to wait. Happens on the last cli_wait_all call "
                            "which is used to ensure that all children terminated.\n"));
                if (pmix_test_verbose) {
                    sleep(1);
                }
                break;
            } else {
                TEST_ERROR(("waitpid(): %d : %s", errno, strerror(errno)));
                exit(0);
            }
        }
        ts.tv_sec = 0;
        ts.tv_nsec = 100000;
        nanosleep(&ts, NULL);
        // calculate current timestamp
        gettimeofday(&tv, NULL);
        cur_time = tv.tv_sec + 1E-6*tv.tv_usec;
    }
}

void cli_kill_all(void)
{
    int i;
    for(i = 0; i < cli_info_cnt; i++){
        if( CLI_UNINIT == cli_info[i].state ){
            TEST_ERROR(("Skip rank %d as it wasn't ever initialized (shouldn't happe)",
                          i));
            continue;
        } else if( CLI_TERM <= cli_info[i].state ){
            TEST_VERBOSE(("Skip rank %d as it was already terminated.", i));
            continue;

        }
        TEST_VERBOSE(("Kill rank %d (pid = %d).", i, cli_info[i].pid));
        kill(cli_info[i].pid, SIGKILL);
        cli_cleanup(&cli_info[i]);
    }
}

void errhandler(size_t evhdlr_registration_id,
                pmix_status_t status,
                const pmix_proc_t *source,
                pmix_info_t info[], size_t ninfo,
                pmix_info_t results[], size_t nresults,
                pmix_event_notification_cbfunc_fn_t cbfunc,
                void *cbdata)
{
    TEST_ERROR((" PMIX server event handler with status = %d", status));
    /* notify clients of error */
    PMIx_Notify_event(status, source,
                      PMIX_RANGE_NAMESPACE,
                      NULL, 0,
                      op_callbk, NULL);
}

void op_callbk(pmix_status_t status,
                      void *cbdata)
{
    TEST_VERBOSE(( "OP CALLBACK CALLED WITH STATUS %d", status));
}

void errhandler_reg_callbk (pmix_status_t status,
                            size_t errhandler_ref,
                            void *cbdata)
{
    TEST_VERBOSE(("ERRHANDLER REGISTRATION CALLBACK CALLED WITH STATUS %d, ref=%lu",
                status, (unsigned long)errhandler_ref));
}