/*
 * CDDL HEADER START
 *
 * The contents of this file are subject to the terms of the
 * Common Development and Distribution License (the "License").
 * You may not use this file except in compliance with the License.
 *
 * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
 * or http://www.opensolaris.org/os/licensing.
 * See the License for the specific language governing permissions
 * and limitations under the License.
 *
 * When distributing Covered Code, include this CDDL HEADER in each
 * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
 * If applicable, add the following below this CDDL HEADER, with the
 * fields enclosed by brackets "[]" replaced with your own identifying
 * information: Portions Copyright [yyyy] [name of copyright owner]
 *
 * CDDL HEADER END
 */

/*
 * Copyright (c) 2002-2003, Network Appliance, Inc. All rights reserved.
 */

/*
 * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
 * Use is subject to license terms.
 */

/*
 *
 * MODULE: dapl_evd_util.c
 *
 * PURPOSE: Manage EVD Info structure
 *
 * $Id: dapl_evd_util.c,v 1.41 2003/08/20 13:18:36 sjs2 Exp $
 */

#include <sys/time.h>
#include <strings.h>
#include "dapl_evd_util.h"
#include "dapl_ia_util.h"
#include "dapl_cno_util.h"
#include "dapl_ring_buffer_util.h"
#include "dapl_adapter_util.h"
#include "dapl_tavor_ibtf_impl.h"
#include "dapl_cookie.h"
#include "dapl.h"


#ifdef  DAPL_DBG        /* For debugging.  */
static void
dapli_evd_eh_print_cqe(
        IN  ib_work_completion_t        cqe);
#endif

static DAT_BOOLEAN
dapli_evd_cqe_to_event(
    IN DAPL_EVD                 *evd_ptr,
    IN ib_work_completion_t     *cqe_ptr,
    IN DAT_BOOLEAN              process_premature_events,
    OUT DAT_EVENT               *event_ptr);

static DAT_RETURN
dapli_evd_event_alloc(
        IN  DAPL_EVD            *evd_ptr,
        IN  DAPL_CNO            *cno_ptr,
        IN  DAT_COUNT           qlen);


/*
 * dapls_evd_internal_create
 *
 * actually create the evd.  this is called after all parameter checking
 * has been performed in dapl_ep_create.  it is also called from dapl_ia_open
 * to create the default async evd.
 *
 * Input:
 *      ia_ptr
 *      cno_ptr
 *      qlen
 *      evd_flags
 *
 * Output:
 *      evd_ptr_ptr
 *
 * Returns:
 *      none
 *
 */

DAT_RETURN
dapls_evd_internal_create(
    DAPL_IA             *ia_ptr,
    DAPL_CNO            *cno_ptr,
    DAT_COUNT           min_qlen,
    DAT_EVD_FLAGS       evd_flags,
    DAPL_EVD            **evd_ptr_ptr)
{
        DAPL_EVD        *evd_ptr;
        DAT_COUNT       cq_len;
        DAT_RETURN      dat_status;

        dat_status      = DAT_SUCCESS;
        *evd_ptr_ptr    = NULL;
        cq_len          = min_qlen;

        evd_ptr = dapls_evd_alloc(ia_ptr,
            cno_ptr,
            evd_flags,
            min_qlen);
        if (!evd_ptr) {
                dat_status = DAT_ERROR(DAT_INSUFFICIENT_RESOURCES,
                    DAT_RESOURCE_MEMORY);
                goto bail;
        }

        /*
         * If we are dealing with event streams besides a CQ event stream,
         * be conservative and set producer side locking.  Otherwise, no.
         */
        evd_ptr->evd_producer_locking_needed =
            ((evd_flags & ~ (DAT_EVD_DTO_FLAG|DAT_EVD_RMR_BIND_FLAG)) != 0);

        /* Before we setup any callbacks, transition state to OPEN.  */
        evd_ptr->evd_state = DAPL_EVD_STATE_OPEN;

        /*
         * we need to call cq_alloc even for connection/cr/async evds
         * since all the allocation happens there.
         */
        dat_status = dapls_ib_cq_alloc(ia_ptr,
            evd_ptr, cno_ptr, &cq_len);
        if (dat_status != DAT_SUCCESS) {
                goto bail;
        }

#if 0
        /*
         * Current implementation of dapls_ib_setup_async_callback() does
         * nothing and returns DAT_SUCCESS. However, it is declared to expect
         * function pointers with different signatures. We do leave the code
         * block out till dapls_ib_setup_async_callback() is implemented.
         */
        dat_status = dapls_ib_setup_async_callback(
            ia_ptr,
            DAPL_ASYNC_CQ_COMPLETION,
            (unsigned int *) evd_ptr->ib_cq_handle,
            (ib_async_handler_t)dapl_evd_dto_callback,
            evd_ptr);
        if (dat_status != DAT_SUCCESS) {
                goto bail;
        }
#endif
        /*
         * cq_notify is not required since when evd_wait is called
         * time we go and poll cq anyways.
         * dat_status = dapls_set_cq_notify(ia_ptr, evd_ptr);
         */

        /*
         * We now have an accurate count of events, so allocate them into
         * the EVD
         */
        dat_status = dapli_evd_event_alloc(evd_ptr, cno_ptr, cq_len);
        if (dat_status != DAT_SUCCESS) {
                goto bail;
        }

        /* We're assuming success in the following.   */
        dapl_os_assert(dat_status == DAT_SUCCESS);
        dapl_ia_link_evd(ia_ptr, evd_ptr);
        *evd_ptr_ptr = evd_ptr;

bail:
        if (dat_status != DAT_SUCCESS) {
                if (evd_ptr) {
                        (void) dapls_evd_dealloc(evd_ptr);
                }
        }

        return (dat_status);
}

/*
 * dapls_evd_alloc
 *
 * alloc and initialize an EVD struct
 *
 * Input:
 *      ia
 *
 * Output:
 *      evd_ptr
 *
 * Returns:
 *      none
 *
 */
