VILLASnode/lib/nodes/infiniband.c

/** Node type: infiniband
 *
 * @author Dennis Potter <dennis@dennispotter.eu>
 * @copyright 2018, Institute for Automation of Complex Power Systems, EONERC
 * @license GNU General Public License (version 3)
 *
 * VILLASnode
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *********************************************************************************/

#include <string.h>
#include <math.h>

#include <villas/nodes/infiniband.h>
#include <villas/plugin.h>
#include <villas/utils.h>
#include <villas/format_type.h>
#include <villas/memory.h>
#include <villas/pool.h>
#include <villas/memory_ib.h>

#include <rdma/rdma_cma.h>

int ib_cleanup(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    info("Starting to clean up");

    // Destroy QP
    rdma_destroy_qp(ib->ctx.id);
    info("Destroyed QP");

    // Deregister memory regions
    ibv_dereg_mr(ib->mem.mr_recv);
    if(ib->is_source)
        ibv_dereg_mr(ib->mem.mr_send);
    info("Deregistered memory regions");

    // Destroy pools
    pool_destroy(&ib->mem.p_recv);
    pool_destroy(&ib->mem.p_send);
    info("Destroyed memory pools");

    // Destroy RDMA CM ID
    rdma_destroy_id(ib->ctx.id);
    info("Destroyed rdma_cm_id");

    // Destroy event channel
    rdma_destroy_event_channel(ib->ctx.ec);
    info("Destroyed event channel");

    return 0;
}

int ib_post_recv_wrs(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    struct ibv_recv_wr wr, *bad_wr = NULL;
    int ret;
    struct ibv_sge sge;

    // Prepare receive Scatter/Gather element
    sge.addr = (uintptr_t)pool_get(&ib->mem.p_recv);
    sge.length = ib->mem.p_recv.blocksz;
    sge.lkey = ib->mem.mr_recv->lkey;

    // Prepare a receive Work Request
    wr.wr_id = (uintptr_t)sge.addr;
    wr.next = NULL;
    wr.sg_list = &sge;
    wr.num_sge = 1;

    // Post Work Request
    ret = ibv_post_recv(ib->ctx.id->qp, &wr, &bad_wr);

    return ret;
}

void ib_completion_target(struct node* n, struct ibv_wc* wc, int* size)
{
    //ToDo: No implementation yet. This is still handled in ib_read
}

void ib_completion_source(struct node* n, struct ibv_wc* wc, int* size)
{
	struct infiniband *ib = (struct infiniband *)((struct node *)n)->_vd;
	struct sample* smpl;

	for(int i=0; i<*size; i++)
	{
		//On disconnect, the QP set to error state and will be flushed
		if(wc[i].status == IBV_WC_WR_FLUSH_ERR)
		{
			ib->poll.stopThread = 1;
			return;
		}

		if(wc[i].status != IBV_WC_SUCCESS)
		{
			warn("Work Completion status was not IBV_WC_SUCCES in node %s: %i",
				node_name(n), wc[i].status);
		}
		else
		{
			// Release sample
			smpl = (struct sample*)wc[i].wr_id;
			sample_put(smpl);
		}
	}
}

void * ib_event_thread(void *n)
{
    struct infiniband *ib = (struct infiniband *)((struct node *)n)->_vd;
    struct ibv_wc wc[ib->cq_size];
    int size;

    while(1)
    {
        // Function blocks, until an event occurs
        ibv_get_cq_event(ib->ctx.comp_channel, &ib->ctx.cq, NULL);

        // Poll as long as WCs are available
        while((size = ibv_poll_cq(ib->ctx.cq, ib->cq_size, wc)))
            ib->poll.on_compl(n, wc, &size);

        // Request a new event in the CQ and acknowledge event
        ibv_req_notify_cq(ib->ctx.cq, 0);
        ibv_ack_cq_events(ib->ctx.cq, 1);
    }
}

void * ib_busy_poll_thread(void *n)
{
    struct infiniband *ib = (struct infiniband *)((struct node *)n)->_vd;
    struct ibv_wc wc[ib->cq_size];
    int size;

    while(1)
    {
        // Poll as long as WCs are available
        while((size = ibv_poll_cq(ib->ctx.cq, ib->cq_size, wc)))
            ib->poll.on_compl(n, wc, &size);

        if(ib->poll.stopThread)
            return NULL;
    }
}

