<commit_before>/* Copyright (c) 2010 Stanford University
 *
 * Permission to use, copy, modify, and distribute this software for any purpose
 * with or without fee is hereby granted, provided that the above copyright
 * notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR(S) DISCLAIM ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL AUTHORS BE LIABLE FOR ANY
 * SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
 * RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
 * CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

/**
 * \file
 * This file defines an implementation of Transport for Infiniband
 * using reliable connected queue-pairs (RC).
 */

#include <time.h>
#include <string>
#include <boost/unordered_map.hpp>
#include <vector>

#include "BoostIntrusive.h"
#include "Common.h"
#include "Dispatch.h"
#include "IpAddress.h"
#include "Tub.h"
#include "Segment.h"
#include "Transport.h"
#include "Infiniband.h"

#ifndef RAMCLOUD_INFRCTRANSPORT_H
#define RAMCLOUD_INFRCTRANSPORT_H

namespace RAMCloud {

template<typename Infiniband = RealInfiniband>
class InfRcTransport : public Transport {
    // forward declarations
    class InfRCSession;
    typedef typename Infiniband::BufferDescriptor BufferDescriptor;
    typedef typename Infiniband::QueuePair QueuePair;
    typedef typename Infiniband::QueuePairTuple QueuePairTuple;

  public:
    explicit InfRcTransport(const ServiceLocator* sl = NULL);
    ~InfRcTransport();
    ServerRpc* serverRecv() __attribute__((warn_unused_result));
    SessionRef getSession(const ServiceLocator& sl) {
        return new InfRCSession(this, sl);
    }
    ServiceLocator getServiceLocator();
    void dumpStats() {
        infiniband->dumpStats();
    }
    uint32_t getMaxRpcSize() const;
    void registerMemory(void* base, size_t bytes)
    {
        assert(logMemoryRegion == NULL);
        logMemoryRegion = ibv_reg_mr(infiniband->pd.pd, base, bytes,
            IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE);
        assert(logMemoryRegion != NULL);
        logMemoryBase = reinterpret_cast<uintptr_t>(base);
        logMemoryBytes = bytes;
        LOG(NOTICE, "Registered %Zd bytes at %p", bytes, base);
    }

  private:
    class ServerRpc : public Transport::ServerRpc {
        public:
            explicit ServerRpc(InfRcTransport* transport, QueuePair* qp,
                               uint64_t nonce);
            void sendReply();
        private:
            InfRcTransport* transport;
            QueuePair*      qp;
            /// Uniquely identifies the RPC.
            uint64_t        nonce;
            DISALLOW_COPY_AND_ASSIGN(ServerRpc);
    };

    class ClientRpc : public Transport::ClientRpc {
        public:
            explicit ClientRpc(InfRcTransport* transport,
                               InfRCSession* session,
                               Buffer* request,
                               Buffer* response,
                               uint64_t nonce);
            bool isReady();
            void sendOrQueue();
            void wait();

        private:
            bool
            tryZeroCopy(Buffer* request);

            InfRcTransport*     transport;
            InfRCSession*       session;
            Buffer*             request;
            Buffer*             response;
            /// Uniquely identifies the RPC.
            uint64_t            nonce;
            enum {
                PENDING,
                REQUEST_SENT,
                RESPONSE_RECEIVED,
            } state;
        public:
            IntrusiveListHook   queueEntries;
            friend class InfRCSession;
            friend class InfRcTransport;
            DISALLOW_COPY_AND_ASSIGN(ClientRpc);
    };

    /**
     * A header that goes at the start of every RPC request and response.
     */
    struct Header {
        explicit Header(uint64_t nonce) : nonce(nonce) {}
        /// Uniquely identifies the RPC.
        uint64_t nonce;
    };

    // maximum RPC size we'll permit. we'll use the segment size plus a
    // little extra for header overhead, etc.
    static const uint32_t MAX_RPC_SIZE = Segment::SEGMENT_SIZE + 4096;
    static const uint32_t MAX_SHARED_RX_QUEUE_DEPTH = 16;
    static const uint32_t MAX_SHARED_RX_SGE_COUNT = 8;
    static const uint32_t MAX_TX_QUEUE_DEPTH = 8;
    static const uint32_t MAX_TX_SGE_COUNT = 8;
    static const uint32_t QP_EXCHANGE_USEC_TIMEOUT = 50000;
    static const uint32_t QP_EXCHANGE_MAX_TIMEOUTS = 10;

    INTRUSIVE_LIST_TYPEDEF(ClientRpc, queueEntries) ClientRpcList;