DAPL_EVD *
dapls_evd_alloc(
    IN DAPL_IA          *ia_ptr,
    IN DAPL_CNO         *cno_ptr,
    IN DAT_EVD_FLAGS    evd_flags,
    IN DAT_COUNT        qlen) /* ARGSUSED */
{
        DAPL_EVD        *evd_ptr;

        evd_ptr    = NULL;

        /* Allocate EVD */
        evd_ptr = (DAPL_EVD *)dapl_os_alloc(sizeof (DAPL_EVD));
        if (!evd_ptr) {
                goto bail;
        }

        /* zero the structure */
        (void) dapl_os_memzero(evd_ptr, sizeof (DAPL_EVD));

        /*
         * initialize the header
         */
        evd_ptr->header.provider                = ia_ptr->header.provider;
        evd_ptr->header.magic                   = DAPL_MAGIC_EVD;
        evd_ptr->header.handle_type             = DAT_HANDLE_TYPE_EVD;
        evd_ptr->header.owner_ia                = ia_ptr;
        evd_ptr->header.user_context.as_64      = 0;
        evd_ptr->header.user_context.as_ptr     = NULL;
        dapl_llist_init_entry(&evd_ptr->header.ia_list_entry);
        dapl_os_lock_init(&evd_ptr->header.lock);

        /*
         * Initialize the body
         */
        evd_ptr->evd_state      = DAPL_EVD_STATE_INITIAL;
        evd_ptr->evd_flags      = evd_flags;
        evd_ptr->evd_enabled    = DAT_TRUE;
        evd_ptr->evd_waitable   = DAT_TRUE;
        evd_ptr->evd_producer_locking_needed = 1; /* Conservative value.  */
        evd_ptr->ib_cq_handle   = IB_INVALID_HANDLE;
        evd_ptr->evd_ref_count  = 0;
        evd_ptr->catastrophic_overflow = DAT_FALSE;
        evd_ptr->qlen           = qlen;

        dapl_llist_init_entry(&evd_ptr->cno_list_entry);
        evd_ptr->completion_type = DAPL_EVD_STATE_THRESHOLD;
        (void) dapl_os_wait_object_init(&evd_ptr->wait_object);

bail:
        return (evd_ptr);
}


/*
 * dapls_evd_event_alloc
 *
 * alloc events into an EVD.
 *
 * Input:
 *      evd_ptr
 *      qlen
 *
 * Output:
 *      NONE
 *
 * Returns:
 *      DAT_SUCCESS
 *      ERROR
 *
 */
DAT_RETURN
dapli_evd_event_alloc(
    IN DAPL_EVD         *evd_ptr,
    IN  DAPL_CNO        *cno_ptr,
    IN DAT_COUNT        qlen)
{
        DAT_EVENT       *event_ptr;
        DAT_COUNT       i;
        DAT_RETURN      dat_status;

        dat_status = DAT_SUCCESS;
        event_ptr  = NULL;

        /* Allocate EVENTs */
        event_ptr = (DAT_EVENT *) dapl_os_alloc(qlen * sizeof (DAT_EVENT));
        if (!event_ptr) {
                goto bail;
        }
        evd_ptr->events = event_ptr;
        evd_ptr->qlen = qlen;

        /* allocate free event queue */
        dat_status = dapls_rbuf_alloc(&evd_ptr->free_event_queue, qlen);
        if (dat_status != DAT_SUCCESS) {
                goto bail;
        }

        /* allocate pending event queue */
        dat_status = dapls_rbuf_alloc(&evd_ptr->pending_event_queue, qlen);
        if (dat_status != DAT_SUCCESS) {
                goto bail;
        }

        /* add events to free event queue */
        for (i = 0; i < qlen; i++) {
                dat_status = dapls_rbuf_add(&evd_ptr->free_event_queue,
                    (void *)event_ptr);
                dapl_os_assert(dat_status == DAT_SUCCESS);
                event_ptr++;
        }
        evd_ptr->cq_notified = DAT_FALSE;
        evd_ptr->cq_notified_when = 0;
        evd_ptr->cno_active_count = 0;
        if (cno_ptr != NULL) {
                dapl_os_lock(&cno_ptr->header.lock);
                dapl_llist_add_head(&cno_ptr->evd_list_head,
                    &evd_ptr->cno_list_entry, evd_ptr);
                /* Take a reference count on the CNO */
                dapl_os_atomic_inc(&cno_ptr->cno_ref_count);
                dapl_os_unlock(&cno_ptr->header.lock);
        }
        evd_ptr->cno_ptr = cno_ptr;
        evd_ptr->threshold = 0;

bail:
        return (dat_status);
}


/*
 * dapls_evd_dealloc
 *
 * Free the passed in EVD structure. If an error occurs, this function
 * will clean up all of the internal data structures and report the
 * error.
 *
 * Input:
 *      evd_ptr
 *
 * Output:
 *      none
 *
 * Returns:
 *      status
 *
 */
DAT_RETURN
dapls_evd_dealloc(
    IN DAPL_EVD         *evd_ptr)
{
        DAT_RETURN      dat_status;
        DAPL_IA *ia_ptr;

        dat_status = DAT_SUCCESS;

        dapl_os_assert(evd_ptr->header.magic == DAPL_MAGIC_EVD);
        dapl_os_assert(evd_ptr->evd_ref_count == 0);

        /*
         * Destroy the CQ first, to keep any more callbacks from coming
         * up from it.
         */
        if (evd_ptr->ib_cq_handle != IB_INVALID_HANDLE) {
                ia_ptr = evd_ptr->header.owner_ia;

                dat_status = dapls_ib_cq_free(ia_ptr, evd_ptr);
                if (dat_status != DAT_SUCCESS) {
                        goto bail;
                }
        }

        /*
         * We should now be safe to invalidate the EVD; reset the
         * magic to prevent reuse.
         */
        evd_ptr->header.magic = DAPL_MAGIC_INVALID;

        /* Release reference on the CNO if it exists */
        if (evd_ptr->cno_ptr != NULL) {
                dapl_os_lock(&evd_ptr->cno_ptr->header.lock);
                (void) dapl_llist_remove_entry(&evd_ptr->cno_ptr->evd_list_head,
                    &evd_ptr->cno_list_entry);
                dapl_os_atomic_dec(&evd_ptr->cno_ptr->cno_ref_count);
                dapl_os_unlock(&evd_ptr->cno_ptr->header.lock);
        }

        /*
         * If the ring buffer allocation failed, then the dapls_rbuf_destroy
         * function will detect that the ring buffer's internal data (ex. base
         * pointer) are invalid and will handle the situation appropriately
         */
        dapls_rbuf_destroy(&evd_ptr->free_event_queue);
        dapls_rbuf_destroy(&evd_ptr->pending_event_queue);

        if (evd_ptr->events) {
                dapl_os_free(evd_ptr->events,
                    evd_ptr->qlen * sizeof (DAT_EVENT));
        }

        (void) dapl_os_wait_object_destroy(&evd_ptr->wait_object);
        dapl_os_free(evd_ptr, sizeof (DAPL_EVD));

bail:
        return (dat_status);
}


/*
 * dapli_evd_eh_print_cqe
 *
 * Input:
 *      cqe
 *
 * Output:
 *      none
 *
 * Prints out a CQE for debug purposes
 *
 */

