/*
 *  Setup peer state to reflect that connection has been established,
 *  and start any pending sends.
 */
void pmix_server_peer_connected(pmix_server_peer_t* peer)
{
    opal_output_verbose(2, pmix_server_output,
                        "%s-%s usock_peer_connected on socket %d",
                        ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
                        ORTE_NAME_PRINT(&(peer->name)), peer->sd);

    if (peer->timer_ev_active) {
        opal_event_del(&peer->timer_event);
        peer->timer_ev_active = false;
    }
    peer->state = PMIX_SERVER_CONNECTED;

    /* ensure the recv event is active */
    if (!peer->recv_ev_active) {
        opal_event_add(&peer->recv_event, 0);
        peer->recv_ev_active = true;
    }

    /* initiate send of first message on queue */
    if (NULL == peer->send_msg) {
        peer->send_msg = (pmix_server_send_t*)
            opal_list_remove_first(&peer->send_queue);
    }
    if (NULL != peer->send_msg && !peer->send_ev_active) {
        opal_event_add(&peer->send_event, 0);
        peer->send_ev_active = true;
    }
}

/*
 * Check the status of the connection. If the connection failed, will retry
 * later. Otherwise, send this processes identifier to the peer on the
 * newly connected socket.
 */
static void mca_oob_tcp_peer_complete_connect(mca_oob_tcp_peer_t* peer)
{
    int so_error = 0;
    opal_socklen_t so_length = sizeof(so_error);

    /* unregister from receiving event notifications */
    opal_event_del(&peer->peer_send_event);

    /* check connect completion status */
    if(getsockopt(peer->peer_sd, SOL_SOCKET, SO_ERROR, (char *)&so_error, &so_length) < 0) {
        opal_output(0, "[%lu,%lu,%lu]-[%lu,%lu,%lu] mca_oob_tcp_peer_complete_connect: getsockopt() failed: %s (%d)\n", 
            ORTE_NAME_ARGS(orte_process_info.my_name),
            ORTE_NAME_ARGS(&(peer->peer_name)),
            strerror(opal_socket_errno),
            opal_socket_errno);
        mca_oob_tcp_peer_close(peer);
        return;
    }

    if(so_error == EINPROGRESS) {
        opal_event_add(&peer->peer_send_event, 0);
        return;
    } else if (so_error == ECONNREFUSED || so_error == ETIMEDOUT) {
        struct timeval tv = { 1,0 };
        if (mca_oob_tcp_component.tcp_debug >= OOB_TCP_DEBUG_CONNECT) {
            opal_output(0, "[%lu,%lu,%lu]-[%lu,%lu,%lu] mca_oob_tcp_peer_complete_connect: "
                        "connection failed: %s (%d) - retrying\n", 
                        ORTE_NAME_ARGS(orte_process_info.my_name),
                        ORTE_NAME_ARGS(&(peer->peer_name)),
                        strerror(so_error),
                        so_error);
        }
        mca_oob_tcp_peer_shutdown(peer);
        opal_evtimer_add(&peer->peer_timer_event, &tv);
        return;
    } else if(so_error != 0) {
        /* No need to worry about the return code here - we return regardless
           at this point, and if an error did occur a message has already been
           printed for the user */
        mca_oob_tcp_peer_try_connect(peer);
        return;
    }

    if(mca_oob_tcp_component.tcp_debug >= OOB_TCP_DEBUG_CONNECT) {
        opal_output(0, "[%lu,%lu,%lu]-[%lu,%lu,%lu] mca_oob_tcp_peer_complete_connect: "
                    "sending ack, %d",
                    ORTE_NAME_ARGS(orte_process_info.my_name),
                    ORTE_NAME_ARGS(&(peer->peer_name)), so_error);
    }

    if(mca_oob_tcp_peer_send_connect_ack(peer) == ORTE_SUCCESS) {
        peer->peer_state = MCA_OOB_TCP_CONNECT_ACK;
        opal_event_add(&peer->peer_recv_event, 0);
    } else {
        opal_output(0, "[%lu,%lu,%lu]-[%lu,%lu,%lu] mca_oob_tcp_peer_complete_connect: unable to send connect ack.",
            ORTE_NAME_ARGS(orte_process_info.my_name),
            ORTE_NAME_ARGS(&(peer->peer_name)));
        mca_oob_tcp_peer_close(peer);
    }
}