static void ib_init_wc_poll(struct node *n)
{
    int ret;
    struct infiniband *ib = (struct infiniband *) n->_vd;
    ib->ctx.comp_channel = NULL;

    if(ib->poll.poll_mode == EVENT)
    {
        // Create completion channel
        ib->ctx.comp_channel = ibv_create_comp_channel(ib->ctx.id->verbs);
        if(!ib->ctx.comp_channel)
            error("Could not create completion channel in node %s.", node_name(n));
    }

    // Create completion queue and bind to channel (or NULL)
    ib->ctx.cq = ibv_create_cq(ib->ctx.id->verbs,
                               ib->cq_size,
                               NULL,
                               ib->ctx.comp_channel,
                               0);
    if(!ib->ctx.cq)
        error("Could not create completion queue in node %s.", node_name(n));

    if(ib->poll.poll_mode == EVENT)
    {
        // Request notifications from completion queue
        ret = ibv_req_notify_cq(ib->ctx.cq, 0);
        if(ret)
            error("Failed to request notifiy CQ in node %s: %s",
                node_name(n), gai_strerror(ret));
    }

    // Initialize polling pthread
    //ToDo: Remove if(is_source)
    if(ib->is_source)
    {
        ret = pthread_create(&ib->poll.cq_poller_thread, NULL, ib->poll.poll_func, n);
        if(ret)
        {
            error("Failed to create poll thread of node %s: %s",
                node_name(n), gai_strerror(ret));
        }
    }
}

static void ib_build_ibv(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    int ret;

    //Allocate protection domain
    ib->ctx.pd = ibv_alloc_pd(ib->ctx.id->verbs);
    if(!ib->ctx.pd)
        error("Could not allocate protection domain in node %s.", node_name(n));
    info("Allocated Protection Domain");

    // Initiate poll mode
    ib_init_wc_poll(n);

    // Prepare remaining Queue Pair (QP) attributes
    ib->qp_init.send_cq = ib->ctx.cq;
    ib->qp_init.recv_cq = ib->ctx.cq;

    //ToDo: Set maximum inline data

    // Create the actual QP
    ret = rdma_create_qp(ib->ctx.id, ib->ctx.pd, &ib->qp_init);
    if(ret)
        error("Failed to create Queue Pair in node %s.", node_name(n));

    info("Created Queue Pair with %i receive and %i send elements.",
            ib->qp_init.cap.max_recv_wr, ib->qp_init.cap.max_send_wr);

    // Allocate memory
    ib->mem.p_recv.state = STATE_DESTROYED;
    ib->mem.p_recv.queue.state = STATE_DESTROYED;

    // Set pool size to maximum size of Receive Queue
    pool_init(&ib->mem.p_recv,
            ib->qp_init.cap.max_recv_wr,
            64*sizeof(double),
            &memtype_heap);
    if(ret)
    {
        error("Failed to init recv memory pool of node %s: %s",
            node_name(n), gai_strerror(ret));
    }

    //ToDo: initialize r_addr_key struct if mode is RDMA

    // Register memory for IB Device. Not necessary if data is send
    // exclusively inline
    ib->mem.mr_recv = ibv_reg_mr(
            ib->ctx.pd,
            (char*)&ib->mem.p_recv+ib->mem.p_recv.buffer_off,
            ib->mem.p_recv.len,
            IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
    if(!ib->mem.mr_recv) {
        error("Failed to register mr_recv with ibv_reg_mr of node %s.",
            node_name(n));
    }
    info("Allocated receive memory.");

    if(ib->is_source)
    {
        ib->mem.p_send.state = STATE_DESTROYED;
        ib->mem.p_send.queue.state = STATE_DESTROYED;

        // Set pool size to maximum size of Receive Queue
        pool_init(&ib->mem.p_send,
                ib->qp_init.cap.max_send_wr,
                sizeof(double),
                &memtype_heap);
        if(ret)
        {
            error("Failed to init send memory of node %s: %s",
                node_name(n), gai_strerror(ret));
        }

        //ToDo: initialize r_addr_key struct if mode is RDMA

        // Register memory for IB Device. Not necessary if data is send
        // exclusively inline
        ib->mem.mr_send = ibv_reg_mr(
                ib->ctx.pd,
                (char*)&ib->mem.p_send+ib->mem.p_send.buffer_off,
                ib->mem.p_send.len,
                IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE);
        if(!ib->mem.mr_send) {
            error("Failed to register mr_send with ibv_reg_mr of node %s.",
                node_name(n));
        }
        info("Allocated send memory.");

    }

    // Post Receive Work Requests to be able to receive data
    // Fill complete Receive Queue during initialization
    for(int i=0; i<ib->qp_init.cap.max_recv_wr; i++)
    {
        ret = ib_post_recv_wrs(n);
        if(ret)
        {
            error("Failed to post initial receive Work Requests of node %s.",
                node_name(n));
        }
    }
    info("Filled the complete Receive Queue.");
}

static int ib_addr_resolved(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    int ret;

    info("Successfully resolved address.");

    // Build all components from IB Verbs
    ib_build_ibv(n);

    // Resolve address
    ret = rdma_resolve_route(ib->ctx.id, ib->conn.timeout);
    if(ret)
        error("Failed to resolve route in node %s.", node_name(n));

    return 0;
}

static int ib_route_resolved(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    int ret;

    info("Successfully resolved route.");

    //ToDo: Post receive WRs

    struct rdma_conn_param cm_params;
    memset(&cm_params, 0, sizeof(cm_params));

    // Send connection request
    ret = rdma_connect(ib->ctx.id, &cm_params);
    if(ret)
        error("Failed to connect in node %s.", node_name(n));

    info("Called rdma_connect.");

    return 0;
}

static int ib_connect_request(struct node *n, struct rdma_cm_id *id)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    int ret;
    info("Received a connection request!");

    ib->ctx.id = id;
    ib_build_ibv(n);

    struct rdma_conn_param cm_params;
    memset(&cm_params, 0, sizeof(cm_params));

    // Accept connection request
    ret = rdma_accept(ib->ctx.id, &cm_params);
    if(ret)
        error("Failed to connect in node %s.", node_name(n));

    info("Successfully accepted connection request.");

    return 0;
}