#ifdef  DAPL_DBG        /* For debugging.  */
void
dapli_evd_eh_print_cqe(IN ib_work_completion_t cqe)
{
        static char *optable[] = {
                "",
                "OP_SEND",
                "OP_RDMA_READ",
                "OP_RDMA_WRITE",
                "OP_COMP_AND_SWAP",
                "OP_FETCH_AND_ADD",
                "OP_BIND_MW",
                "OP_RECEIVE",
                "OP_RECEIVE_RDMAWI",
                0
        };
        DAPL_COOKIE             *dto_cookie;

        dto_cookie = (DAPL_COOKIE *) (uintptr_t)DAPL_GET_CQE_WRID(&cqe);

        dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
            "\t >>>>>>>>>>>>>>>>>>>>>>><<<<<<<<<<<<<<<<<<<\n");
        dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
            "\t dapl_evd_dto_callback : CQE \n");
        dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
            "\t\t work_req_id 0x%llx\n", DAPL_GET_CQE_WRID(&cqe));
        dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
            "\t\t op_type: %s\n", optable[DAPL_GET_CQE_OPTYPE(&cqe)]);
        if ((DAPL_GET_CQE_OPTYPE(&cqe) == OP_SEND) ||
            (DAPL_GET_CQE_OPTYPE(&cqe) == OP_RDMA_WRITE)) {
                dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
                    "\t\t bytes_num %d\n", dto_cookie->val.dto.size);
        } else {
                dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
                    "\t\t bytes_num %d\n", DAPL_GET_CQE_BYTESNUM(&cqe));
        }
        dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
            "\t\t status %d\n", DAPL_GET_CQE_STATUS(&cqe));
        dapl_dbg_log(DAPL_DBG_TYPE_CALLBACK,
            "\t >>>>>>>>>>>>>>>>>>>>>>><<<<<<<<<<<<<<<<<<<\n");
}
#endif

/*
 * Event posting code follows.
 */

/*
 * These next two functions (dapli_evd_get_event and dapli_evd_post_event)
 * are a pair.  They are always called together, from one of the functions
 * at the end of this file (dapl_evd_post_*_event).
 *
 * Note that if producer side locking is enabled, the first one takes the
 * EVD lock and the second releases it.
 */

/*
 * dapli_evd_get_event
 *
 * Get an event struct from the evd.  The caller should fill in the event
 * and call dapl_evd_post_event.
 *
 * If there are no events available, an overflow event is generated to the
 * async EVD handler.
 *
 * If this EVD required producer locking, a successful return implies
 * that the lock is held.
 *
 * Input:
 *      evd_ptr
 *
 * Output:
 *      event
 *
 */

static DAT_EVENT *
dapli_evd_get_event(
    DAPL_EVD *evd_ptr)
{
        DAT_EVENT       *event;

        if (evd_ptr->evd_producer_locking_needed) {
                dapl_os_lock(&evd_ptr->header.lock);
        }

        event = (DAT_EVENT *)dapls_rbuf_remove(&evd_ptr->free_event_queue);

        /* Release the lock if it was taken and the call failed.  */
        if (!event && evd_ptr->evd_producer_locking_needed) {
                dapl_os_unlock(&evd_ptr->header.lock);
        }

        return (event);
}

/*
 * dapli_evd_post_event
 *
 * Post the <event> to the evd.  If possible, invoke the evd's CNO.
 * Otherwise post the event on the pending queue.
 *
 * If producer side locking is required, the EVD lock must be held upon
 * entry to this function.
 *
 * Input:
 *      evd_ptr
 *      event
 *
 * Output:
 *      none
 *
 */

static void
dapli_evd_post_event(
    IN  DAPL_EVD        *evd_ptr,
    IN  const DAT_EVENT *event_ptr)
{
        DAT_RETURN      dat_status;
        DAPL_CNO        *cno_to_trigger = NULL;

        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
            "dapli_evd_post_event: Called with event # %x\n",
            event_ptr->event_number);

        dat_status = dapls_rbuf_add(&evd_ptr->pending_event_queue,
            (void *)event_ptr);
        dapl_os_assert(dat_status == DAT_SUCCESS);

        dapl_os_assert(evd_ptr->evd_state == DAPL_EVD_STATE_WAITED ||
            evd_ptr->evd_state == DAPL_EVD_STATE_OPEN);

        if (evd_ptr->evd_state == DAPL_EVD_STATE_OPEN) {
                /* No waiter.  Arrange to trigger a CNO if it exists.  */

                if (evd_ptr->evd_enabled) {
                        cno_to_trigger = evd_ptr->cno_ptr;
                }
                if (evd_ptr->evd_producer_locking_needed) {
                        dapl_os_unlock(&evd_ptr->header.lock);
                }
        } else {
                /*
                 * This routine gets called
                 *  - In the context of the waiting thread when CQ, CM or ASYNC
                 *    events need to be put on to the EVD ring buffer.
                 *  - Due to a post of a software event.
                 *
                 * In the first case the waiting thread is pulling the events
                 * from various streams into the evd so there is no need to
                 * wake any thread. In the second case if the evd is in waited
                 * state then we need to wakeup the waiting thread.
                 */
                if (event_ptr->event_number == DAT_SOFTWARE_EVENT) {
                        /*
                         * We're in DAPL_EVD_STATE_WAITED.  Take the lock if
                         * we don't have it, recheck, and signal.
                         */

                        if (!evd_ptr->evd_producer_locking_needed) {
                                dapl_os_lock(&evd_ptr->header.lock);
                        }

                        if (evd_ptr->evd_state == DAPL_EVD_STATE_WAITED) {
                                dapl_os_unlock(&evd_ptr->header.lock);
                                (void) dapls_ib_event_wakeup(evd_ptr);
                        } else {
                                dapl_os_unlock(&evd_ptr->header.lock);
                        }
                } else {
                        if (evd_ptr->evd_producer_locking_needed) {
                                dapl_os_unlock(&evd_ptr->header.lock);
                        }
                }
        }

        if (cno_to_trigger != NULL) {
                dapl_cno_trigger(cno_to_trigger, evd_ptr);
        }
}

/*
 * dapli_evd_post_event_nosignal
 *
 * Post the <event> to the evd.  Do not do any wakeup processing.
 * This function should only be called if it is known that there are
 * no waiters that it is appropriate to wakeup on this EVD.  An example
 * of such a situation is during internal dat_evd_wait() processing.
 *
 * If producer side locking is required, the EVD lock must be held upon
 * entry to this function.
 *
 * Input:
 *      evd_ptr
 *      event
 *
 * Output:
 *      none
 *
 */

static void
dapli_evd_post_event_nosignal(
    IN  DAPL_EVD        *evd_ptr,
    IN  const DAT_EVENT *event_ptr)
{
        DAT_RETURN      dat_status;

        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
            "dapli_evd_post_event: Called with event # %x\n",
            event_ptr->event_number);

        dat_status = dapls_rbuf_add(&evd_ptr->pending_event_queue,
            (void *)event_ptr);
        dapl_os_assert(dat_status == DAT_SUCCESS);

        dapl_os_assert(evd_ptr->evd_state == DAPL_EVD_STATE_WAITED ||
            evd_ptr->evd_state == DAPL_EVD_STATE_OPEN);

        if (evd_ptr->evd_producer_locking_needed) {
                dapl_os_unlock(&evd_ptr->header.lock);
        }
}