void mca_btl_sctp_component_accept(void)
{
    if(mca_btl_sctp_component.sctp_if_11) {
        /* 1 to 1 */
        while(true) {
            opal_socklen_t addrlen = sizeof(struct sockaddr_in);
            struct sockaddr_in addr;
            mca_btl_sctp_event_t *event;
            int rc, sd = accept(mca_btl_sctp_component.sctp_listen_sd, (struct sockaddr*)&addr, &addrlen);
            if(sd < 0) {
                if(opal_socket_errno == EINTR) {
                    continue;
                }
                if(opal_socket_errno == ECONNRESET || opal_socket_errno == EBADF) {
                    /* closed remotely while on listen queue */
                    close(sd);
                }
                else if(opal_socket_errno != EAGAIN && opal_socket_errno != EWOULDBLOCK) {
                    BTL_ERROR(("accept() failed with errno %d.", opal_socket_errno));
                }
                return;
            }

            if((rc = mca_btl_sctp_set_socket_options(sd)) != OMPI_SUCCESS) {
                BTL_ERROR(("failed to set socket options"));
                return;
            }

            /* wait for receipt of peers process identifier to complete this connection */

            event = OBJ_NEW(mca_btl_sctp_event_t);
            opal_event_set(&event->event, sd, OPAL_EV_READ, mca_btl_sctp_component_recv_handler, event);
            opal_event_add(&event->event, 0);
        }
    }

    else {
        /*  1 to many */

        /* Called by mca_btl_sctp_recv_handler to get a valid *user pointer */

        mca_btl_sctp_event_t *event;
        int sd = mca_btl_sctp_component.sctp_listen_sd;

        if(sd < 0) {
            BTL_ERROR(("mca_btl_sctp_component_accept(): Invalid socket descriptor.\n"));
        }

        /* wait for receipt of peers process identifier to complete this connection */

        event = OBJ_NEW(mca_btl_sctp_event_t);
        opal_event_set(&event->event, sd, OPAL_EV_READ, mca_btl_sctp_recv_handler, event);
        opal_event_add(&event->event, 0);
    }
}

static void* progress_engine(opal_object_t *obj)
{
    /* define an event that will be used to kick us out of a blocking
     * situation when we want to exit
     */
    /* define an event that will be used to kick us out of a blocking
     * situation when we want to exit
     */
    opal_event_set(my_base, &stop_event,
                   progress_thread_pipe[0], OPAL_EV_READ, stop_handler, NULL);
    opal_event_add(&stop_event, 0);

    while (1) {
        OPAL_ACQUIRE_THREAD(&lock, &cond, &active);
        if (progress_thread_stop) {
            fprintf(stderr, "Thread stopping\n");
            OPAL_RELEASE_THREAD(&lock, &cond, &active);
            opal_event_del(&stop_event);
            return OPAL_THREAD_CANCELLED;
        }
        OPAL_RELEASE_THREAD(&lock, &cond, &active);
        fprintf(stderr, "Looping...\n");
        opal_event_loop(my_base, OPAL_EVLOOP_ONCE);
    }
}

int orte_wait_event(opal_event_t **event, orte_trigger_event_t *trig,
                    char *trigger_name,
                    void (*cbfunc)(int, short, void*))
{
    int p[2];
    
    if (pipe(p) < 0) {
        ORTE_ERROR_LOG(ORTE_ERR_SYS_LIMITS_PIPES);
        return ORTE_ERR_SYS_LIMITS_PIPES;
    }

    /* save the trigger name */
    trig->name = strdup(trigger_name);
    
    /* create the event */
    *event =  (opal_event_t *) malloc(sizeof(opal_event_t));
    
    /* pass back the write end of the pipe */
    trig->channel = p[1];
    
    /* define the event to fire when someone writes to the pipe */
    opal_event_set(opal_event_base, *event, p[0], OPAL_EV_READ, cbfunc, trig);
    
    /* Add it to the active events, without a timeout */
    opal_event_add(*event, NULL);

    /* all done */
    return ORTE_SUCCESS;
}