static int ib_event(struct node *n, struct rdma_cm_event *event)
{
    int ret = 0;

    switch(event->event)
    {
        case RDMA_CM_EVENT_ADDR_RESOLVED:
            ret = ib_addr_resolved(n);
            break;
        case RDMA_CM_EVENT_ADDR_ERROR:
            error("Address resolution (rdma_resolve_addr) failed!");
        case RDMA_CM_EVENT_ROUTE_RESOLVED:
            ret = ib_route_resolved(n);
            break;
        case RDMA_CM_EVENT_ROUTE_ERROR:
            error("Route resolution (rdma_resovle_route) failed!");
        case RDMA_CM_EVENT_CONNECT_REQUEST:
            ret = ib_connect_request(n, event->id);
            break;
        case RDMA_CM_EVENT_CONNECT_ERROR:
            error("An error has occurred trying to establish a connection!");
        case RDMA_CM_EVENT_REJECTED:
            error("Connection request or response was rejected by the remote end point!");
        case RDMA_CM_EVENT_ESTABLISHED:
            info("Connection established!");
            ret = 1;
            break;
        case RDMA_CM_EVENT_DISCONNECTED:
            ret = ib_cleanup(n);
            break;
        default:
            error("Unknown event occurred: %u",
                event->event);
    }

    return ret;
}

int ib_reverse(struct node *n)
{
    return 0;
}