/*
 * dapli_evd_format_overflow_event
 *
 * format an overflow event for posting
 *
 * Input:
 *      evd_ptr
 *      event_ptr
 *
 * Output:
 *      none
 *
 */
static void
dapli_evd_format_overflow_event(
        IN  DAPL_EVD  *evd_ptr,
        OUT DAT_EVENT *event_ptr)
{
        DAPL_IA *ia_ptr;

        ia_ptr = evd_ptr->header.owner_ia;

        event_ptr->evd_handle   = (DAT_EVD_HANDLE)evd_ptr;
        event_ptr->event_number = DAT_ASYNC_ERROR_EVD_OVERFLOW;
        event_ptr->event_data.asynch_error_event_data.dat_handle =
            (DAT_HANDLE)ia_ptr;
}

/*
 * dapli_evd_post_overflow_event
 *
 * post an overflow event
 *
 * Input:
 *      async_evd_ptr
 *      evd_ptr
 *
 * Output:
 *      none
 *
 */
static void
dapli_evd_post_overflow_event(
    IN  DAPL_EVD  *async_evd_ptr,
    IN  DAPL_EVD  *overflow_evd_ptr)
{
        DAT_EVENT *overflow_event;

        /*
         * The overflow_evd_ptr mght be the same as evd.
         * In that case we've got a catastrophic overflow.
         */
        if (async_evd_ptr == overflow_evd_ptr) {
                async_evd_ptr->catastrophic_overflow = DAT_TRUE;
                async_evd_ptr->evd_state = DAPL_EVD_STATE_DEAD;
                return;
        }

        overflow_event = dapli_evd_get_event(overflow_evd_ptr);
        if (!overflow_event) {
                /* this is not good */
                overflow_evd_ptr->catastrophic_overflow = DAT_TRUE;
                overflow_evd_ptr->evd_state = DAPL_EVD_STATE_DEAD;
                return;
        }
        dapli_evd_format_overflow_event(overflow_evd_ptr, overflow_event);
        dapli_evd_post_event(overflow_evd_ptr, overflow_event);
}

static DAT_EVENT *
dapli_evd_get_and_init_event(
    IN DAPL_EVD                         *evd_ptr,
    IN DAT_EVENT_NUMBER                 event_number)
{
        DAT_EVENT               *event_ptr;

        event_ptr = dapli_evd_get_event(evd_ptr);
        if (NULL == event_ptr) {
                dapli_evd_post_overflow_event(
                    evd_ptr->header.owner_ia->async_error_evd, evd_ptr);
        } else {
                event_ptr->evd_handle = (DAT_EVD_HANDLE) evd_ptr;
                event_ptr->event_number = event_number;
        }

        return (event_ptr);
}

DAT_RETURN
dapls_evd_post_cr_arrival_event(
    IN DAPL_EVD                         *evd_ptr,
    IN DAT_EVENT_NUMBER                 event_number,
    IN DAT_SP_HANDLE                    sp_handle,
    DAT_IA_ADDRESS_PTR                  ia_address_ptr,
    DAT_CONN_QUAL                       conn_qual,
    DAT_CR_HANDLE                       cr_handle)
{
        DAT_EVENT               *event_ptr;
        event_ptr = dapli_evd_get_and_init_event(evd_ptr, event_number);
        /*
         * Note event lock may be held on successful return
         * to be released by dapli_evd_post_event(), if provider side locking
         * is needed.
         */

        if (!event_ptr) {
                return (DAT_INSUFFICIENT_RESOURCES | DAT_RESOURCE_MEMORY);
        }

        event_ptr->event_data.cr_arrival_event_data.sp_handle = sp_handle;
        event_ptr->event_data.cr_arrival_event_data.local_ia_address_ptr
            = ia_address_ptr;
        event_ptr->event_data.cr_arrival_event_data.conn_qual = conn_qual;
        event_ptr->event_data.cr_arrival_event_data.cr_handle = cr_handle;

        dapli_evd_post_event(evd_ptr, event_ptr);
        return (DAT_SUCCESS);
}


DAT_RETURN
dapls_evd_post_connection_event(
    IN DAPL_EVD                         *evd_ptr,
    IN DAT_EVENT_NUMBER                 event_number,
    IN DAT_EP_HANDLE                    ep_handle,
    IN DAT_COUNT                        private_data_size,
    IN DAT_PVOID                        private_data)
{
        DAT_EVENT               *event_ptr;
        event_ptr = dapli_evd_get_and_init_event(evd_ptr, event_number);
        /*
         * Note event lock may be held on successful return
         * to be released by dapli_evd_post_event(), if provider side locking
         * is needed.
         */

        if (!event_ptr) {
                return (DAT_INSUFFICIENT_RESOURCES | DAT_RESOURCE_MEMORY);
        }

        event_ptr->event_data.connect_event_data.ep_handle = ep_handle;
        event_ptr->event_data.connect_event_data.private_data_size
            = private_data_size;
        event_ptr->event_data.connect_event_data.private_data = private_data;

        dapli_evd_post_event(evd_ptr, event_ptr);
        return (DAT_SUCCESS);
}


DAT_RETURN
dapls_evd_post_async_error_event(
    IN DAPL_EVD                         *evd_ptr,
    IN DAT_EVENT_NUMBER                 event_number,
    IN DAT_IA_HANDLE                    ia_handle)
{
        DAT_EVENT               *event_ptr;
        event_ptr = dapli_evd_get_and_init_event(evd_ptr, event_number);
        /*
         * Note event lock may be held on successful return
         * to be released by dapli_evd_post_event(), if provider side locking
         * is needed.
         */

        if (!event_ptr) {
                return (DAT_INSUFFICIENT_RESOURCES | DAT_RESOURCE_MEMORY);
        }

        event_ptr->event_data.asynch_error_event_data.dat_handle = ia_handle;

        dapli_evd_post_event(evd_ptr, event_ptr);
        return (DAT_SUCCESS);
}


DAT_RETURN
dapls_evd_post_software_event(
    IN DAPL_EVD                         *evd_ptr,
    IN DAT_EVENT_NUMBER                 event_number,
    IN DAT_PVOID                        pointer)
{
        DAT_EVENT               *event_ptr;
        event_ptr = dapli_evd_get_and_init_event(evd_ptr, event_number);
        /*
         * Note event lock may be held on successful return
         * to be released by dapli_evd_post_event(), if provider side locking
         * is needed.
         */

        if (!event_ptr) {
                return (DAT_QUEUE_FULL);
        }

        event_ptr->event_data.software_event_data.pointer = pointer;

        dapli_evd_post_event(evd_ptr, event_ptr);
        return (DAT_SUCCESS);
}

