/* -*- Mode: C; c-basic-offset:4 ; indent-tabs-mode:nil ; -*- */
/*
 *  (C) 2010 by Argonne National Laboratory.
 *      See COPYRIGHT in top-level directory.
 */

/*
 * Uncomment this definition to disable the OPA library and instead use naive
 * (non-atomic) operations.  This should cause failures.
 */
/*
 * Note: The NAIVE implementation is likely to cause an infinite loop because of
 * the busy wait in OPA_Queue_dequeue that might never exit if an element was
 * enqueued improperly.
 */
/*
#define OPA_TEST_NAIVE
*/

#include "opa_test.h"
#include "opa_queue.h"

/* Definitions for test_queue_sanity */
typedef struct {
    OPA_Queue_element_hdr_t hdr;        /* Queue element header */
    int         val;                    /* Value of element */
} sanity_t;

/* Definitions for test_queue_threaded */
#define THREADED_NITER 500000
typedef struct {
    OPA_Queue_element_hdr_t hdr;        /* Queue element header */
    int         threadno;               /* Unique thread number */
    int         sequenceno;             /* Unique sequence number within thread */
} threaded_element_t;
typedef enum {
    THREADED_MODE_DEFAULT,              /* Test does not try to influence scheduling */
    THREADED_MODE_EMPTY,                /* Test tries to keep queue empty */
    THREADED_MODE_FULL,                 /* Test tries to keep queue full */
    THREADED_MODE_NMODES                /* Number of modes (must be last) */
} threaded_mode_t;
typedef struct {
    OPA_Queue_info_t *qhead;            /* Head for shared queue */
    int         threadno;               /* Unique thread number */
    threaded_mode_t mode;               /* Test mode (empty or full) */
} threaded_t;


/*-------------------------------------------------------------------------
 * Function: test_queue_sanity
 *
 * Purpose: Tests basic functionality of the OPA Queue in a single
 *          threaded environment.
 *
 * Return: Success: 0
 *         Failure: 1
 *
 * Programmer: Neil Fortner
 *             Wednesday, April 1, 2009
 *
 * Modifications:
 *
 *-------------------------------------------------------------------------
 */