int ib_parse(struct node *n, json_t *cfg)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;

    int ret;
    char *local = NULL;
    char *remote = NULL;
    const char *port_space = "RDMA_PC_TCP";
    const char *poll_mode = "BUSY";
    const char *qp_type = "IBV_QPT_RC";
    int timeout = 1000;
    int cq_size = 128;
    int max_send_wr = 128;
    int max_recv_wr = 128;

    json_error_t err;
    ret = json_unpack_ex(cfg, &err, 0, "{ s?: s, s?: s, s?: s, s?: i, \
                                        s?: s, s?: i, s?: s, s?: i, s?: i}",
        "remote", &remote,
        "local", &local,
        "rdma_port_space", &port_space,
        "resolution_timeout", &timeout,
        "poll_mode", &poll_mode,
        "cq_size", &cq_size,
        "qp_type", &qp_type,
        "max_send_wr", &max_send_wr,
        "max_recv_wr", &max_recv_wr
    );
    if(ret)
        jerror(&err, "Failed to parse configuration of node %s", node_name(n));

    // Translate IP:PORT to a struct addrinfo
    char* ip_adr = strtok(local, ":");
    char* port = strtok(NULL, ":");
    ret = getaddrinfo(ip_adr, port, NULL, &ib->conn.src_addr);
    if(ret)
    {
        error("Failed to resolve local address '%s' of node %s: %s",
            local, node_name(n), gai_strerror(ret));
    }

    // Translate port space
    if(strcmp(port_space, "RDMA_PS_IPOIB") == 0)    ib->conn.port_space = RDMA_PS_IPOIB;
    else if(strcmp(port_space, "RDMA_PS_TCP") == 0) ib->conn.port_space = RDMA_PS_TCP;
    else if(strcmp(port_space, "RDMA_PS_UDP") == 0) ib->conn.port_space = RDMA_PS_UDP;
    else if(strcmp(port_space, "RDMA_PS_IB") == 0)  ib->conn.port_space = RDMA_PS_IB;
    else {
        error("Failed to translate rdma_port_space in node %s. %s is not a valid \
            port space supported by rdma_cma.h!", node_name(n), port_space);
    }

    // Set timeout
    ib->conn.timeout = timeout;

    // Translate poll mode
    if(strcmp(poll_mode, "EVENT") == 0)
    {
        ib->poll.poll_mode = EVENT;
        ib->poll.poll_func = ib_event_thread;

    }
    else if(strcmp(poll_mode, "BUSY") == 0)
    {
        ib->poll.poll_mode = BUSY;
        ib->poll.poll_func = ib_busy_poll_thread;
    }
    else
    {
        error("Failed to translate poll_mode in node %s. %s is not a valid \
            poll mode!", node_name(n), poll_mode);
    }

    // Set completion queue size
    ib->cq_size = cq_size;

    // Translate QP type
    if(strcmp(qp_type, "IBV_QPT_RC") == 0)    ib->qp_init.qp_type = IBV_QPT_RC;
    else if(strcmp(qp_type, "IBV_QPT_UC") == 0) ib->qp_init.qp_type = IBV_QPT_UC;
    else if(strcmp(qp_type, "IBV_QPT_UD") == 0) ib->qp_init.qp_type = IBV_QPT_UD;
    else {
        error("Failed to translate qp_type in node %s. %s is not a valid \
            qp_type!", node_name(n), qp_type);
    }

    // Set max. send and receive Work Requests
    // First check if the set value is a power of 2, and warn the user if this is not the case
    int max_send_pow = (int) pow(2, ceil(log2(max_send_wr)));
    int max_recv_pow = (int) pow(2, ceil(log2(max_recv_wr)));

    if(max_send_wr != max_send_pow)
        warn("Max. number of send WRs (%i) is not a power of 2! The HCA will change this to a power of 2: %i",
                max_send_wr, max_send_pow);

    if(max_recv_wr != max_recv_pow)
        warn("Max. number of recv WRs (%i) is not a power of 2! The HCA will change this to a power of 2: %i",
                max_recv_wr, max_recv_pow);

    ib->qp_init.cap.max_send_wr = max_send_wr;
    ib->qp_init.cap.max_recv_wr = max_recv_wr;

    // Set used receive Work Requests to 0
    ib->conn.used_recv_wrs = 0;

    // Set remaining QP attributes
    ib->qp_init.cap.max_send_sge = 1;
    ib->qp_init.cap.max_recv_sge = 1;

    //Check if node is a source and connect to target
    if(remote)
    {
        ib->is_source = 1;

        // Translate address info
        char* ip_adr = strtok(remote, ":");
        char* port = strtok(NULL, ":");
        ret = getaddrinfo(ip_adr, port, NULL, &ib->conn.dst_addr);
        if(ret)
        {
            error("Failed to resolve remote address '%s' of node %s: %s",
                remote, node_name(n), gai_strerror(ret));
        }

        // Set correct Work Completion function
        ib->poll.on_compl = ib_completion_source;
    }
    else
    {
        ib->is_source = 0;

        // Set correct Work Completion function
        ib->poll.on_compl = ib_completion_target;
    }

    return 0;
}

