RDMAQueue.cc

Go to the documentation of this file.

//=============================================================================
//  Computational Process Networks class library
//  Copyright (C) 1997-2006  Gregory E. Allen and The University of Texas
//
//  This library is free software; you can redistribute it and/or modify it
//  under the terms of the GNU Library General Public License as published
//  by the Free Software Foundation; either version 2 of the License, or
//  (at your option) any later version.
//
//  This library 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
//  Library General Public License for more details.
//
//  The GNU Public License is available in the file LICENSE, or you
//  can write to the Free Software Foundation, Inc., 59 Temple Place -
//  Suite 330, Boston, MA 02111-1307, USA, or you can find it on the
//  World Wide Web at http://www.fsf.org.
//=============================================================================
/*
 * Some comments about how things work:
 *
 * The queue endpoint starts up in the S_INIT state. After the infiniband
 * connection has completely come up we transition to the S_RUNNING state.
 * When one of the endpoints shutdown it goes into the S_FLUSHING state if the
 * other end is not already dead.  The flushing state flushes all communication
 * between the two endpoints and then goes through the S_DIEING state then
 * finally ends in the S_DEAD state when we know that the other end will not
 * send any more command.
 *
 * Note S_FLUSHING has nothing to do with the flush and reset functionality
 * of the queue. S_FLUSHING has to do with the proper shutdown of the queue only.
 *
 * Comments on growing the queue:
 *
 * We cannot just copy the method from the RemoteQueue as the write end actually
 * has address information about the read ends queue buffers. The write end
 * must flush its RDMA operations before we can fully go over to the other
 * side. Lets go through a walk through of how things should proceed from both
 * sides.
 *
 * There are problems if both sides try to grow at the time time, so, we have
 * to do some sort of mutual exclusion. The function EnterCriticalSection and
 * LeaveCriticalSection implement Ricart and Agrawala's mutual exclusion
 * algorithm.
 *
 * Grow initiated from read side:
 *
 * Read side:
 *  * Send grow command to write side.
 *  * Wait for acknowledgement from write side.
 *  * Create new buffer, copy data over, send command to write side for new buffer information
 *  * Lazy cleanup
 * Write side:
 *  * When receive grow command, finish up pending enqueues but initiate no new ones
 *  * When finished, create new buffer, copy data over, and send acknowledgement.
 *  * Wait for new queue information, when got start up enqueues again.
 *  * Lazy cleanup
 *
 * Grow initiated from write side:
 * 
 * Write side:
 *  * Create new buffer, copy data over, finish up pending enqueues but initiate no new ones
 *  * When finished, Send grow command to read side.
 *  * Wait for acknowledgement with new queue data then proceed with enqueues
 *  * Lazy cleanup
 * Read side:
 *  * When receive grow command, Create new buffer, copy data over, send acknowledgement with queue data
 *  * Lazy cleanup
 */
#include "common_priv.h"
#include "RDMAQueue.h"
#include <cpn/QueueAttr.h>
#include "RemoteQueueHolder.h"
#include <cpn/Exceptions.h>
#include <cpn/bits/KernelBase.h>
#include "ConnectionServer.h"
#include <cpn/utils/ThrowingAssert.h>
#include <cpn/threading/PthreadFunctional.h>
#include <cpn/io/SocketHandle.h>
#include <cpn/d4r/D4RNode.h>
#include <stdexcept>
#include <vector>
#include <map>
#include <algorithm>
#include <sys/select.h>

#if ENABLE_RDMA

#if RDMATRACE
#define TRACE(fmt, ...) logger.Trace(fmt, ## __VA_ARGS__)
#else
#define TRACE(fmt, ...)
#endif

#define IBV_ACCESS_DEFAULT (IBV_ACCESS_LOCAL_WRITE | IBV_ACCESS_REMOTE_WRITE\
        | IBV_ACCESS_REMOTE_READ)

using std::max;
using std::min;

namespace CPN {
    namespace Internal {

        CommandQueue::CommandQueue(uint64_t size)
            : outstanding_commands(0)
        {
            queue.base = (char*)malloc(size*sizeof(Command));
            ASSERT(queue.base);
            queue.length = size*sizeof(Command);
        }

        CommandQueue::~CommandQueue() {
            free(queue.base);
            queue.base = 0;
        }

        Command *CommandQueue::NewCommand() {
            if (Full()) return 0;
            Command *ptr = (Command*)queue.GetHead(outstanding_commands*sizeof(Command));
            memset(ptr, 0, sizeof(Command));
            outstanding_commands++;
            return ptr;
        }

        void CommandQueue::Enqueue() {
            queue.AdvanceHead(sizeof(Command));
            --outstanding_commands;
        }

        Command *CommandQueue::GetCommand() {
            if (Empty()) return 0;
            return (Command*)queue.GetTail();
        }

        void CommandQueue::Dequeue() {
            queue.AdvanceTail(sizeof(Command));
        }

        void CommandQueue::Reset() {
            outstanding_commands = 0;
            queue.head = 0;
            queue.tail = 0;
        }

        const char* StateStr(State s) {
            switch (s) {
            case S_INIT: return "INIT";
            case S_RUNNING: return "RUNNING";
            case S_FLUSHING: return "FLUSHING";
            case S_DIEING: return "DIEING";
            case S_DEAD: return "DEAD";
            case S_ERROR: return "ERROR";
            }
            return "UNKNOWN";
        }

        const char* CommandTypeStr(CommandType ct) {
            switch (ct) {
            case CT_UPDATE: return "UPDATE";
            case CT_ENQUEUE: return "ENQUEUE";
            case CT_DEQUEUE: return "DEQUEUE";
            case CT_BLOCK: return "BLOCK";
            case CT_SHUTDOWN: return "SHUTDOWN";
            case CT_GROW: return "GROW";
            case CT_D4R: return "D4R";
            case CT_CS_REQ: return "CS_REQ";
            case CT_CS_OK: return "CS_OK";
            case CT_FLUSH: return "FLUSH";
            case CT_RESET: return "RESET";
            }
            return "UNKNOWN";
        }
    }

    using namespace Internal;

    unsigned ComputeQueueLength(unsigned length, unsigned maxthresh, double alpha, RDMAQueue::Mode mode) {
        unsigned writerlen = unsigned(((double)length)*alpha);
        if (mode == RDMAQueue::READ) {
            return std::max<unsigned>(length - writerlen, maxthresh);
        } else {
            return std::max<unsigned>(writerlen, maxthresh);
        }
    }

    void ParseQueueHint(const std::string &qhint, std::map<std::string, std::string> &result) {
        typedef std::string::const_iterator stritr;
        stritr cur = qhint.begin();
        stritr end = qhint.end();
        std::string key;
        std::string value;
        bool expect_key = true;
        while (cur != end) {
            switch (*cur) {
            case ' ':
            case '\t':
            case '\v':
            case '\n':
            case '\r':
                break;
            case ':':
                expect_key = false;
                break;
            case ',':
                result.insert(std::make_pair(key, value));
                expect_key = true;
                break;
            default:
                if (expect_key) {
                    key.push_back(*cur);
                } else {
                    value.push_back(*cur);
                }
            }
            ++cur;
        }
    }