static int test_queue_sanity(void)
{
    OPA_Queue_info_t    qhead;
    sanity_t            elements[4];
    sanity_t            *element;
    int                 i;

    TESTING("queue sanity", 0);

    /* Initialize queue */
    OPA_Queue_init(&qhead);

    /* Make sure the queue is empty */
    if(!OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    if(NULL != OPA_Queue_peek_head(&qhead)) TEST_ERROR;

    /* Initialize elements */
    for(i=0; i<4; i++) {
        OPA_Queue_header_init(&elements[i].hdr);
        elements[i].val = i;
    } /* end for */

    /* Enqueue element 0 */
    OPA_Queue_enqueue(&qhead, &elements[0], sanity_t, hdr);

    /* Peek and make sure we get element 0 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    if(NULL == (element = OPA_Queue_peek_head(&qhead))) TEST_ERROR;
    if(element->val != 0) TEST_ERROR;

    /* Dequeue and make sure we get element 0 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    OPA_Queue_dequeue(&qhead, element, sanity_t, hdr);
    if(element->val != 0) TEST_ERROR;

    /* Make sure the queue is now empty */
    if(!OPA_Queue_is_empty(&qhead)) TEST_ERROR;

    /* Enqueue elements 1 and 2 */
    OPA_Queue_enqueue(&qhead, &elements[1], sanity_t, hdr);
    OPA_Queue_enqueue(&qhead, &elements[2], sanity_t, hdr);

    /* Peek and make sure we get element 1 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    if(NULL == (element = OPA_Queue_peek_head(&qhead))) TEST_ERROR;
    if(element->val != 1) TEST_ERROR;

    /* Dequeue and make sure we get element 1 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    OPA_Queue_dequeue(&qhead, element, sanity_t, hdr);
    if(element->val != 1) TEST_ERROR;

    /* Enqueue element 3 */
    OPA_Queue_enqueue(&qhead, &elements[3], sanity_t, hdr);

    /* Dequeue and make sure we get element 2 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    OPA_Queue_dequeue(&qhead, element, sanity_t, hdr);
    if(element->val != 2) TEST_ERROR;

    /* Peek and make sure we get element 3 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    if(NULL == (element = OPA_Queue_peek_head(&qhead))) TEST_ERROR;
    if(element->val != 3) TEST_ERROR;

    /* Dequeue and make sure we get element 3 */
    if(OPA_Queue_is_empty(&qhead)) TEST_ERROR;
    OPA_Queue_dequeue(&qhead, element, sanity_t, hdr);
    if(element->val != 3) TEST_ERROR;

    /* Make sure the queue is now empty */
    if(!OPA_Queue_is_empty(&qhead)) TEST_ERROR;

    PASSED();
    return 0;

error:
    return 1;
} /* end test_queue_sanity() */


#if defined(OPA_HAVE_PTHREAD_H)
/*-------------------------------------------------------------------------
 * Function: test_queue_threaded_enqueue
 *
 * Purpose: Helper (enqueue) routine for test_queue_threaded.  Enqueues
 *          elements to the shared queue with increasing sequence number
 *          and constant thread number.
 *
 * Return: Success: NULL
 *         Failure: non-NULL
 *
 * Programmer: Neil Fortner
 *             Tuesday, June 8, 2010
 *
 * Modifications:
 *
 *-------------------------------------------------------------------------
 */
static void *test_queue_threaded_enqueue(void *_udata)
{
    threaded_t  *udata = (threaded_t *)_udata;
    threaded_element_t *element = NULL;         /* Element to enqueue */
    int         niter = THREADED_NITER / iter_reduction[curr_test];
    int         i;

    /* Main loop */
    for(i=0; i<niter; i++) {
        /* If we're trying to keep the queue empty, yield here so the dequeuer
         * has a chance to empty the queue */
        if(udata->mode == THREADED_MODE_EMPTY)
            OPA_TEST_YIELD();

        /* Create an element to enqueue */
        if(NULL == (element = (threaded_element_t *) malloc(sizeof(threaded_element_t))))
            pthread_exit((void *) 1);
        OPA_Queue_header_init(&element->hdr);
        element->threadno = udata->threadno;
        element->sequenceno = i;

        /* Enqueue the element */
        OPA_Queue_enqueue(udata->qhead, element, threaded_element_t, hdr);
    } /* end for */

    /* Exit */
    pthread_exit(NULL);
} /* end test_queue_threaded_enqueue() */


/*-------------------------------------------------------------------------
 * Function: test_queue_threaded_dequeue
 *
 * Purpose: Helper (dequeue) routine for test_queue_threaded.  Dequeues
 *          elements from the shared queue and verifies that each thread
 *          added each sequence number exactly once and in the right
 *          order, and that order was preserved by the queue.
 *
 * Return: Success: 0
 *         Failure: 1
 *
 * Programmer: Neil Fortner
 *             Tuesday, June 8, 2010
 *
 * Modifications:
 *
 *-------------------------------------------------------------------------
 */
static int test_queue_threaded_dequeue(void *_udata)
{
    threaded_t  *udata = (threaded_t *)_udata;
    threaded_element_t *element = NULL;         /* Dequeued element */
    threaded_element_t *peek_element = NULL;    /* Peeked element */
    int         *expect_sn = NULL;              /* Expected sequence numbers */
    const int   nenqueuers = num_threads[curr_test] - 1;
    int         ndequeues = 0;  /* Number of times we have dequeued */
    int         peek_counter = 0; /* Iterations since last peek */
    const int   niter = (THREADED_NITER / iter_reduction[curr_test])
                        * nenqueuers;
    int         nerrors = 0;    /* Number of errors */
    int         i;

    /* Allocate expect_sn array */
    if(NULL == (expect_sn = (int *) calloc(nenqueuers, sizeof(int))))
        return(1);

    /* Main loop.  Terminates when either the expected number of elements have
     * been dequeued or the loop has continued too many times. */
    for(i=0; ndequeues < niter && i < (1000 * niter); i++) {
        /* If we're trying to keep the queue full, yield here so the enqueuers
         * have a chance to fill the queue */
        if(udata->mode == THREADED_MODE_FULL)
            OPA_TEST_YIELD();

        /* Check if the queue is nonempty */
        if(!OPA_Queue_is_empty(udata->qhead)) {
            /* Increment peek counter, and peek if the counter is 100 */
            peek_counter++;
            if(peek_counter == 100) {
                if(NULL == (peek_element = OPA_Queue_peek_head(udata->qhead)))
                    nerrors++;
                peek_counter = 0;
                if(OPA_Queue_is_empty(udata->qhead))
                    return(1);
            } /* end if */

            /* Dequeue the head element */
            OPA_Queue_dequeue(udata->qhead, element, threaded_element_t, hdr);

            /* Increment the counter */
            ndequeues++;

            /* Make sure the dequeued element is the same as the peeked element,
             * if applicable */
            if(peek_element) {
                if(peek_element->threadno != element->threadno
                        || peek_element->sequenceno != element->sequenceno) {
                    printf("    Dequeued element does not match peeked element\n");
                    nerrors++;
                } /* end if */
                peek_element = NULL;
            } /* end if */

            /* Check if the thread number is valid */
            if(element->threadno < 0 || element->threadno >= nenqueuers) {
                printf("    Unexpected thread number: %d\n", element->threadno);
                nerrors++;
            } else {
                /* Check if the sequence number is equal to that expected, and
                 * update the expected sequence number */
                if(element->sequenceno != expect_sn[element->threadno]) {
                    printf("    Unexpected sequence number: %d Expected: %d\n",
                            element->sequenceno, expect_sn[element->threadno]);
                    nerrors++;
                } /* end if */
                expect_sn[element->threadno] = element->sequenceno + 1;
            } /* end else */

            /* Free the element */
            free(element);
        } else
            OPA_TEST_YIELD();
    } /* end for */

    /* Check if the expected number of elements were dequeued */
    if(ndequeues != niter) {
        printf("    Incorrect number of elements dequeued: %d Expected: %d\n",
                ndequeues, niter);
        nerrors++;
    } /* end if */

    /* Verify that the queue is now empty */
    if(!OPA_Queue_is_empty(udata->qhead)) {
        printf("    Queue is not empty at end of test!\n");
        nerrors++;
    } /* end if */

    /* Sanity checking that expect_sn is consistent if no errors */
    if(!nerrors)
        for(i=0; i<nenqueuers; i++)
            assert(expect_sn[i] == THREADED_NITER / iter_reduction[curr_test]);

    /* Free expect_sn */
    free(expect_sn);

    /* Any non-NULL exit value indicates an error, we use (void *) 1 here */
    return(nerrors ? 1 : 0);
} /* end test_queue_threaded_dequeue() */
#endif /* OPA_HAVE_PTHREAD_H */


/*-------------------------------------------------------------------------
 * Function: test_queue_threaded
 *
 * Purpose: Tests that the queue behaves correctly in a multithreaded
 *          environment.  Creates many enqueuers and one dequeuer, and
 *          makes sure that elements are dequeued in the expected order.
 *
 * Return: Success: 0
 *         Failure: 1
 *
 * Programmer: Neil Fortner
 *             Wednesday, April 1, 2009
 *
 * Modifications:
 *
 *-------------------------------------------------------------------------
 */
static int test_queue_threaded(threaded_mode_t mode)
{
#if defined(OPA_HAVE_PTHREAD_H)
    pthread_t           *threads = NULL; /* Threads */
    pthread_attr_t      ptattr;         /* Thread attributes */
    static threaded_t   *thread_data = NULL; /* User data structs for threads */
    OPA_Queue_info_t    qhead;          /* The queue */
    void                *ret;           /* Thread return value */
    unsigned            nthreads = num_threads[curr_test];
    int                 nerrors = 0;    /* number of errors */
    int                 i;

    switch(mode) {
        case THREADED_MODE_DEFAULT:
            TESTING("multithreaded queue", nthreads);
            break;
        case THREADED_MODE_EMPTY:
            TESTING("multithreaded queue (empty queue)", nthreads);
            break;
        case THREADED_MODE_FULL:
            TESTING("multithreaded queue (full queue)", nthreads);
            break;
        default:
            FAIL_OP_ERROR(printf("    Unknown mode\n"));
    } /* end switch */

    /* Allocate array of threads */
    if(NULL == (threads = (pthread_t *) malloc(nthreads * sizeof(pthread_t))))
        TEST_ERROR;

    /* Set threads to be joinable */
    pthread_attr_init(&ptattr);
    pthread_attr_setdetachstate(&ptattr, PTHREAD_CREATE_JOINABLE);

    /* Allocate array of thread data */
    if(NULL == (thread_data = (threaded_t *) calloc(nthreads, sizeof(threaded_t))))
        TEST_ERROR;

    /* Initialize queue */
    OPA_Queue_init(&qhead);

    /* Initialize thread data structs */
    for(i=0; i<nthreads; i++) {
        thread_data[i].qhead = &qhead;
        thread_data[i].threadno = i;
        thread_data[i].mode = mode;
    } /* end for */

    /* Create the threads. */
    for(i=0; i<(nthreads - 1); i++)
        if(pthread_create(&threads[i], &ptattr, test_queue_threaded_enqueue,
                &thread_data[i])) TEST_ERROR;
    nerrors = test_queue_threaded_dequeue(&thread_data[i]);

    /* Free the attribute */
    if(pthread_attr_destroy(&ptattr)) TEST_ERROR;

    /* Join the threads */
    for (i=0; i<(nthreads - 1); i++) {
        if(pthread_join(threads[i], &ret)) TEST_ERROR;
        if(ret)
            nerrors++;
    } /* end for */

    /* Check for errors */
    if(nerrors)
        FAIL_OP_ERROR(printf("    Unexpected return from %d thread%s\n", nerrors,
                nerrors == 1 ? "" : "s"));

    /* Free memory */
    free(threads);
    free(thread_data);

    PASSED();

#else /* OPA_HAVE_PTHREAD_H */
    TESTING("multithreaded queue", 0);
    SKIPPED();
    puts("    pthread.h not available");
#endif /* OPA_HAVE_PTHREAD_H */

    return 0;

#if defined(OPA_HAVE_PTHREAD_H)
error:
    if(threads) free(threads);
    if(thread_data) free(thread_data);
    return 1;
#endif /* OPA_HAVE_PTHREAD_H */
} /* end test_barriers_linear_array() */


/*-------------------------------------------------------------------------
 * Function:    main
 *
 * Purpose:     Tests the opa queue
 *
 * Return:      Success:        exit(0)
 *
 *              Failure:        exit(1)
 *
 * Programmer:  Neil Fortner
 *              Tuesday, June 8, 2010
 *
 * Modifications:
 *
 *-------------------------------------------------------------------------
 */
int main(int argc, char **argv)
{
    threaded_mode_t mode;       /* Mode for threaded test */
    unsigned nerrors = 0;
#if defined(OPA_USE_LOCK_BASED_PRIMITIVES)
    OPA_emulation_ipl_t shm_lock;
    OPA_Interprocess_lock_init(&shm_lock, 1/*isLeader*/);
#endif

    /* Initialize shared memory.  Because we are just running threads in one
     * process, the memory will always be symmetric.  Just set the base address
     * to 0. */
    if(OPA_Shm_asymm_init((char *)0) != 0) {
        printf("Unable to initialize shared memory\n");
        goto error;
    } /* end if */

    /* Simple tests */
    nerrors += test_queue_sanity();

    /* Loop over test configurations */
    for(curr_test=0; curr_test<num_thread_tests; curr_test++) {
        /* Don't test with only 1 thread */
        if(num_threads[curr_test] == 1)
            continue;

        /* Threaded tests */
        for(mode = THREADED_MODE_DEFAULT; mode < THREADED_MODE_NMODES; mode++)
            nerrors += test_queue_threaded(mode);
    } /* end for */

    if(nerrors)
        goto error;
    printf("All queue tests passed.\n");

    return 0;

error:
    if(!nerrors)
        nerrors = 1;
    printf("***** %d QUEUE TEST%s FAILED! *****\n",
            nerrors, 1 == nerrors ? "" : "S");
    return 1;
} /* end main() */