char * ib_print(struct node *n)
{
    return 0;
}

int ib_destroy(struct node *n)
{
    return 0;
}

void * ib_disconnect_thread(void *n)
{
    struct node *node = (struct node *)n;
    struct infiniband *ib = (struct infiniband *)((struct node *)n)->_vd;
    struct rdma_cm_event *event;

    while(rdma_get_cm_event(ib->ctx.ec, &event) == 0)
    {
        if(event->event == RDMA_CM_EVENT_DISCONNECTED)
        {
            rdma_ack_cm_event(event);
            ib->conn.rdma_disconnect_called = 1;

            node_stop(node);
            return NULL;
        }
    }
    return NULL;
}

int ib_start(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    struct rdma_cm_event *event = NULL;
    int ret;

    // Create event channel
    ib->ctx.ec = rdma_create_event_channel();
    if(!ib->ctx.ec) {
        error("Failed to create event channel in node %s!",
            node_name(n));
    }

    ret = rdma_create_id(ib->ctx.ec, &ib->ctx.id, NULL, ib->conn.port_space);
    if(ret)
    {
        error("Failed to create rdma_cm_id of node %s: %s",
            node_name(n), gai_strerror(ret));
    }
    info("Succesfully created rdma_cm_id.");

    // Bind rdma_cm_id to the HCA
    ret = rdma_bind_addr(ib->ctx.id, ib->conn.src_addr->ai_addr);
    if(ret)
    {
        error("Failed to bind to local device of node %s: %s",
            node_name(n), gai_strerror(ret));
    }
    info("Bound rdma_cm_id to Infiniband device.");

    if(ib->is_source)
    {
        // Resolve address
        ret = rdma_resolve_addr(ib->ctx.id,
                                NULL,
                                ib->conn.dst_addr->ai_addr,
                                ib->conn.timeout);
        if(ret)
        {
            error("Failed to resolve remote address after %ims of node %s: %s",
                ib->conn.timeout, node_name(n), gai_strerror(ret));
        }
    }
    else
    {
        // The ID will be overwritten for the target. If the event type is
        // RDMA_CM_EVENT_CONNECT_REQUEST, >then this references a new id for
        // that communication.
        ib->ctx.listen_id = ib->ctx.id;

        // Listen on rdma_cm_id for events
        ret = rdma_listen(ib->ctx.listen_id, 10);
        if(ret)
        {
            error("Failed to listen to rdma_cm_id on node %s", node_name(n));
        }
    }

    // Several events should occur on the event channel, to make
    // sure the nodes are succesfully connected.
    info("Starting to monitor events on rdma_cm_id.");

    while(rdma_get_cm_event(ib->ctx.ec, &event) == 0)
    {
        struct rdma_cm_event event_copy;
        memcpy(&event_copy, event, sizeof(*event));

        rdma_ack_cm_event(event);

        if(ib_event(n, &event_copy))
            break;
    }

    ret = pthread_create(&ib->conn.stop_thread, NULL, ib_disconnect_thread, n);
    if(ret)
    {
        error("Failed to create thread to monitor disconnects in node %s: %s",
            node_name(n), gai_strerror(ret));
    }

    return 0;
}

int ib_stop(struct node *n)
{
    struct infiniband *ib = (struct infiniband *) n->_vd;
    struct rdma_cm_event *event = NULL;
    int ret;

    // Call RDMA disconnect function
    // Will flush all outstanding WRs to the Completion Queue and
    // will call RDMA_CM_EVENT_DISCONNECTED if that is done.
    ret = rdma_disconnect(ib->ctx.id);
    if(ret)
    {
        error("Error while calling rdma_disconnect in node %s: %s",
            node_name(n), gai_strerror(ret));
    }
    info("Called rdma_disconnect.");

    // If disconnected event already occured, directly call cleanup function
    if(ib->conn.rdma_disconnect_called)
    {
        ib_cleanup(n);
    }
    // Else, wait for event to occur
    else
    {
        ib->conn.rdma_disconnect_called = 1;
        rdma_get_cm_event(ib->ctx.ec, &event);

        rdma_ack_cm_event(event);

        ib_event(n, event);
    }

    return 0;
}