void
dapls_evd_post_premature_events(IN DAPL_EP *ep_ptr)
{
        DAPL_EVD                *evd_ptr;
        DAT_EVENT               *event;
        ib_work_completion_t    *cqe;
        uint32_t                qpn;
        int                     prm_idx;
        int                     nevents;
        int                     i;

        dapls_ib_poll_premature_events(ep_ptr, &cqe, &nevents);
        /* premature events are always recv events */
        evd_ptr = ep_ptr->param.recv_evd_handle;
        qpn = ep_ptr->qpn;

        i = 0;
        prm_idx = 0;
        while (i < nevents) {
                /*
                 * If srq_attached, premature events cannot exceed max_recv_dtos
                 */
                dapl_os_assert(!ep_ptr->srq_attached ||
                    (prm_idx <= ((DAPL_SRQ *)ep_ptr->param.srq_handle)->
                    param.max_recv_dtos));

                /*
                 * The SRQ premature event list could potentially have
                 * holes (ie. free entries in the middle) or premature
                 * events for other QPs. These need to be skipped.
                 */
                if (ep_ptr->srq_attached &&
                    (!DAPL_CQE_IS_VALID(&cqe[prm_idx]) ||
                    (DAPL_GET_CQE_QPN(&cqe[prm_idx]) != qpn))) {
                        prm_idx++;
                        continue;
                }

                dapl_dbg_log(DAPL_DBG_TYPE_DTO_COMP_ERR,
                    " Premature DTO processing\n");

#ifdef  DAPL_DBG        /* For debugging.  */
                dapli_evd_eh_print_cqe(cqe[i]);
#endif
                /*
                 * Can use DAT_DTO_COMPLETION_EVENT because
                 * dapli_evd_cqe_to_event will overwrite.
                 */
                event = dapli_evd_get_and_init_event(evd_ptr,
                    DAT_DTO_COMPLETION_EVENT);
                if (event == NULL) {
                        /* We've already attempted the overflow post, return */
                        return;
                }
                (void) dapli_evd_cqe_to_event(evd_ptr, &cqe[i], DAT_TRUE,
                    event);
                dapli_evd_post_event_nosignal(evd_ptr, event);
                /*
                 * For SRQ attached QPs recycle the premature event
                 */
                if (ep_ptr->srq_attached) {
                        dapls_ib_free_premature_events(ep_ptr, prm_idx);
                        prm_idx++;
                }
                i++;
        }
}

/*
 * dapli_evd_cqe_to_event
 *
 * Convert a CQE into an event structure.
 *
 * Input:
 *      evd_ptr
 *      cqe_ptr
 *
 * Output:
 *      event_ptr
 *
 * Returns:
 *      none
 *
 */
