/* -*- 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

        /* 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

            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

        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

        /* 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() */