int orte_iof_base_endpoint_ack(
    orte_iof_base_endpoint_t* endpoint,
    uint32_t seq)
{
    bool window_closed, window_open;

    OPAL_THREAD_LOCK(&orte_iof_base.iof_lock);
    window_closed =
        ORTE_IOF_BASE_SEQDIFF(endpoint->ep_seq,endpoint->ep_ack) >= orte_iof_base.iof_window_size;
    endpoint->ep_ack = seq;
    window_open =
        ORTE_IOF_BASE_SEQDIFF(endpoint->ep_seq,endpoint->ep_ack) < orte_iof_base.iof_window_size;

    /* someone is waiting on all output to be flushed */
    if(orte_iof_base.iof_waiting && endpoint->ep_seq == endpoint->ep_ack) {
        opal_condition_signal(&orte_iof_base.iof_condition);
    }

    /* check to see if we need to reenable forwarding */
    if(window_closed && window_open) {
        opal_output(orte_iof_base.iof_output, "iof_base_endpoint ack; re-enabled reading for endpoint");
        opal_event_add(&endpoint->ep_event, 0);
    }
    OPAL_THREAD_UNLOCK(&orte_iof_base.iof_lock);
    return ORTE_SUCCESS;
}

opal_event_base_t *opal_progress_thread_init(const char *name)
{
    assert(NULL == name);

    /* Create the event base */
    agent_evbase = opal_event_base_create();
    if (NULL == agent_evbase) {
        return NULL;
    }

    /* add an event to the new event base (if there are no events,
       opal_event_loop() will return immediately) */
    opal_event_set(agent_evbase, &blocker, -1, OPAL_EV_PERSIST,
                   blocker_timeout_cb, NULL);
    opal_event_add(&blocker, &long_timeout);

    /* Spawn the agent thread event loop */
    OBJ_CONSTRUCT(&agent_thread, opal_thread_t);
    agent_thread.t_run = agent_thread_main;
    agent_thread.t_arg = NULL;
    int ret;
    ret = opal_thread_start(&agent_thread);
    if (OPAL_SUCCESS != ret) {
        OPAL_ERROR_LOG(ret);
        ABORT("Failed to start usNIC agent thread");
        /* Will not return */
    }

    return agent_evbase;
}

/*
 * Add an fd to the listening set
 */
static int service_pipe_cmd_add_fd(bool use_libevent, cmd_t *cmd)
{
    registered_item_t *ri = OBJ_NEW(registered_item_t);
    if (NULL == ri) {
        return OMPI_ERR_OUT_OF_RESOURCE;
    }
    ri->ri_event_used = false;
    ri->ri_fd = cmd->pc_fd;
    ri->ri_flags = cmd->pc_flags;
    ri->ri_callback.event = cmd->pc_fn.event;
    ri->ri_context = cmd->pc_context;

    if (use_libevent) {
        /* Make an event for this fd */
        ri->ri_event_used = true;
        opal_event_set(opal_event_base, &ri->ri_event, ri->ri_fd,
                       ri->ri_flags | OPAL_EV_PERSIST, service_fd_callback,
                       ri);
        opal_event_add(&ri->ri_event, 0);
    } else {
        /* Add the fd to the relevant fd local sets and update max_fd */
        if (OPAL_EV_READ & ri->ri_flags) {
            FD_SET(ri->ri_fd, &read_fds);
        }
        if (OPAL_EV_WRITE & cmd->pc_flags) {
            FD_SET(ri->ri_fd, &write_fds);
        }
        max_fd = (max_fd > ri->ri_fd) ? max_fd : ri->ri_fd + 1;
    }

    opal_list_append(&registered_items, &ri->super);
    return OMPI_SUCCESS;
}

static void restart_stdin(int fd, short event, void *cbdata)
{
    if (NULL != mca_iof_hnp_component.stdinev &&
        !orte_job_term_ordered) {
        mca_iof_hnp_component.stdinev->active = true;
        opal_event_add(&(mca_iof_hnp_component.stdinev->ev), 0);
    }
}

static void stop_handler(int sd, short flags, void* cbdata)
{
    char byte;

    opal_fd_read(progress_thread_pipe[0], 1, &byte);
    fprintf(stderr, "Stop handler called\n");
    /* reset the event */
    opal_event_add(&stop_event, 0);
    return;
}