static DAT_BOOLEAN
dapli_evd_cqe_to_event(
    IN DAPL_EVD                 *evd_ptr,
    IN ib_work_completion_t     *cqe_ptr,
    IN DAT_BOOLEAN              process_premature_events,
    OUT DAT_EVENT               *event_ptr)
{
        DAPL_EP                 *ep_ptr;
        DAPL_SRQ                *srq_ptr;
        DAPL_COOKIE             *cookie;
        DAT_EP_STATE            ep_state;
        ib_qp_handle_t          qp;
        ib_uint32_t             ib_status;
        ib_uint32_t             ibtype;
        int                     srq_enabled;
        int                     dto_error = 0;


        /*
         * All that can be relied on if the status is bad is the status
         * and WRID.
         */
        ib_status = DAPL_GET_CQE_STATUS(cqe_ptr);

        cookie = (DAPL_COOKIE *)((uintptr_t)DAPL_GET_CQE_WRID(cqe_ptr));
        dapl_os_assert((NULL != cookie));

        if (cookie->queue_type == DAPL_COOKIE_QUEUE_EP) {
                srq_enabled = 0;
                ep_ptr = cookie->queue.ep;
        } else {
                srq_enabled = 1;
                srq_ptr = cookie->queue.srq;
                dapl_os_assert(NULL != srq_ptr);
                dapl_os_assert(srq_ptr->header.magic == DAPL_MAGIC_SRQ);
                ib_status = DAPL_GET_CQE_STATUS(cqe_ptr);
                ep_ptr = dapls_ib_srq_lookup_ep(srq_ptr, cqe_ptr);
        }

        dapl_os_assert((NULL != ep_ptr));
        dapl_os_assert((ep_ptr->header.magic == DAPL_MAGIC_EP) ||
            (ep_ptr->header.magic == DAPL_MAGIC_EP_EXIT));

        event_ptr->evd_handle = (DAT_EVD_HANDLE) evd_ptr;

        /*
         * Check if the DTO completion arrived before CONNECTION_ESTABLISHED
         * event -
         *
         * Send DTOs can occur only if ep state is CONNECTED/DISCONNECTED
         * therefore it cannot occur before connection established event.
         * Receive DTO can potentially complete before connection established
         * event has been delivered to the client. In this case if the
         * ep state is ACTIVE_CONNECTION_PENDING (active side) or
         * COMPLETION_PENDING (passive side) the event is put in a special
         * event queue in the qp_handle.
         *
         */
        if (!process_premature_events &&
            (cookie->type == DAPL_COOKIE_TYPE_DTO) &&
            (ib_status == IB_COMP_ST_SUCCESS)) {
                ep_state = ep_ptr->param.ep_state;
                qp = ep_ptr->qp_handle;
                if ((ep_state == DAT_EP_STATE_ACTIVE_CONNECTION_PENDING) ||
                    (ep_state == DAT_EP_STATE_COMPLETION_PENDING) ||
                    (qp->qp_num_premature_events > 0)) {
                        /*
                         * not yet ready to put the event in the evd ring
                         * buffer
                         */
                        dapls_ib_store_premature_events(qp, cqe_ptr);
                        return (DAT_FALSE);
                }
        }

        switch (cookie->type) {
        case DAPL_COOKIE_TYPE_DTO:
        {
                DAPL_COOKIE_BUFFER      *buffer;

                if (DAPL_DTO_TYPE_RECV == cookie->val.dto.type) {
                        if (srq_enabled) {
                                dapl_os_atomic_dec(&srq_ptr->recv_count);
                                buffer = &srq_ptr->recv_buffer;
                        } else {
                                dapl_os_atomic_dec(&ep_ptr->recv_count);
                                buffer = &ep_ptr->recv_buffer;
                        }
                } else {
                        dapl_os_atomic_dec(&ep_ptr->req_count);
                        buffer = &ep_ptr->req_buffer;
                }

                event_ptr->event_number = DAT_DTO_COMPLETION_EVENT;
                event_ptr->event_data.dto_completion_event_data.ep_handle =
                    ep_ptr;
                event_ptr->event_data.dto_completion_event_data.user_cookie =
                    cookie->val.dto.cookie;

                switch (ib_status) {
                case IB_COMP_ST_SUCCESS:
                {
                        ibtype = DAPL_GET_CQE_OPTYPE(cqe_ptr);

                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_SUCCESS;
                        dapl_os_assert((ibtype == OP_SEND &&
                            cookie->val.dto.type == DAPL_DTO_TYPE_SEND) ||
                            (ibtype == OP_RECEIVE &&
                            cookie->val.dto.type == DAPL_DTO_TYPE_RECV) ||
                            (ibtype == OP_RDMA_WRITE &&
                            cookie->val.dto.type ==
                            DAPL_DTO_TYPE_RDMA_WRITE) ||
                            (ibtype == OP_RDMA_READ &&
                            cookie->val.dto.type ==
                            DAPL_DTO_TYPE_RDMA_READ));
                        break;
                }
                case IB_COMP_ST_LOCAL_LEN_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_LOCAL_LENGTH;
                        break;
                }
                case IB_COMP_ST_LOCAL_PROTECT_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_LOCAL_PROTECTION;
                        break;
                }
                case IB_COMP_ST_WR_FLUSHED_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_FLUSHED;
                        break;
                }
                case IB_COMP_ST_BAD_RESPONSE_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_BAD_RESPONSE;
                        break;
                }
                case IB_COMP_ST_REM_REQ_ERR:
                case IB_COMP_ST_REM_OP_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_REMOTE_RESPONDER;
                        break;
                }
                case IB_COMP_ST_REM_ACC_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_REMOTE_ACCESS;
                        break;
                }
                /*
                 * Unsupported RD errors
                 * case IB_COMP_ST_EE_STATE_ERR:
                 * case IB_COMP_ST_EE_CTX_NO_ERR:
                 */
                case IB_COMP_ST_TRANSP_COUNTER:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_TRANSPORT;
                        break;
                }
                case IB_COMP_ST_RNR_COUNTER:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_RECEIVER_NOT_READY;
                        break;
                }
                case IB_COMP_ST_MW_BIND_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_RMR_OPERATION_FAILED;
                        break;
                }
                case IB_COMP_ST_LOCAL_OP_ERR:
                {
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_ERR_LOCAL_EP;
                        break;
                }
                default:
                {
                        dapl_dbg_log(DAPL_DBG_TYPE_DTO_COMP_ERR,
                            " DTO completion ERROR: %d: op %#x\n",
                            DAPL_GET_CQE_STATUS(cqe_ptr),
                            DAPL_GET_CQE_OPTYPE(cqe_ptr));
                        event_ptr->event_data.dto_completion_event_data.status =
                            DAT_DTO_FAILURE;
                        break;
                }
                }

                /* Most error DTO ops result in disconnecting the EP */
                if ((event_ptr->event_data.dto_completion_event_data.status !=
                    DAT_DTO_SUCCESS) &&
                    (event_ptr->event_data.dto_completion_event_data.status !=
                    DAT_RMR_OPERATION_FAILED)) {
                        dto_error = 1;
                        dapl_dbg_log(DAPL_DBG_TYPE_DTO_COMP_ERR,
                            " DTO completion ERROR: %d: op %#x\n",
                            DAPL_GET_CQE_STATUS(cqe_ptr),
                            DAPL_GET_CQE_OPTYPE(cqe_ptr));
                }

                if (cookie->val.dto.type == DAPL_DTO_TYPE_SEND ||
                    cookie->val.dto.type == DAPL_DTO_TYPE_RDMA_WRITE) {
                        /* Get size from DTO; CQE value may be off.  */
                        event_ptr->event_data.dto_completion_event_data.
                            transfered_length = cookie->val.dto.size;
                } else {
                        event_ptr->event_data.dto_completion_event_data.
                            transfered_length = DAPL_GET_CQE_BYTESNUM(cqe_ptr);
                }

                dapls_cookie_dealloc(buffer, cookie);
                break;
        }

        case DAPL_COOKIE_TYPE_RMR:
        {
                dapl_os_atomic_dec(&ep_ptr->req_count);

                event_ptr->event_number = DAT_RMR_BIND_COMPLETION_EVENT;

                event_ptr->event_data.rmr_completion_event_data.rmr_handle =
                    cookie->val.rmr.rmr;
                event_ptr->event_data.rmr_completion_event_data.user_cookie =
                    cookie->val.rmr.cookie;
                if (ib_status == IB_COMP_ST_SUCCESS) {
                        ibtype = DAPL_GET_CQE_OPTYPE(cqe_ptr);

                        event_ptr->event_data.rmr_completion_event_data.status =
                            DAT_RMR_BIND_SUCCESS;
                        dapl_os_assert(ibtype == OP_BIND_MW);
                } else {
                        event_ptr->event_data.rmr_completion_event_data.status =
                            DAT_RMR_BIND_FAILURE;
                        dto_error = 1;
                }

                dapls_cookie_dealloc(&ep_ptr->req_buffer, cookie);
                break;
        }
        default:
        {
                dapl_os_assert(!"Invalid Operation type");
                break;
        }
        }

        /*
         * A DTO failed this will cause the connection to be broken
         */
        if ((dto_error) && (ep_ptr->param.ep_state == DAT_EP_STATE_CONNECTED)) {
                ep_ptr->param.ep_state = DAT_EP_STATE_DISCONNECTED;
                /*
                 * Disconnect at the IB level.
                 */
                dapls_ib_disconnect_clean(ep_ptr, DAT_TRUE, IB_CME_CONNECTED);
        }
        /* convert premature rec to error flush on disconnect */
        if (process_premature_events && (ep_ptr->param.ep_state ==
            DAT_EP_STATE_DISCONNECTED) && (ib_status == IB_COMP_ST_SUCCESS)) {
                dapl_os_assert(ibtype == OP_RECEIVE &&
                    cookie->val.dto.type == DAPL_DTO_TYPE_RECV);
                event_ptr->event_data.dto_completion_event_data.status =
                    DAT_DTO_ERR_FLUSHED;
        }
        return (DAT_TRUE);
}

/*
 * dapls_evd_copy_cq
 *
 * Copy all entries on a CQ associated with the EVD onto that EVD
 * Up to caller to handle races, if any.  Note that no EVD waiters will
 * be awoken by this copy.
 *
 * Input:
 *      evd_ptr
 *
 * Output:
 *      nevents
 *
 * Returns:
 *      none
 *
 */