    RDMAQueue::RDMAQueue(KernelBase *k, Mode m,
                ConnectionServer *s, RemoteQueueHolder *h, const SimpleQueueAttr &attr)
        : QueueBase(k, attr),
        mode(m),
        state(S_INIT),
        read_command_queue(std::max(attr.GetNumChannels()*2, 100u)),
        write_command_queue(std::max(attr.GetNumChannels()*2, 100u)),
        max_threshold(attr.GetMaxThreshold()),
        num_channels(attr.GetNumChannels()),
        num_enqueued(0),
        num_dequeued(0),
        alpha(attr.GetAlpha()),
        max_send_wr(std::max(attr.GetNumChannels()*2, 100u)),
        max_recv_wr(std::max(attr.GetNumChannels()*2, 100u)),
        num_send_wr(0),
        pending_dequeue(0),
        pending_rdma(0),
        confirmed_rdma(0),
        cork(0),
        queue_updated(false),
        pending_grow(false),
        grow_ack(false),
        pending_flush(false),
        clock(0), remote_clock(0), old_clock(0),
        local_cs_ts(0),
        local_want_cs(false),
        remote_want_cs(false),
        cs_ok(false),
        if_name(),
        if_port(0),
        enqueue_use_old(false),
        dequeue_use_old(false),
        grow_length(0), grow_thresh(0),
        server(s),
        holder(h),
        mocknode(new D4R::Node(m == READ ? attr.GetWriterNodeKey() : attr.GetReaderNodeKey())),
        ctx(0), pd(0), cc(0), cq(0), qp(0), mbs_mr(0), mbs_old_mr(0),
        read_command_queue_mr(0), write_command_queue_mr(0)
    {
        if (attr.GetHint().size() > 4) {
            std::map<std::string, std::string> options;
            ParseQueueHint(attr.GetHint().substr(4), options);
            if (options.find("if_name") != options.end()) {
                if_name = options["if_name"];
            }
            if (options.find("if_port") != options.end()) {
                std::istringstream iss(options["if_port"]);
                iss >> if_port;
                if (!iss) {
                    if_port = 0;
                }
            }
        }

        std::ostringstream oss;
        oss << "RDMAQueue(m:";
        if (mode == READ) { oss << "r"; }
        else { oss << "w"; }
        oss << ", r:" << readerkey << ", w:" << writerkey << ")";
        logger.Name(oss.str());
        if (mode == READ) {
            SetWriterNode(mocknode);
        } else {
            SetReaderNode(mocknode);
        }
        local_queue.length = ComputeQueueLength(attr.GetLength(), max_threshold, alpha, mode);
        remote_queue.length = 0;
        mbs.reset(new MirrorBufferSet(local_queue.length, max_threshold, num_channels));
        local_queue.length = mbs->BufferSize();
        local_queue.base = (char*)(void*)*mbs;
        local_queue.stride = mbs->BufferSize() + mbs->MirrorSize();
        max_threshold = std::min(mbs->MirrorSize() + 1, local_queue.length);
        event_thread.reset(CreatePthreadFunctional(this, &RDMAQueue::EventThread));
        try {
            Initialize();
            mbs_mr = ibv_reg_mr(pd, local_queue.base, local_queue.stride*num_channels,
                    IBV_ACCESS_DEFAULT);
            ASSERT(mbs_mr);
            local_queue.lkey = mbs_mr->lkey;
            logger.Trace("mbs constructed: base: %p, length: %llu, lkey: %llu",
                    local_queue.base, (unsigned long long)local_queue.length,
                    (unsigned long long)local_queue.lkey);

            read_command_queue_mr = ibv_reg_mr(pd, read_command_queue.GetBase(),
                    read_command_queue.Size()*sizeof(Command), IBV_ACCESS_LOCAL_WRITE);
            ASSERT(read_command_queue_mr);
            read_command_queue.SetKey(read_command_queue_mr->lkey);
            write_command_queue_mr = ibv_reg_mr(pd, write_command_queue.GetBase(),
                    write_command_queue.Size()*sizeof(Command), 0);
            ASSERT(write_command_queue_mr);
            write_command_queue.SetKey(write_command_queue_mr->lkey);
        } catch (...) {
            Cleanup();
            throw;
        }
        logger.Trace("Constructed");
    }

    RDMAQueue::~RDMAQueue() {
        Shutdown();
        event_thread->Join();
        logger.Trace("%s %s Destructed", StateStr(state), UnlockedClockStr());
        Cleanup();
    }

    void RDMAQueue::Start() {
        event_thread->Start();
    }

    void RDMAQueue::Shutdown() {
        AutoLock<QueueBase> al(*this);
        logger.Trace("%s %s Shutdown", StateStr(state), UnlockedClockStr());
        if (UnlockedIsDead()) return;
        if (mode == WRITE) {
            UnlockedShutdownWriter();
        } else {
            UnlockedShutdownReader();
        }
    }

    void RDMAQueue::LogState() {
        QueueBase::LogState();
        logger.Error("State: %s, Clock: %s", StateStr(state), UnlockedClockStr());
        logger.Error("local_queue: size: %llu, count: %llu, remote_queue: size: %llu, count: %llu, lstride: %llu",
                (unsigned long long)local_queue.Size(),
                (unsigned long long)local_queue.Count(),
                (unsigned long long)remote_queue.Size(),
                (unsigned long long)remote_queue.Count(),
                (unsigned long long)local_queue.stride
                );
        logger.Error("num_send_wr: %u, pending_rdma: %u, confirmed_rdma: %u",
                num_send_wr, pending_rdma, confirmed_rdma);
        logger.Error("cork: %u, queue_updated: %d, pending_grow: %d, grow_ack: %d"
                ", pending_flush: %d",
                cork, (int)queue_updated, (int)pending_grow, (int)grow_ack, (int)pending_flush);
        logger.Error("local_cs_ts: %llu, local_want_cs: %d, remote_want_cs: %d, cs_ok: %d",
                (unsigned long long)local_cs_ts, (int)local_want_cs, (int)remote_want_cs,
                (int)cs_ok);
    }

    void RDMAQueue::WaitForData() {
        ASSERT(mode == READ);
        ASSERT(!writeshutdown);
        if (state == S_INIT) {
            while (state == S_INIT && !readshutdown && !writeshutdown && ReadBlocked()) { Wait(); }
            return;
        }
        try {
            while (ActiveTransfers() && ReadBlocked()) { Wait(); }
            if (!ReadBlocked()) return;
            PostBlock(readrequest);
            uint64_t sync_clock = clock;
            while ((ActiveTransfers() || sync_clock > old_clock) && WriteBlocked()) { Wait(); }
            if (ReadBlocked()) { QueueBase::WaitForData(); }
        } catch (const BrokenQueueException &e) {}
    }