static int setup_channel(rmcast_base_channel_t *chan, uint8_t direction)
{
    int rc;
    int xmitsd, recvsd;
    
    if (0 <= chan->xmit && 0 <= chan->recv) {
        /* already setup */
        OPAL_OUTPUT_VERBOSE((2, orte_rmcast_base.rmcast_output,
                             "%s setup:channel %d already setup",
                             ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
                             chan->channel));
        return ORTE_SUCCESS;
    }
    
    /* setup the IPv4 addr info */
    chan->addr.sin_family = AF_INET;
    chan->addr.sin_addr.s_addr = htonl(chan->network);
    chan->addr.sin_port = htons(chan->port);
    
    OPAL_OUTPUT_VERBOSE((2, orte_rmcast_base.rmcast_output,
                         "%s setup:channel addr %03d.%03d.%03d.%03d port %d for %s:%s",
                         ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),
                         OPAL_IF_FORMAT_ADDR(chan->network), (int)chan->port,
                         (ORTE_RMCAST_RECV & direction) ? " RECV" : " ",
                         (ORTE_RMCAST_XMIT & direction) ? " XMIT" : " "));
    
    if (0 > chan->xmit && (ORTE_RMCAST_XMIT & direction)) {
        /* create a xmit socket */
        if (ORTE_SUCCESS != (rc = setup_socket(&xmitsd, chan, false))) {
            ORTE_ERROR_LOG(rc);
            return rc;
        }
        chan->xmit = xmitsd;
    }
    
    if (0 > chan->recv && (ORTE_RMCAST_RECV & direction)) {
        /* create a recv socket */
        if (ORTE_SUCCESS != (rc = setup_socket(&recvsd, chan, true))) {
            ORTE_ERROR_LOG(rc);
            return rc;
        }
        chan->recv = recvsd;
        
        /* setup an event to catch messages */
        OPAL_OUTPUT_VERBOSE((2, orte_rmcast_base.rmcast_output,
                             "%s setup:channel activating recv event on fd %d",
                             ORTE_NAME_PRINT(ORTE_PROC_MY_NAME),(int)chan->recv));
        
        opal_event_set(opal_event_base, &chan->recv_ev, chan->recv,
                       OPAL_EV_READ|OPAL_EV_PERSIST, recv_handler, chan);
        opal_event_add(&chan->recv_ev, 0);
    }

    return ORTE_SUCCESS;
}

static void orte_iof_base_endpoint_stdin_cb(int sd, short flags, void *user)
{
    orte_iof_base_endpoint_t* endpoint = (orte_iof_base_endpoint_t*)user;
    bool should_process = orte_iof_base_endpoint_stdin_check(endpoint->ep_fd);

    if (should_process) {
        opal_event_add(&endpoint->ep_event, 0);
    } else {
        opal_event_del(&endpoint->ep_event);
    }
}

int
main (int argc, char **argv)
{
    struct opal_event signal_int, signal_term;

    /* Initalize the event library */
    opal_event_init();

    /* Initalize one event */
    opal_event_set(&signal_term, SIGUSR1, OPAL_EV_SIGNAL|OPAL_EV_PERSIST, signal_cb,
                   &signal_term);
    opal_event_set(&signal_int, SIGUSR2, OPAL_EV_SIGNAL|OPAL_EV_PERSIST, signal_cb,
                   &signal_int);

    opal_event_add(&signal_int, NULL);
    opal_event_add(&signal_term, NULL);

    opal_event_dispatch();

    return (0);
}

void mca_oob_ud_event_start_monitor (mca_oob_ud_device_t *device)
{
    if (!event_started) {
#if !ORTE_ENABLE_PROGRESS_THREADS
        opal_progress_event_users_increment ();
#endif
        opal_event_set (orte_event_base, &device->event, device->ib_channel->fd,
                        OPAL_EV_READ, mca_oob_ud_event_dispatch, (void *) device);
        opal_event_add (&device->event, NULL);
        event_started = true;
    }
}

void mca_oob_ud_event_queue_completed (mca_oob_ud_req_t *req)
{
    struct timeval now = {0, 0};

    mca_oob_ud_req_append_to_list (req, &mca_oob_ud_component.ud_event_queued_reqs);

    if (!opal_event_evtimer_pending (&mca_oob_ud_component.ud_complete_event, &now)) {
        opal_event_evtimer_set (orte_event_base, &mca_oob_ud_component.ud_complete_event,
                                mca_oob_ud_complete_dispatch, NULL);
        opal_event_add (&mca_oob_ud_component.ud_complete_event, &now);
    }
}

void orte_iof_mrhnp_stdin_cb(int fd, short event, void *cbdata)
{
    bool should_process = orte_iof_mrhnp_stdin_check(0);
    
    if (should_process) {
        mca_iof_mr_hnp_component.stdinev->active = true;
        opal_event_add(mca_iof_mr_hnp_component.stdinev->ev, 0);
    } else {
        opal_event_del(mca_iof_mr_hnp_component.stdinev->ev);
        mca_iof_mr_hnp_component.stdinev->active = false;
    }
}