void
dapls_evd_copy_cq(
        DAPL_EVD        *evd_ptr,
        int             *nevents)
{
        ib_work_completion_t    cqe[MAX_CQES_PER_POLL];
        DAT_RETURN              dat_status;
        ib_cq_handle_t          cq_handle;
        DAT_EVENT               *event;
        uint_t                  num_cqes_polled = 0;
        int                     cqe_events;
        int                     i;

        cq_handle = evd_ptr->ib_cq_handle;

        *nevents = 0;

        if (cq_handle == IB_INVALID_HANDLE) {
                /* Nothing to do if no CQ.  */
                return;
        }
        dat_status = DAPL_POLL(evd_ptr)(cq_handle,
            cqe, MAX_CQES_PER_POLL, &num_cqes_polled);

        if (dat_status == DAT_SUCCESS) {
                dapl_dbg_log(DAPL_DBG_TYPE_EVD, "dapls_evd_copy_cq: %u\n",
                    num_cqes_polled);
                cqe_events = 0;
                for (i = 0; i < num_cqes_polled; i++) {
#ifdef  DAPL_DBG        /* For debugging.  */
                        dapli_evd_eh_print_cqe(cqe[i]);
#endif

                        /*
                         * Can use DAT_DTO_COMPLETION_EVENT because
                         * dapli_evd_cqe_to_event will overwrite.
                         */

                        event = dapli_evd_get_and_init_event(
                            evd_ptr, DAT_DTO_COMPLETION_EVENT);
                        if (event == NULL) {
                        /*
                         * We've already attempted the overflow post; return.
                         */
                                return;
                        }
                        if (dapli_evd_cqe_to_event(evd_ptr, &cqe[i], DAT_FALSE,
                            event)) {
                                dapli_evd_post_event_nosignal(evd_ptr, event);
                                cqe_events++;
                        } else {
                                dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                                    "dapls_evd_copy_cq: premature event\n");
                                /*
                                 * We've deferred processing the CQE, so add
                                 * the event_ptr back to free queue
                                 */
                                dat_status = dapls_rbuf_add(&evd_ptr->
                                    free_event_queue, (void *)event);
                                dapl_os_assert(dat_status == DAT_SUCCESS);
                                if (evd_ptr->evd_producer_locking_needed) {
                                        dapl_os_unlock(&evd_ptr->header.lock);
                                }
                        }
                }
                *nevents = cqe_events;
        } else if (DAT_GET_TYPE(dat_status) != DAT_QUEUE_EMPTY) {
                dapl_dbg_log(DAPL_DBG_TYPE_ERR,
                    "dapls_evd_copy_cq: dapls_ib_completion_poll "
                    "returned 0x%x\n", dat_status);
                dapl_os_assert(!"Bad return from dapls_ib_completion_poll");
        }
}

/*
 * dapls_evd_copy_events
 *
 * Copy all events associated with the EVD onto that EVD
 *
 * Input:
 *      evd_ptr
 *      timeout
 *
 * Output:
 *      return status
 *
 * Returns:
 *      none
 *
 */