int ib_init(struct super_node *n)
{

    return 0;
}

int ib_deinit()
{
    return 0;
}

int ib_read(struct node *n, struct sample *smps[], unsigned cnt)
{
	//Create separate thread for polling! This impelemtation is just
	//for testing purposes
	struct infiniband *ib = (struct infiniband *) n->_vd;
	int ret;
	struct ibv_wc wc[cnt];
	char *ptr;

	ret = ibv_poll_cq(ib->ctx.cq, cnt, wc);

	if(ret)
	{
	    ib->conn.used_recv_wrs += ret;

	    for(int i=0; i<ret; i++)
	    {
	        if(wc[i].status == IBV_WC_WR_FLUSH_ERR)
	            return 0;

	        if(wc[i].status != IBV_WC_SUCCESS)
	        {
	            warn("Work Completion status was not IBV_WC_SUCCES in node %s", node_name(n));
	            ret--;
	        }
	        else
	        {
	            //Copy Data
	            ptr = (char*)(wc[i].wr_id);

	            //Release memory
	            pool_put(&ib->mem.p_recv, (double*)(wc[i].wr_id));
	        }
		smps[i]->length = wc[i].byte_len/sizeof(double);
	    	smps[i]->capacity = 64;
	    	memcpy(smps[i]->data, ptr, wc[i].byte_len);
	    }
	}
	else
	{
	    //No data received? Put new receive Work Requests to Receive Queue
	    for(int i=0; i<ib->conn.used_recv_wrs; i++)
	        ib_post_recv_wrs(n);

	    ib->conn.used_recv_wrs = 0;
	}

	return ret;
}

int ib_write(struct node *n, struct sample *smps[], unsigned cnt)
{
	struct infiniband *ib = (struct infiniband *) n->_vd;
	struct ibv_send_wr wr[cnt], *bad_wr = NULL;
	struct ibv_sge sge[cnt];
	struct pool *p;
	struct ibv_mr ** mr;
	int ret;

	memset(&wr, 0, sizeof(wr));

	//ToDo: Place this into configuration and create checks if settings are valid
	int send_inline = 0;

	// Get Memory Region
	p = sample_pool(smps[0]);
	mr = (struct ibv_mr **)((char *)(p)+p->buffer_off-8);

	for(int i=0; i<cnt; i++)
	{
		// Increase refcnt of sample
		sample_get(smps[i]);

		//Set Scatter/Gather element to data of sample
		sge[i].addr = (uint64_t)&smps[i]->data->f;
		sge[i].length = smps[i]->length*sizeof(double);
		sge[i].lkey = (*mr)->lkey;

		// Set Send Work Request
		wr[i].wr_id = (uint64_t)&smps[i]; //This way the sample can be release in WC
		wr[i].sg_list = &sge[i];
		wr[i].num_sge = 1;

		if(i == (cnt-1))
		    wr[i].next = NULL;
		else
		    wr[i].next = &wr[i+1];

		wr[i].send_flags = IBV_SEND_SIGNALED | (send_inline<<3);
		wr[i].imm_data = htonl(0); //ToDo: set this to a useful value
		wr[i].opcode = IBV_WR_SEND_WITH_IMM;
	}

	//Send linked list of Work Requests
	ret = ibv_post_send(ib->ctx.id->qp, wr, &bad_wr);
	if(ret)
	{
	    error("Failed to send message in node %s: %s",
	        node_name(n), gai_strerror(ret));

	    return -ret;
	}

	return cnt;
}

int ib_fd(struct node *n)
{
    return 0;
}

static struct plugin p = {
    .name           = "infiniband",
    .description    = "Infiniband",
    .type           = PLUGIN_TYPE_NODE,
    .node           = {
        .vectorize      = 0,
        .size           = sizeof(struct infiniband),
        .reverse        = ib_reverse,
        .parse          = ib_parse,
        .print          = ib_print,
        .start          = ib_start,
        .destroy        = ib_destroy,
        .stop           = ib_stop,
        .init           = ib_init,
        .deinit         = ib_deinit,
        .read           = ib_read,
        .write          = ib_write,
        .fd             = ib_fd,
        .memtype        = ib_memtype
    }
};

REGISTER_PLUGIN(&p)
LIST_INIT_STATIC(&p.node.instances)