    void RDMAQueue::WaitForFreespace() {
        ASSERT(mode == WRITE);
        ASSERT(!readshutdown);
        if (state == S_INIT) {
            while (state == S_INIT && !readshutdown && !writeshutdown && WriteBlocked()) { Wait(); }
        }
        try {
            CheckRDMAWrite();
            while (ActiveTransfers() && WriteBlocked()) { Wait(); }
            if (!WriteBlocked()) return;
            PostBlock(writerequest);
            uint64_t sync_clock = clock;
            while ((ActiveTransfers() || sync_clock > old_clock) && WriteBlocked()) { Wait(); }
            if (WriteBlocked()) { QueueBase::WaitForFreespace(); }
        } catch (const BrokenQueueException &e) {}
    }

    const void *RDMAQueue::InternalGetRawDequeuePtr(unsigned thresh, unsigned chan) {
        ASSERT(mode == READ);
        ASSERT(chan < num_channels);
        if (dequeue_use_old) {
            ASSERT(indequeue);
            return (const void*)(local_old_queue.GetTail() + chan*UnlockedDequeueChannelStride()); 
        } else {
            if (thresh > max_threshold) return 0;
            if (thresh > local_queue.Count()) return 0;
            return (const void*)(local_queue.GetTail() + chan*UnlockedDequeueChannelStride());
        }
    }

    void RDMAQueue::InternalDequeue(unsigned count) {
        ASSERT(mode == READ);
        ASSERT(count <= local_queue.Count());
        if (dequeue_use_old) {
            dequeue_use_old = false;
            ibv_dereg_mr(mbs_old_mr);
            mbs_old_mr = 0;
            mbs_old.reset();
        }
        local_queue.AdvanceTail(count);
        pending_dequeue += count;
        CheckDequeue();
        num_dequeued += count;
    }

    void *RDMAQueue::InternalGetRawEnqueuePtr(unsigned thresh, unsigned chan) {
        ASSERT(mode == WRITE);
        ASSERT(chan < num_channels);
        if (enqueue_use_old) {
            ASSERT(inenqueue);
            return (void*)(local_old_queue.GetHead() + chan*UnlockedEnqueueChannelStride()); 
        } else {
            if (thresh > max_threshold) return 0;
            if (thresh > local_queue.Freespace()) return 0;
            return (void*)(local_queue.GetHead() + chan*UnlockedEnqueueChannelStride());
        }
    }

    void RDMAQueue::InternalEnqueue(unsigned count) {
        ASSERT(mode == WRITE);
        ASSERT(count <= local_queue.Freespace());
        if (enqueue_use_old) {
            for (unsigned chan = 0; chan < num_channels; ++chan) {
                memcpy((local_queue.GetHead() + chan*local_queue.stride),
                        (local_old_queue.GetHead() + chan*local_old_queue.stride),
                        count);
            }
            enqueue_use_old = false;
            ibv_dereg_mr(mbs_old_mr);
            mbs_old_mr = 0;
            mbs_old.reset();
        }
        local_queue.AdvanceHead(count);
        CheckRDMAWrite();
        num_enqueued += count;
    }

    void RDMAQueue::InternalFlush() {
        ASSERT(mode == WRITE);
        QueueBase::InternalFlush();
        if (state == S_RUNNING && !ActiveTransfers() && local_queue.Empty()) {
            PostFlush();
        } else {
            pending_flush = true;
        }
    }

    void RDMAQueue::InternalReset() {
        ASSERT(mode == READ);
        ASSERT(flushed);
        readrequest = 0;
        writerequest = 0;
        enqueuethresh = 0;
        dequeuethresh = 0;
        indequeue = false;
        inenqueue = false;
        flushed = false;
        while (state != S_RUNNING && !UnlockedIsDead() && !writeshutdown && !readshutdown) { Wait(); }
        if (UnlockedIsDead() || writeshutdown || readshutdown) return;
        ASSERT(num_enqueued >= num_dequeued);
        while (num_enqueued > num_dequeued) {
            while (local_queue.Count() == 0 && !(UnlockedIsDead() || writeshutdown || readshutdown)) { Wait(); }
            if (UnlockedIsDead() || writeshutdown || readshutdown) return;
            InternalDequeue(local_queue.Count());
        }
        PostReset();
        num_enqueued = 0;
        num_dequeued = 0;
    }

    unsigned RDMAQueue::UnlockedEnqueueChannelStride() const {
        if (enqueue_use_old) {
            return local_old_queue.stride;
        } else {
            return local_queue.stride;
        }
    }

    unsigned RDMAQueue::UnlockedDequeueChannelStride() const {
        if (dequeue_use_old) {
            return local_old_queue.stride;
        } else {
            return local_queue.stride;
        }
    }

    void RDMAQueue::UnlockedGrow(unsigned queueLen, unsigned maxThresh) {
        queueLen = max(queueLen, maxThresh);
        if (queueLen < UnlockedQueueLength() && maxThresh < UnlockedMaxThreshold()) return;
        while (state != S_RUNNING && !UnlockedIsDead() && !writeshutdown && !readshutdown) { Wait(); }
        if (UnlockedIsDead() || writeshutdown || readshutdown) return;
        EnterCriticalSection();
        if (queueLen <= UnlockedQueueLength() && maxThresh <= UnlockedMaxThreshold()) {
            LeaveCriticalSection();
            return;
        }
        try {
            logger.Trace("Growing %s %s from %llu %llu to %llu %llu", UnlockedClockStr(), StateStr(state),
                    (unsigned long long)UnlockedQueueLength(),
                    (unsigned long long)UnlockedMaxThreshold(),
                    (unsigned long long)queueLen,
                    (unsigned long long)maxThresh
                    );
            grow_length = max(grow_length, (uint64_t)queueLen);
            grow_thresh = max(grow_thresh, (uint64_t)maxThresh);
            Cork();
            if (mode == WRITE) {
                while (ActiveTransfers() && UnlockedIsNormal()) { Wait(); }
                if (!UnlockedIsNormal()) { return; }
                PostGrow();
                queue_updated = false;
                uint64_t sync_clock = clock;
                InternalGrow();
                PostUpdateQueue();
                while (sync_clock > old_clock && UnlockedIsNormal()) {
                    while (!queue_updated && UnlockedIsNormal()) { Wait(); }
                    queue_updated = false;
                }
            } else {
                PostGrow();
                uint64_t sync_clock = clock;
                while (sync_clock > old_clock && UnlockedIsNormal()) {
                    queue_updated = false;
                    while (!queue_updated && UnlockedIsNormal()) { Wait(); }
                }
                if (!UnlockedIsNormal()) { return; }
                InternalGrow();
                PostUpdateQueue();
            }
            Uncork();
        } catch (...) {
            LeaveCriticalSection();
            throw;
        }
        LeaveCriticalSection();
    }