DAT_RETURN
dapls_evd_copy_events(DAPL_EVD *evd_ptr, DAT_TIMEOUT timeout)
{
        dapl_ib_event_t evp_arr[NUM_EVENTS_PER_POLL];
        dapl_ib_event_t *evpp_start;
        dapl_ib_event_t *evpp;
        DAPL_IA         *ia_ptr;
        DAT_RETURN      dat_status;
        int             waited;
        uint64_t        curr_time;
        uint64_t        final_time;
        uint64_t        time_left;
        int             events_needed = 0;
        int             nevents = 0;
        int             num_cqe = 0;
        int             num_ke = 0; /* kernel events - CM or ASYNC events */
        int             i;

        /* rbuf count is zero on entry */

        if (evd_ptr->evd_flags & (DAT_EVD_CONNECTION_FLAG |
            DAT_EVD_CR_FLAG | DAT_EVD_ASYNC_FLAG)) {
                if (evd_ptr->threshold <= NUM_EVENTS_PER_POLL) {
                        evpp = evp_arr;
                } else {
                        /* need to allocate on the heap */
                        evpp = (dapl_ib_event_t *)dapl_os_alloc(
                            evd_ptr->threshold * sizeof (dapl_ib_event_t));
                        if (evpp == NULL) {
                                return (DAT_INSUFFICIENT_RESOURCES);
                        }
                }
                evpp_start = evpp;
                /* for evd_dequeue, check for ke before returning Q_EMPTY */
                if (evd_ptr->threshold == 0 && timeout == 0)
                        evd_ptr->threshold = 1;
        } else {
                evpp = NULL;
                evpp_start = NULL;
        }
        ia_ptr = evd_ptr->header.owner_ia;
        waited = 0;
        dat_status = DAT_SUCCESS;

        /* calculate various time wait elements */
        if (timeout == 0) {
                final_time = 0;
                time_left = 0;
        } else if (timeout == DAT_TIMEOUT_INFINITE) {
                /*
                 * The real value of DAT_TIMEOUT_INFINITE is fairly small
                 * ~71 mins, to prevent premature timeouts map it to
                 * 1 year.  NOTE: 64-bit integers are needed here
                 * because 32 bits is not enough.  Other types,
                 * such as clock_t are not 64-bit, so are not
                 * sufficient for this.  Similarly, hrtime_t is
                 * defined as a "nanosecond counter", which does not
                 * match our need for time in microseconds, so we
                 * just use the more general uint64_t here.
                 */
#define DAPL_ONE_YEAR_IN_USEC   ((365 * 24 * 3600) * 1000000LL)
                curr_time = gethrtime();
                time_left = DAPL_ONE_YEAR_IN_USEC;
                final_time = curr_time + DAPL_ONE_YEAR_IN_USEC * 1000;
        } else {
                /*
                 * maximum time by which the routine needs to return
                 * DAT_TIMEOUT_INFINITE is defined as ~0 but its of type int
                 * so mask the MSB to avoid overflow
                 */
                curr_time = gethrtime();
                final_time = curr_time + (uint64_t)(timeout&0x7fffffff)*1000;
                time_left = (final_time - curr_time)/1000;
        }

        do {
                /*
                 * If this evd has a CQ event stream check the CQs first
                 */
                if (evd_ptr->evd_flags & (DAT_EVD_DTO_FLAG |
                    DAT_EVD_RMR_BIND_FLAG)) {
                        /*
                         * Poll CQ for events, update the total number of CQEs
                         * so far
                         */
                        nevents = 0;
                        dapls_evd_copy_cq(evd_ptr, &nevents);
                        num_cqe += nevents;
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: copy_cq num_cqe(%d)\n",
                            num_cqe);
                }

                /*
                 * We use the dapls_rbuf_count since it includes
                 *  - CQ events pulled by dapls_evd_copy_cq
                 *  - events added by dat_evd_post_se()
                 */
                events_needed = evd_ptr->threshold - num_ke -
                    dapls_rbuf_count(&evd_ptr->pending_event_queue);

                /*
                 * check for pending events
                 * note: threshold=0 implies dapl_evd_dequeue
                 */
                if (events_needed < 0) {
                        /* There are more than sufficient events */
                        break;
                } else if (events_needed == 0) {
                        /* report queue empty on dat_evd_dequeue */
                        /* non CQ events are expected to be polled */
                        /* by dat_evd_wait */
                        if (evd_ptr->threshold == 0)
                                dat_status =  DAT_ERROR(DAT_QUEUE_EMPTY, 0);
                        /*
                         * when threshold > 0, we have sufficient events
                         */
                        break;
                } else {
                        /*
                         * when we reach here, this implies dat_evd_wait
                         * return on any dto completion as
                         * threshold > 1 will be taken as hint only
                         */
                        if (num_cqe)
                                break;
                }

                /* check we've already waited */
                if (waited > 0) {
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: waited[%d]\n", waited);
                        if (dat_status != DAT_SUCCESS)
                                break;
                        curr_time = gethrtime();
                        /* exit on time expired */
                        if (curr_time >= final_time)
                                break;
                        time_left = (final_time - curr_time)/1000;
                }

                /* check for DTO type evd's */
                if (evd_ptr->evd_flags & (DAT_EVD_DTO_FLAG |
                    DAT_EVD_RMR_BIND_FLAG)) {
                        if (events_needed == 1) {
                                /*
                                 * Need only one event so enable cq
                                 * notification
                                 */
                                /*
                                 * XXX: Things need to be modified here to
                                 * implement the NOTIFICATION suppression
                                 * correctly - relies on THRESHOLD flag
                                 * and UNSIGNALLED flag to be stored
                                 * in the evd.
                                 */
                                dat_status = dapls_set_cq_notify(ia_ptr,
                                    evd_ptr);
                                if (dat_status != DAT_SUCCESS) {
                                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                                            "dapls_evd_copy_event:"
                                            " set_cq_notify(%d)\n", dat_status);
                                        return (dat_status);
                                }
                        } else if (events_needed > 1) {
                                /*
                                 * We need multiple events so lets enable CQ for
                                 * notification on N events.
                                 * dat_status = dapls_set_cqN_notify(ia_ptr,
                                 * evd_ptr, (uint32_t)events_needed);
                                 */
                                dat_status = dapls_set_cq_notify(ia_ptr,
                                    evd_ptr);
                                if (dat_status != DAT_SUCCESS) {
                                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                                            "dapls_evd_copy_event:"
                                            " set_cqN_notify:%d\n", dat_status);
                                        return (dat_status);
                                }
                        }

                        /*
                         * Per Tavor PRM if completions occur after polling
                         * the CQ and before arming it, upon arming the CQ
                         * handler will be immediately fired. Hence it
                         * recommends that a re-poll of the CQ can be skipped
                         * as an optimization.
                         */
                }

                nevents = 0;

                /*
                 * non-NULL evpp_start denotes either
                 * DAT_EVD_CONNECTION_FLAG, DAT_EVD_CR_FLAG, DAT_EVD_ASYNC_FLAG
                 * is set and thus needs to check events from kernel
                 */
                if (evpp_start) {
                        /*
                         * Even if dat_status is not DAT_SUCCESS, num_events
                         * could be non-zero.
                         */
                        dat_status = dapls_ib_event_poll(evd_ptr, time_left,
                            (evd_ptr->threshold - (num_cqe + num_ke)), evpp,
                            &nevents);
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: poll returned 0x%x(%d)\n",
                            dat_status, nevents);

                        num_ke += nevents;
                        evpp += nevents;
                } else {
                        /* perform a timewait */
                        dat_status = dapls_ib_event_poll(evd_ptr, time_left,
                            0, NULL, &nevents);
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: poll(cq_notification) "
                            "returned 0x%x\n", dat_status);
                        if (DAT_GET_TYPE(dat_status) == DAT_INTERRUPTED_CALL)
                                return (dat_status);
                }

                waited++;
        } while (dapls_rbuf_count(&evd_ptr->pending_event_queue) + num_ke <
            evd_ptr->threshold);

        /* process the cm events now */
        for (i = 0; i < num_ke; i++) {
                switch (evpp_start[i].ibe_ev_family) {
                case DAPL_CR_EVENTS: /* PASSIVE side events */
                case DAPL_PASSIVE_CONNECTION_EVENTS:
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: Passive side Event %d\n",
                            evpp_start[i].ibe_ce.ibce_event);
                        dapls_cr_callback((ib_cm_handle_t)
                            evpp_start[i].ibe_ce.ibce_psep_cookie,
                            evpp_start[i].ibe_ce.ibce_event,
                            evpp_start[i].ibe_ce.ibce_priv_data_ptr, (void *)
                            (uintptr_t)evpp_start[i].ibe_ce.ibce_cookie);
                        break;
                case DAPL_ACTIVE_CONNECTION_EVENTS: /* ACTIVE side events */
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: Active Conn Event %d\n",
                            evpp_start[i].ibe_ce.ibce_event);
                        dapl_evd_connection_callback((ib_cm_handle_t)
                            IB_INVALID_HANDLE,
                            evpp_start[i].ibe_ce.ibce_event,
                            evpp_start[i].ibe_ce.ibce_priv_data_ptr, (void *)
                            (uintptr_t)evpp_start[i].ibe_ce.ibce_cookie);
                        break;
                case DAPL_ASYNC_EVENTS:
                        dapl_dbg_log(DAPL_DBG_TYPE_EVD,
                            "dapls_evd_copy_event: Async Event %d\n",
                            evpp_start[i].ibe_async.ibae_type);
                        dapls_ib_async_callback(evd_ptr,
                            ia_ptr->hca_ptr->ib_hca_handle,
                            &(evpp_start[i].ibe_async), ia_ptr);
                        break;
                default:
                        dapl_dbg_log(DAPL_DBG_TYPE_ERR,
                            "dapls_evd_copy_event: dapls_ib_event_poll %d "
                            "returned 0x%x\n", i, evpp_start[i].ibe_ev_family);
                        dapl_os_assert(!"Bad return from dapls_ib_event_poll");
                        break;
                }
        }

        return (dat_status);
}

/*
 * dapls_evd_cq_poll_to_event
 *
 * Attempt to dequeue a single CQE from a CQ and turn it into
 * an event.
 *
 * Input:
 *      evd_ptr
 *
 * Output:
 *      event
 *
 * Returns:
 *      Status of operation
 *
 */
DAT_RETURN
dapls_evd_cq_poll_to_event(IN DAPL_EVD *evd_ptr, OUT DAT_EVENT *event)
{
        DAT_RETURN              dat_status;
        ib_work_completion_t    cur_cqe;

        /* skip one layer of do-nothing function */
        dat_status = DAPL_POLL1(evd_ptr)(evd_ptr->ib_cq_handle, &cur_cqe);

        if (dat_status == DAT_SUCCESS) {
#ifdef  DAPL_DBG        /* For debugging.  */
                dapli_evd_eh_print_cqe(cur_cqe);
#endif
                (void) dapli_evd_cqe_to_event(evd_ptr, &cur_cqe, DAT_FALSE,
                    event);
        }

        return (dat_status);
}

/*
 * Local variables:
 *  c-indent-level: 4
 *  c-basic-offset: 4
 *  tab-width: 8
 * End:
 */