    class InfRCSession : public Session {
      public:
        explicit InfRCSession(InfRcTransport *transport,
            const ServiceLocator& sl);
        Transport::ClientRpc* clientSend(Buffer* request, Buffer* response)
            __attribute__((warn_unused_result));
        void release();
<commit_msg>Increase the number of Infiniband send and receive buffers<commit_after>/* Copyright (c) 2010 Stanford University
 *
 * Permission to use, copy, modify, and distribute this software for any purpose
 * with or without fee is hereby granted, provided that the above copyright
 * notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR(S) DISCLAIM ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL AUTHORS BE LIABLE FOR ANY
 * SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
 * RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
 * CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
 * CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */

/**
 * \file
 * This file defines an implementation of Transport for Infiniband
 * using reliable connected queue-pairs (RC).
 */

#include <time.h>
#include <string>
#include <boost/unordered_map.hpp>
#include <vector>

#include "BoostIntrusive.h"
#include "Common.h"
#include "Dispatch.h"
#include "IpAddress.h"
#include "Tub.h"
#include "Segment.h"
#include "Transport.h"
#include "Infiniband.h"

#ifndef RAMCLOUD_INFRCTRANSPORT_H
#define RAMCLOUD_INFRCTRANSPORT_H

namespace RAMCloud {

template<typename Infiniband = RealInfiniband>
class InfRcTransport : public Transport {
    // forward declarations
    class InfRCSession;
    typedef typename Infiniband::BufferDescriptor BufferDescriptor;
    typedef typename Infiniband::QueuePair QueuePair;
    typedef typename Infiniband::QueuePairTuple QueuePairTuple;

  public:
    explicit InfRcTransport(const ServiceLocator* sl = NULL);
    ~InfRcTransport();
    ServerRpc* serverRecv() __attribute__((warn_unused_result));
    SessionRef getSession(const ServiceLocator& sl) {
        return new InfRCSession(this, sl);
    }
    ServiceLocator getServiceLocator();
    void dumpStats() {
        infiniband->dumpStats();
    }
    uint32_t getMaxRpcSize() const;
    void registerMemory(void* base, size_t bytes)
    {
        assert(logMemoryRegion == NULL);
        logMemoryRegion = ibv_reg_mr(infiniband->pd.pd, base, bytes,
            IBV_ACCESS_REMOTE_WRITE | IBV_ACCESS_LOCAL_WRITE);
        assert(logMemoryRegion != NULL);
        logMemoryBase = reinterpret_cast<uintptr_t>(base);
        logMemoryBytes = bytes;
        LOG(NOTICE, "Registered %Zd bytes at %p", bytes, base);
    }

  private:
    class ServerRpc : public Transport::ServerRpc {
        public:
            explicit ServerRpc(InfRcTransport* transport, QueuePair* qp,
                               uint64_t nonce);
            void sendReply();
        private:
            InfRcTransport* transport;
            QueuePair*      qp;
            /// Uniquely identifies the RPC.
            uint64_t        nonce;
            DISALLOW_COPY_AND_ASSIGN(ServerRpc);
    };

    class ClientRpc : public Transport::ClientRpc {
        public:
            explicit ClientRpc(InfRcTransport* transport,
                               InfRCSession* session,
                               Buffer* request,
                               Buffer* response,
                               uint64_t nonce);
            bool isReady();
            void sendOrQueue();
            void wait();

        private:
            bool
            tryZeroCopy(Buffer* request);

            InfRcTransport*     transport;
            InfRCSession*       session;
            Buffer*             request;
            Buffer*             response;
            /// Uniquely identifies the RPC.
            uint64_t            nonce;
            enum {
                PENDING,
                REQUEST_SENT,
                RESPONSE_RECEIVED,
            } state;
        public:
            IntrusiveListHook   queueEntries;
            friend class InfRCSession;
            friend class InfRcTransport;
            DISALLOW_COPY_AND_ASSIGN(ClientRpc);
    };

    /**
     * A header that goes at the start of every RPC request and response.
     */
    struct Header {
        explicit Header(uint64_t nonce) : nonce(nonce) {}
        /// Uniquely identifies the RPC.
        uint64_t nonce;
    };

    // maximum RPC size we'll permit. we'll use the segment size plus a
    // little extra for header overhead, etc.
    static const uint32_t MAX_RPC_SIZE = Segment::SEGMENT_SIZE + 4096;
    static const uint32_t MAX_SHARED_RX_QUEUE_DEPTH = 24;
    static const uint32_t MAX_SHARED_RX_SGE_COUNT = 8;
    static const uint32_t MAX_TX_QUEUE_DEPTH = 12;
    static const uint32_t MAX_TX_SGE_COUNT = 8;
    static const uint32_t QP_EXCHANGE_USEC_TIMEOUT = 50000;
    static const uint32_t QP_EXCHANGE_MAX_TIMEOUTS = 10;

    INTRUSIVE_LIST_TYPEDEF(ClientRpc, queueEntries) ClientRpcList;

    class InfRCSession : public Session {
      public:
        explicit InfRCSession(InfRcTransport *transport,
            const ServiceLocator& sl);
        Transport::ClientRpc* clientSend(Buffer* request, Buffer* response)
            __attribute__((warn_unused_result));
        void release();

{'fraction_non_alphanumeric': 0.07254290171606864, 'fraction_numerical': 0.009457878315132605, 'mean_word_length': 2.466373774923961, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '878886', 'n_tokens_mistral': 3045, 'n_tokens_neox': 2894, 'n_words': 848}