    void RDMAQueue::UnlockedShutdownReader() {
        QueueBase::UnlockedShutdownReader();
        if (UnlockedIsNormal()) {
            PostShutdown();
            State oldstate = state;
            if (writeshutdown) {
                state = S_DEAD;
            } else {
                state = S_FLUSHING;
            }
            logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                    StateStr(oldstate), StateStr(state));
        }
    }

    void RDMAQueue::UnlockedShutdownWriter() {
        QueueBase::UnlockedShutdownWriter();
        if (UnlockedIsNormal()) {
            State oldstate = state;
            state = S_FLUSHING;
            logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                    StateStr(oldstate), StateStr(state));
            if (local_queue.Empty() && !ActiveTransfers()) {
                oldstate = state;
                PostShutdown();
                state = S_DIEING;;
                logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                        StateStr(oldstate), StateStr(state));
            }
        }
    }

    void RDMAQueue::UnlockedSignalReaderTagChanged() {
        ASSERT(mode == READ);
        if (UnlockedIsNormal()) {
            PostD4RTag();
        }
        QueueBase::UnlockedSignalReaderTagChanged();
    }

    void RDMAQueue::UnlockedSignalWriterTagChanged() {
        ASSERT(mode == WRITE);
        if (UnlockedIsNormal()) {
            PostD4RTag();
        }
        QueueBase::UnlockedSignalWriterTagChanged();
    }

    void *RDMAQueue::EventThread() {
        try {
            Connect();
            std::vector<ibv_wc> wcs(max_send_wr + max_recv_wr);
            int wake_fd = holder->GetWakeup()->FD();
            while (!IsDead()) {
                kernel->CheckTerminated();
                fd_set fdsr;
                FD_ZERO(&fdsr);
                int maxfd = max(cc->fd, wake_fd) + 1;
                FD_SET(cc->fd, &fdsr);
                FD_SET(wake_fd, &fdsr);
                select(maxfd, &fdsr, 0, 0, 0);
                if (FD_ISSET(cc->fd, &fdsr)) {
                    ibv_cq *l_cq;
                    void *context;
                    if (ibv_get_cq_event(cc, &l_cq, &context)) {
                        throw ErrnoException();
                    }
                    ibv_req_notify_cq(cq, 0);
                    while (true) {
                        int num = ibv_poll_cq(cq, wcs.size(), &wcs[0]);
                        for (int i = 0; i < num; ++i) {
                            ibv_wc *wc = &wcs[i];
                            if (wc->status != IBV_WC_SUCCESS) {
                                logger.Error("%s A work request did not complete, aborting!!!\n %s",
                                        ClockStr(), ibv_wc_status_str(wc->status));
                                switch (wc->opcode) {
                                case IBV_WC_SEND:
                                    if (write_command_queue.GetCommand()) {
                                        logger.Error("Error with sending a %s command",
                                                CommandTypeStr(write_command_queue.GetCommand()->type));
                                    } else {
                                        logger.Error("No commands in write command queue?!?!?");
                                    }
                                    break;
                                case IBV_WC_RDMA_WRITE:
                                    logger.Error("Error with RDMA write");
                                    break;
                                case IBV_WC_RECV:
                                    logger.Error("Error on command receive");
                                    break;
                                default:
                                    logger.Error("opcode: %d", (int)wc->opcode);
                                    break;
                                }
                                ShutdownError();
                            } else {
                                HandleWorkCompletion(wc);
                            }
                        }
                        ProcessEvents();
                        if (num < (int)wcs.size()) break;
                    }
                    ibv_ack_cq_events(l_cq, 1);
                }
            }
        } catch (const ShutdownException &e) {
            {
                AutoLock<QueueBase> al(*this);
                if (!UnlockedIsDead()) { state = S_DEAD; }
                Signal();
            }
            logger.Debug("%s Forced Shutdown threw exception", ClockStr());
        } catch (const ErrnoException &e) {
            logger.Error("%s: %s", ClockStr(), e.what());
            ShutdownError();
        } catch (...) {
            ShutdownError();
            holder->CleanupQueue(GetKey());
            throw;
        }
        holder->CleanupQueue(GetKey());
        server->Wakeup();
        return 0;
    }

    void RDMAQueue::PostDequeue(unsigned count) {
        ASSERT(mode == READ);
        ASSERT(UnlockedIsNormal());
        Command *command = NewCommand(CT_DEQUEUE);
        command->data.dequeue.amount = count;
        PostCommand(command);
        TRACE("%s Posting dequeue %u", UnlockedClockStr(), count);
    }

    void RDMAQueue::PostEnqueue(unsigned count) {
        ASSERT(mode == WRITE);
        ASSERT(UnlockedIsNormal() || state == S_FLUSHING);
        Command *command = NewCommand(CT_ENQUEUE);
        command->data.enqueue.amount = count;
        PostCommand(command);
        TRACE("%s Posting enqueue %u", UnlockedClockStr(), count);
    }

    void RDMAQueue::PostBlock(unsigned count) {
        ASSERT(UnlockedIsNormal());
        Command *command = NewCommand(CT_BLOCK);
        command->data.block.amount = count;
        PostCommand(command);
        TRACE("%s Posting block %u", UnlockedClockStr(), count);
    }

    void RDMAQueue::PostShutdown() {
        Command *command = NewCommand(CT_SHUTDOWN);
        PostCommand(command);
        TRACE("%s Posting shutdown", UnlockedClockStr());
    }

    void RDMAQueue::PostGrow() {
        Command *command = NewCommand(CT_GROW);
        command->data.grow.length = grow_length;
        command->data.grow.threshold = grow_thresh;
        PostCommand(command);
        TRACE("%s Posted grow len: %u, thresh: %u", UnlockedClockStr(), grow_length, grow_thresh);
    }

    void RDMAQueue::PostD4RTag() {
        ASSERT(UnlockedIsNormal());
        Command *command = NewCommand(CT_D4R);
        D4R::Tag tag;
        if (mode == READ) {
            tag = reader->GetPublicTag();
        } else {
            tag = writer->GetPublicTag();
        }
        command->data.d4r.count = tag.Count();
        command->data.d4r.key = tag.Key();
        command->data.d4r.qsize = tag.QueueSize();
        command->data.d4r.qkey = tag.QueueKey();
        PostCommand(command);
        TRACE("%s Posting D4R tag count: %llu, key: %llu, qsize: %llu, qkey: %llu",
                UnlockedClockStr(),
                (unsigned long long)tag.Count(),
                (unsigned long long)tag.Key(),
                (unsigned long long)tag.QueueSize(),
                (unsigned long long)tag.QueueKey());
    }

    void RDMAQueue::PostCSRequest() {
        Command *command = NewCommand(CT_CS_REQ);
        command->data.cs_info.timestamp = local_cs_ts;
        PostCommand(command);
        TRACE("%s Posting CS_REQ with ts: %llu", UnlockedClockStr(), (unsigned long long)local_cs_ts);
    }

    void RDMAQueue::PostCSOK() {
        Command *command = NewCommand(CT_CS_OK);
        PostCommand(command);
        TRACE("%s Posting CS_OK", UnlockedClockStr());
    }

    void RDMAQueue::PostFlush() {
        Command *command = NewCommand(CT_FLUSH);
        command->data.flush.num_enqueued = num_enqueued;
        PostCommand(command);
        TRACE("%s Posting FLUSH", UnlockedClockStr());
    }

    void RDMAQueue::PostReset() {
        Command *command = NewCommand(CT_RESET);
        PostCommand(command);
        TRACE("%s Posting RESET", UnlockedClockStr());
    }

    void RDMAQueue::CheckRDMAWrite() {
        ASSERT(mode == WRITE);
        if (readshutdown) return;
        if (state != S_RUNNING && state != S_FLUSHING) return;
        if (cork > 0) return;
        unsigned old_pending_rdma = pending_rdma;
        while (true) {
            if (local_queue.Count() - pending_rdma <= 0) break;
            if (num_send_wr >= max_send_wr/2) break;
            unsigned count = std::min(remote_queue.Freespace() - pending_rdma,
                    local_queue.Count() - pending_rdma);
            count = std::min(count, max_threshold);
            if (count == 0) return;
            std::vector<ibv_send_wr> wr(num_channels);
            memset(&wr[0], 0, num_channels*sizeof(ibv_send_wr));
            std::vector<ibv_sge> sge(num_channels);
            memset(&sge[0], 0, num_channels*sizeof(ibv_sge));
            for (unsigned chan = 0; chan < num_channels; ++chan) {
                sge[chan].addr = (uint64_t)local_queue.GetTail(pending_rdma)
                    + chan*local_queue.stride;
                sge[chan].length = count;
                sge[chan].lkey = local_queue.lkey;
                wr[chan].wr_id = 0;
                wr[chan].sg_list = &sge[chan];
                wr[chan].num_sge = 1;
                wr[chan].opcode = IBV_WR_RDMA_WRITE;
                wr[chan].wr.rdma.remote_addr = (uint64_t)remote_queue.GetHead(pending_rdma)
                    + chan*remote_queue.stride;
                wr[chan].wr.rdma.rkey = remote_queue.lkey;
                if (chan+1 < num_channels) {
                    wr[chan].next = &wr[chan+1];
                }
            }
            wr[num_channels-1].wr_id = count;
            ibv_send_wr *bad;
            if (ibv_post_send(qp, &wr[0], &bad)) {
                throw ErrnoException();
            }
            num_send_wr += num_channels;
            pending_rdma += count;
        }
        if (pending_rdma > old_pending_rdma) {
            TRACE("%s Posted RDMA write, pending_rdma: %llu, to key: %llu", UnlockedClockStr(),
                    (unsigned long long)pending_rdma, (unsigned long long)remote_queue.lkey);
        }
    }

    void RDMAQueue::CheckDequeue() {
        if (pending_dequeue > 0 && num_send_wr < max_send_wr && UnlockedIsNormal() && !writeshutdown) {
            PostDequeue(pending_dequeue);
            pending_dequeue = 0;
        }
    }

    void RDMAQueue::PostUpdateQueue() {
        Command *command = NewCommand(CT_UPDATE);
        command->data.update.base = local_queue.base;
        command->data.update.length = local_queue.length;
        command->data.update.head = local_queue.head;
        command->data.update.tail = local_queue.tail;
        command->data.update.stride = local_queue.stride;
        command->data.update.lkey = local_queue.lkey;
        PostCommand(command);
        TRACE("%s Posting update queue", UnlockedClockStr());
    }

    Command *RDMAQueue::NewCommand(CommandType type) {
        while (num_send_wr >= max_send_wr && !UnlockedIsDead()) {
            ASSERT(!pthread_equal(pthread_self(), (pthread_t)*event_thread),
                    "Blocking in event thread");
            Wait();
        }
        if (UnlockedIsDead()) throw BrokenQueueException(GetKey());
        Command *command = write_command_queue.NewCommand();
        memset(command, 0, sizeof(Command));
        command->type = type;
        clock++;
        command->clock = clock;
        command->remote_clock = remote_clock;
        return command;
    }

    void RDMAQueue::PostCommand(Command *command) {
        if ((mode == READ ? writeshutdown : readshutdown)) {
            ASSERT(false);
            return;
        }
        ASSERT(num_send_wr < max_send_wr);
        ibv_send_wr wr, *bad;
        ibv_sge sge;
        memset(&sge, 0, sizeof(sge));
        sge.addr = (uint64_t)(void*)command;
        sge.length = sizeof(Command);
        sge.lkey = write_command_queue.GetKey();
        memset(&wr, 0, sizeof(wr));
        wr.sg_list = &sge;
        wr.num_sge = 1;
        wr.opcode = IBV_WR_SEND;
        if (ibv_post_send(qp, &wr, &bad)) {
            throw ErrnoException();
        }
        ++num_send_wr;
        write_command_queue.Enqueue();
    }

    void RDMAQueue::PostReceives() {
        uint64_t addr = 0;
        int num = read_command_queue.Freespace();
        if (num <= 0) return;
        std::vector<ibv_recv_wr> wr(num);
        memset(&wr[0], 0, num*sizeof(ibv_recv_wr));
        std::vector<ibv_sge> sge(num);
        memset(&sge[0], 0, num*sizeof(ibv_sge));
        for (int i = 0; i < num; ++i) {
            addr = (uint64_t)(void*)read_command_queue.NewCommand();
            sge[i].addr = addr;
            sge[i].length = sizeof(Command);
            sge[i].lkey = read_command_queue.GetKey();
            wr[i].wr_id = addr;
            wr[i].sg_list = &sge[i];
            wr[i].num_sge = 1;
            wr[i].next = (i + 1 < num ? &wr[i+1] : 0);
        }
        ibv_recv_wr *bad = 0;
        if (ibv_post_recv(qp, &wr[0], &bad)) {
            throw ErrnoException();
        }
    }

    void RDMAQueue::HandleWorkCompletion(ibv_wc *wc) {
        AutoLock<QueueBase> al(*this);
        uint64_t count;
        switch (wc->opcode) {
        case IBV_WC_SEND:
            write_command_queue.Dequeue();
            num_send_wr--;
            Signal();
            break;
        case IBV_WC_RDMA_WRITE:
            ASSERT(mode == WRITE);
            count = wc->wr_id;
            local_queue.AdvanceTail(count);
            remote_queue.AdvanceHead(count);
            pending_rdma -= count;
            confirmed_rdma += count;
            num_send_wr--;
            Signal();
            break;
        case IBV_WC_RECV:
            read_command_queue.Enqueue();
            break;
        default:
            ASSERT(false, "Unknown work completion type: %d", wc->opcode);
            break;
        }
    }

    void RDMAQueue::ProcessEvents() {
        AutoLock<QueueBase> al(*this);
        while (!read_command_queue.Empty()) {
            ProcessCommand(read_command_queue.GetCommand());
            read_command_queue.Dequeue();
        }
        PostReceives();
        if (mode == WRITE && !readshutdown) {
            CheckRDMAWrite();
            if (confirmed_rdma > 0 && num_send_wr < max_send_wr) {
                PostEnqueue(confirmed_rdma);
                confirmed_rdma = 0;
            }
            if (pending_grow && pending_rdma == 0 && num_send_wr < max_send_wr) {
                InternalGrow();
                PostUpdateQueue();
                pending_grow = false;
                grow_ack = true;
            }
            if (pending_flush && !ActiveTransfers() && local_queue.Empty()) {
                PostFlush();
                pending_flush = false;
            }
        } else if (mode == READ && !writeshutdown) {
            CheckDequeue();
        }
        switch (state) {
        case S_INIT:
            break;
        case S_RUNNING:
            if (writeshutdown || readshutdown) {
                logger.Trace("Shutdown? writeshutdown: %d, readshutdown: %d",
                        (int)writeshutdown, (int)readshutdown);
                if (mode == READ) {
                    if (num_send_wr == max_send_wr) { break; }
                    PostShutdown();
                }
                State oldstate = state;
                state = S_FLUSHING;
                logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                        StateStr(oldstate), StateStr(state));
            } else {
                break;
            }
        case S_FLUSHING:
            if (mode == WRITE) {
                State oldstate = state;
                if (readshutdown) {
                    state = S_DEAD;
                    logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                            StateStr(oldstate), StateStr(state));
                } else if (local_queue.Empty() && !ActiveTransfers()) {
                    PostShutdown();
                    state = S_DIEING;;
                    logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                            StateStr(oldstate), StateStr(state));
                }
            } else { // READ
                if (writeshutdown || !ActiveTransfers()) {
                    State oldstate = state;
                    state = S_DEAD;
                    logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                            StateStr(oldstate), StateStr(state));
                }
            }
            break;
        case S_DIEING:
            ASSERT(mode == WRITE);
            if (!ActiveTransfers()) {
                State oldstate = state;
                state = S_DEAD;
                logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                        StateStr(oldstate), StateStr(state));
            }
            break;
        case S_DEAD:
            break;
        case S_ERROR:
            break;
        }
    }

    void RDMAQueue::ProcessCommand(Command *command) {
        clock = std::max(clock, command->clock) + 1;
        remote_clock = command->clock;
        old_clock = command->remote_clock;
        State oldstate = state;
        switch (command->type) {
        case CT_UPDATE:
            remote_queue.base = command->data.update.base;
            remote_queue.length = command->data.update.length;
            if (mode == WRITE) {
                remote_queue.head = command->data.update.head;
                remote_queue.tail = command->data.update.tail;
            }
            remote_queue.stride = command->data.update.stride;
            remote_queue.lkey = command->data.update.lkey;
            queue_updated = true;
            logger.Trace("%s Receved update command: base %p, length: %llu, "
                    "head: %llu, tail: %llu, stride: %llu, lkey: %llu", UnlockedClockStr(),
                    remote_queue.base, (unsigned long long)remote_queue.length,
                    (unsigned long long)remote_queue.head,
                    (unsigned long long)remote_queue.tail,
                    (unsigned long long)remote_queue.stride,
                    (unsigned long long)remote_queue.lkey);
            if (state == S_INIT) {
                state = S_RUNNING;
                logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                        StateStr(oldstate), StateStr(state));
            }
            if (grow_ack && mode == WRITE) {
                Uncork();
                grow_ack = false;
            }
            Signal();
            break;
        case CT_ENQUEUE:
            ASSERT(mode == READ);
            TRACE("%s %s Received enqueue command: amount: %llu, free: %llu", StateStr(state),
                    UnlockedClockStr(), (unsigned long long)command->data.enqueue.amount,
                    (unsigned long long)local_queue.Freespace());
            ASSERT(local_queue.Freespace() >= command->data.enqueue.amount, "Freespace: %llu"
                    "\n%s\n%llu", (unsigned long long)local_queue.Freespace(), UnlockedClockStr(),
                    (unsigned long long)GetKey());
            local_queue.AdvanceHead(command->data.enqueue.amount);
            writerequest = 0;
            NotifyData();
            break;
        case CT_DEQUEUE:
            ASSERT(mode == WRITE);
            TRACE("%s %s Received dequeue command: amount: %llu count: %llu", StateStr(state),
                    UnlockedClockStr(), (unsigned long long)command->data.dequeue.amount,
                    (unsigned long long)remote_queue.Count());
            ASSERT(remote_queue.Count() >= command->data.dequeue.amount);
            remote_queue.AdvanceTail(command->data.dequeue.amount);
            readrequest = 0;
            break;
        case CT_BLOCK:
            TRACE("%s %s Received block command: amount: %llu, count: %llu",
                    StateStr(state), UnlockedClockStr(),
                    (unsigned long long)command->data.block.amount,
                    (unsigned long long)local_queue.Count());
            if (mode == READ) {
                if (local_queue.Full()) {
                    writerequest = command->data.block.amount;
                }
            } else {
                if (local_queue.Empty()) {
                    readrequest = command->data.block.amount;
                }
            }
            if (useD4R && UnlockedIsNormal()) { PostD4RTag(); }
            break;
        case CT_SHUTDOWN:
            logger.Trace("%s %s Received shutdown command", StateStr(state), UnlockedClockStr());
            if (mode == READ) {
                QueueBase::UnlockedShutdownWriter();
                state = S_DEAD;
            } else {
                QueueBase::UnlockedShutdownReader();
                state = S_DEAD;
            }
            logger.Trace("%s Transition from %s to %s state", UnlockedClockStr(),
                    StateStr(oldstate), StateStr(state));
            break;
        case CT_GROW:
            logger.Trace("%s %s Received grow command len: %llu thresh: %llu",
                    StateStr(state), UnlockedClockStr(),
                    (unsigned long long)command->data.grow.length,
                    (unsigned long long)command->data.grow.threshold);
            grow_length = max(grow_length, command->data.grow.length);
            grow_thresh = max(grow_thresh, command->data.grow.threshold);
            if (mode == WRITE) {
                Cork();
                pending_grow = true;
            } else {
                Cork();
                InternalGrow();
                PostUpdateQueue();
                Uncork();
            }
            break;
        case CT_D4R:
            TRACE("%s %s Received D4R tag", StateStr(state), UnlockedClockStr());
            {
                D4R::Tag tag;
                tag.Count(command->data.d4r.count);
                tag.Key(command->data.d4r.key);
                tag.QueueSize(command->data.d4r.qsize);
                tag.QueueKey(command->data.d4r.qkey);
                mocknode->SetPublicTag(tag);
            }
            if (mode == WRITE) {
                QueueBase::UnlockedSignalReaderTagChanged();
            } else {
                QueueBase::UnlockedSignalWriterTagChanged();
            }
            break;
        case CT_CS_REQ:
            {
                uint64_t timestamp = command->data.cs_info.timestamp;
                TRACE("%s %s Receive CS_REQ, ts: %llu", StateStr(state), UnlockedClockStr(),
                        (unsigned long long)timestamp);
                if (
                        !local_want_cs ||
                        local_cs_ts > timestamp ||
                        (local_cs_ts == timestamp && mode == READ)
                   ) {
                    PostCSOK();
                } else {
                    remote_want_cs = true;
                }
            }
            break;
        case CT_CS_OK:
            TRACE("%s %s Receive CS_OK", StateStr(state), UnlockedClockStr());
            cs_ok = true;
            Signal();
            break;
        case CT_FLUSH:
            TRACE("%s %s Receive FLUSH", StateStr(state), UnlockedClockStr());
            num_enqueued = command->data.flush.num_enqueued;
            flushed = true;
            Signal();
            break;
        case CT_RESET:
            TRACE("%s %s Receive RESET", StateStr(state), UnlockedClockStr());
            num_enqueued = 0;
            num_dequeued = 0;
            QueueBase::InternalReset();
            break;
        default:
            ASSERT(false);
        }
    }

    void RDMAQueue::Initialize() {
        int num_devices = 0;
        ibv_device **devices = ibv_get_device_list(&num_devices);
        ibv_device *device = 0;
        if (num_devices < 1) {
            throw std::runtime_error("Error getting devices, no devices");
        }
        for (int i = 0; i < num_devices; ++i) {
            if (if_name.empty()) {
                device = devices[i];
                if_name = ibv_get_device_name(devices[i]);
                break;
            } else if (if_name == ibv_get_device_name(devices[i])) {
                device = devices[i];
                break;
            }
        }
        if (!device) {
            throw std::runtime_error("Error getting device, no matching device.");
        }
        ctx = ibv_open_device(device);
        if (!ctx) { throw ErrnoException(); }
        pd = ibv_alloc_pd(ctx);
        if (!pd) { throw ErrnoException(); }
        cc = ibv_create_comp_channel(ctx);
        if (!cc) { throw ErrnoException(); }
        cq = ibv_create_cq(ctx, max_send_wr + max_recv_wr, 0, cc, 0);
        if (!cq) { throw ErrnoException(); }
        ibv_device_attr device_attr;
        memset(&device_attr, 0, sizeof(device_attr));
        if (ibv_query_device(ctx, &device_attr)) { throw ErrnoException(); }
        ibv_qp_init_attr qp_attr;
        memset(&qp_attr, 0, sizeof(qp_attr));
        qp_attr.qp_context = 0;
        qp_attr.send_cq = cq;
        qp_attr.recv_cq = cq;
        qp_attr.sq_sig_all = 1;
        qp_attr.cap.max_send_wr = max_send_wr;
        qp_attr.cap.max_recv_wr = max_recv_wr;
        qp_attr.cap.max_send_sge = 1;
        qp_attr.cap.max_recv_sge = 1;
        qp_attr.qp_type = IBV_QPT_RC;
        qp = ibv_create_qp(pd, &qp_attr);
        if (!qp) { throw ErrnoException(); }
        local_info.qp_num = qp->qp_num;
        ibv_port_attr port_attr;
        memset(&port_attr, 0, sizeof(port_attr));
        if (if_port <= 0) {
            int i;
            for (i = 1; i <= device_attr.phys_port_cnt; ++i) {
                if (ibv_query_port(ctx, i, &port_attr)) {
                    throw ErrnoException();
                }
                if (port_attr.state == IBV_PORT_ACTIVE) {
                    if_port = i;
                    local_info.lid = port_attr.lid;
                    break;
                }
            }
            if (i > device_attr.phys_port_cnt) {
                throw std::runtime_error("Unable to find an active port");
            }
        } else {
            if (ibv_query_port(ctx, if_port, &port_attr)) {
                throw ErrnoException();
            }
            if (port_attr.state == IBV_PORT_ACTIVE) {
                local_info.lid = port_attr.lid;
            } else {
                throw std::runtime_error("Port is not active");
            }
        }
    }

    void RDMAQueue::Cleanup() {
        if (write_command_queue_mr) {
            ibv_dereg_mr(write_command_queue_mr);
            write_command_queue_mr = 0;
        }
        if (read_command_queue_mr) {
            ibv_dereg_mr(read_command_queue_mr);
            read_command_queue_mr = 0;
        }
        if (mbs_mr) {
            ibv_dereg_mr(mbs_mr);
            mbs_mr = 0;
        }
        if (mbs_old_mr) {
            ibv_dereg_mr(mbs_old_mr);
            mbs_old_mr = 0;
        }
        if (qp) {
            ibv_destroy_qp(qp);
            qp = 0;
        }
        if (cq) {
            ibv_destroy_cq(cq);
            cq = 0;
        }
        if (cc) {
            ibv_destroy_comp_channel(cc);
            cc = 0;
        }
        if (pd) {
            ibv_dealloc_pd(pd);
            pd = 0;
        }
        if (ctx) {
            ibv_close_device(ctx);
            ctx = 0;
        }
    }

    static void ExchangeData(SocketHandle &sock, char *src, char *dst, unsigned len) {
        unsigned num_written = 0;
        while (num_written < len) {
            num_written += sock.Write(src + num_written, len - num_written);
        }
        unsigned num_read = 0;
        while (num_read < len) {
            unsigned nr = sock.Read(dst + num_read, len - num_read);
            if (nr == 0 && sock.Eof()) {
                throw std::runtime_error("Unexpected EOF encountered on socket.");
            }
            num_read += nr;
        }
    }

    void RDMAQueue::Connect() {
        SocketHandle sock;
        while (sock.Closed()) {
            kernel->CheckTerminated();
            logger.Debug("Connecting Socket");
            shared_ptr<Sync::Future<int> > conn;
            if (mode == WRITE) {
                conn = server->ConnectWriter(GetKey());
            } else {
                conn = server->ConnectReader(GetKey());
            }
            if (conn) {
                sock.Reset();
                sock.FD(conn->Get());
            }
        }
        logger.Debug("Socket Connected, exchanging data (lid: %d, qp_num: %d",
                local_info.lid, local_info.qp_num);
        ExchangeData(sock, (char*)&local_info, (char*)&remote_info, sizeof(PortInfo));
        sock.Close();
        logger.Debug("Got data (lid: %d, qp_num: %d) connecting the qp", remote_info.lid, remote_info.qp_num);

        AutoLock<QueueBase> al(*this);
        ibv_qp_attr attr;
        memset(&attr, 0, sizeof(attr));
        attr.qp_state = IBV_QPS_INIT;
        attr.pkey_index = 0;
        attr.port_num = if_port;
        attr.qp_access_flags = IBV_ACCESS_DEFAULT;
        if (ibv_modify_qp(qp, &attr,
                    IBV_QP_STATE | IBV_QP_ACCESS_FLAGS |
                    IBV_QP_PORT | IBV_QP_PKEY_INDEX))
        {
            throw ErrnoException();
        }
        PostReceives();

        logger.Debug("QP now in INIT state");

        memset(&attr, 0, sizeof(attr));
        attr.qp_state = IBV_QPS_RTR;
        attr.path_mtu = IBV_MTU_512;

        attr.ah_attr.is_global = 0;
        attr.ah_attr.dlid = remote_info.lid;
        attr.ah_attr.sl = 0;
        attr.ah_attr.src_path_bits = 0;
        attr.ah_attr.port_num = if_port;

        attr.dest_qp_num = remote_info.qp_num;
        attr.rq_psn = 0;
        attr.max_dest_rd_atomic = 1;
        attr.min_rnr_timer = 12; // Recommended value from documentation

        if (ibv_modify_qp(qp, &attr,
                    IBV_QP_STATE | IBV_QP_PATH_MTU |
                    IBV_QP_AV | IBV_QP_DEST_QPN | IBV_QP_RQ_PSN |
                    IBV_QP_MAX_DEST_RD_ATOMIC | IBV_QP_MIN_RNR_TIMER
                    ))
        {
            throw ErrnoException();
        }

        logger.Debug("QP now in RTR state");

        memset(&attr, 0, sizeof(attr));
        attr.qp_state = IBV_QPS_RTS;
        attr.timeout = 14; // Recommended from docs
        attr.retry_cnt = 7; // Recommended from docs
        attr.rnr_retry = 7; // Recommended from docs
        attr.sq_psn = 0;
        attr.max_rd_atomic = 1;
        if (ibv_modify_qp(qp, &attr,
                    IBV_QP_STATE | IBV_QP_TIMEOUT | IBV_QP_RETRY_CNT |
                    IBV_QP_RNR_RETRY | IBV_QP_SQ_PSN | IBV_QP_MAX_QP_RD_ATOMIC
                    ))
        {
            throw ErrnoException();
        }

        logger.Debug("QP now in RTS state");

        ibv_req_notify_cq(cq, 0);

        PostUpdateQueue();
        logger.Debug("Connection complete");
    }

    void RDMAQueue::Cork() {
        ++cork;
        TRACE("Cork: %u", cork);
    }

    void RDMAQueue::Uncork() {
        ASSERT(cork > 0);
        --cork;
        TRACE("Uncork: %u", cork);
        if (cork == 0) {
            if (mode == WRITE) {
                CheckRDMAWrite();
            } else {
            }
        }
        Signal();
    }

    void RDMAQueue::EnterCriticalSection() {
        if (!UnlockedIsNormal()) { throw BrokenQueueException(GetKey()); }
        local_want_cs = true;
        local_cs_ts = clock;
        cs_ok = false;
        PostCSRequest();
        while ((!cs_ok || local_cs_ts > old_clock) && UnlockedIsNormal()) { Wait(); }
        TRACE("%s Entering Critical Section", UnlockedClockStr());
    }

    void RDMAQueue::LeaveCriticalSection() {
        TRACE("%s Leaving Critical Section", UnlockedClockStr());
        if (!UnlockedIsNormal()) { return; }
        local_want_cs = false;
        if (remote_want_cs) {
            PostCSOK();
            remote_want_cs = false;
        }
    }

    void RDMAQueue::InternalGrow() {
        logger.Trace("Internal grow, inenqueue: %d, indequeue: %d, grow_length: %u, grow_thresh: %u",
                (int)inenqueue, (int)indequeue, (unsigned)grow_length, (unsigned)grow_thresh);
        enqueue_use_old = inenqueue;
        dequeue_use_old = indequeue;

        // Allocate new buffer
        QueueModel new_queue;
        new_queue.length = ComputeQueueLength(grow_length, grow_thresh, alpha, mode);
        auto_ptr<MirrorBufferSet> new_mbs;
        new_mbs.reset(new MirrorBufferSet(new_queue.length, grow_thresh, num_channels));
        new_queue.length = new_mbs->BufferSize();
        new_queue.base = (char*)(void*)*new_mbs;
        new_queue.stride = new_mbs->BufferSize() + new_mbs->MirrorSize();
        grow_thresh = new_mbs->MirrorSize();
        ibv_mr *new_mbs_mr = ibv_reg_mr(pd, new_queue.base,
                (new_queue.length + grow_thresh)*num_channels, IBV_ACCESS_DEFAULT);
        new_queue.lkey = new_mbs_mr->lkey;

        // Copy the data over
        uint64_t thresh = min((uint64_t)max_threshold, (uint64_t)grow_thresh + 1);
        while (new_queue.Count() < local_queue.Count()) {
            uint64_t amount = min((uint64_t)thresh, local_queue.Count() - new_queue.Count());
            for (unsigned chan = 0; chan < num_channels; ++chan) {
                memcpy((new_queue.GetHead() + chan*new_queue.stride),
                        (local_queue.GetTail(new_queue.Count())
                         + chan*local_queue.stride),
                         amount);
            }
            new_queue.AdvanceHead(amount);
        }
        max_threshold = std::min(grow_thresh + 1, new_queue.length);

        // Transition over
        if ((enqueue_use_old || dequeue_use_old) && !mbs_old.get()) {
            logger.Trace("Keeping old queue, old key %llu, new key %llu",
                    (unsigned long long)local_queue.lkey, (unsigned long long)new_queue.lkey);
            mbs_old = mbs;
            local_old_queue = local_queue;
            mbs_old_mr = mbs_mr;

            mbs = new_mbs;
            local_queue = new_queue;
            mbs_mr = new_mbs_mr;
        } else {
            logger.Trace("Replacing old queue, old key %llu, new key %llu",
                    (unsigned long long)local_queue.lkey, (unsigned long long)new_queue.lkey);
            ibv_dereg_mr(mbs_mr);
            mbs_mr = new_mbs_mr;
            mbs = new_mbs;
            local_queue = new_queue;
        }
    }

    bool RDMAQueue::ActiveTransfers() {
        return (num_send_wr != 0 || pending_rdma > 0 || confirmed_rdma > 0);
    }

    bool RDMAQueue::IsDead() {
        AutoLock<QueueBase> al(*this);
        return UnlockedIsDead();
    }
    bool RDMAQueue::UnlockedIsDead() {
        return (state == S_DEAD || state == S_ERROR);
    }
    bool RDMAQueue::UnlockedIsNormal() {
        return (state == S_RUNNING);
    }

    const char* RDMAQueue::ClockStr() {
        AutoLock<QueueBase> al(*this);
        return UnlockedClockStr();
    }

    const char* RDMAQueue::UnlockedClockStr() {
        std::ostringstream oss;
        oss << "<" << clock << ", " << remote_clock << ", " << old_clock << ">";
        clock_str = oss.str();
        return clock_str.c_str();
    }

    void RDMAQueue::ShutdownError() {
        AutoLock<QueueBase> al(*this);
        readshutdown = true;
        writeshutdown = true;
        state = S_ERROR;
        Signal();
    }

}
#endif // ENABLE_RDMA