/*
 *  Setup peer state to reflect that connection has been established,
 *  and start any pending sends.
 */
static void mca_oob_tcp_peer_connected(mca_oob_tcp_peer_t* peer)
{
    opal_event_del(&peer->peer_timer_event);
    peer->peer_state = MCA_OOB_TCP_CONNECTED;
    peer->peer_retries = 0;
    if(opal_list_get_size(&peer->peer_send_queue) > 0) {
        if(NULL == peer->peer_send_msg)
            peer->peer_send_msg = (mca_oob_tcp_msg_t*)
                opal_list_remove_first(&peer->peer_send_queue);
        opal_event_add(&peer->peer_send_event, 0);
    }
}

static int
bufferevent_add(struct opal_event *ev, int timeout)
{
	struct timeval tv, *ptv = NULL;

	if (timeout) {
		timerclear(&tv);
		tv.tv_sec = timeout;
		ptv = &tv;
	}

	return (opal_event_add(ev, ptv));
}

/*
 * Check the status of the connection. If the connection failed, will retry
 * later. Otherwise, send this processes identifier to the endpoint on the 
 * newly connected socket.
 */
static void mca_btl_tcp2_endpoint_complete_connect(mca_btl_base_endpoint_t* btl_endpoint)
{
    int so_error = 0;
    opal_socklen_t so_length = sizeof(so_error);
    struct sockaddr_storage endpoint_addr;

    mca_btl_tcp2_proc_tosocks(btl_endpoint->endpoint_addr, &endpoint_addr);

    /* unregister from receiving event notifications */
    opal_event_del(&btl_endpoint->endpoint_send_event);

    /* check connect completion status */
    if(getsockopt(btl_endpoint->endpoint_sd, SOL_SOCKET, SO_ERROR, (char *)&so_error, &so_length) < 0) {
        BTL_ERROR(("getsockopt() to %s failed: %s (%d)", 
                   opal_net_get_hostname((struct sockaddr*) &endpoint_addr),
                   strerror(opal_socket_errno), opal_socket_errno));
        mca_btl_tcp2_endpoint_close(btl_endpoint);
        return;
    }
    if(so_error == EINPROGRESS || so_error == EWOULDBLOCK) {
        opal_event_add(&btl_endpoint->endpoint_send_event, 0);
        return;
    }
    if(so_error != 0) {
        BTL_ERROR(("connect() to %s failed: %s (%d)", 
                   opal_net_get_hostname((struct sockaddr*) &endpoint_addr),
                   strerror(so_error), so_error));
        mca_btl_tcp2_endpoint_close(btl_endpoint);
        return;
    }

    if(mca_btl_tcp2_endpoint_send_connect_ack(btl_endpoint) == OMPI_SUCCESS) {
        btl_endpoint->endpoint_state = MCA_BTL_TCP_CONNECT_ACK;
        opal_event_add(&btl_endpoint->endpoint_recv_event, 0);
    } else {
        mca_btl_tcp2_endpoint_close(btl_endpoint);
    }
}

static void *mca_oob_ud_event_dispatch(int fd, int flags, void *context)
{
    int rc;
    mca_oob_ud_device_t *device = (mca_oob_ud_device_t *) context;
    mca_oob_ud_port_t *port = NULL;
    struct ibv_cq *event_cq = NULL;
    void *event_context     = NULL;

    do {
        rc = ibv_get_cq_event (device->ib_channel, &event_cq, &event_context);
    } while (rc && errno == EINTR);

    if (NULL == event_cq) {
        /* re-arm the event */
        opal_event_add (&port->device->event, NULL);

        return NULL;
    }

    port = (mca_oob_ud_port_t *) event_context;

    rc = mca_oob_ud_process_messages (event_cq, port);
    if (rc < 0) {
        opal_output (0, "%s oob:ud:event_dispatch error processing messages",
                     ORTE_NAME_PRINT(ORTE_PROC_MY_NAME));
        return NULL;
    }

    if (ibv_req_notify_cq(event_cq, 0)) {
        opal_output (0, "%s oob:ud:event_dispatch error asking for cq notifications",
                     ORTE_NAME_PRINT(ORTE_PROC_MY_NAME));
    }

    /* re-arm the event */
    opal_event_add (&port->device->event, NULL);
 
    return NULL;
}