yanjianglu commented on code in PR #1836: URL: https://github.com/apache/incubator-brpc/pull/1836#discussion_r943260172
########## src/brpc/rdma/rdma_endpoint.h: ########## @@ -0,0 +1,258 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +#ifndef BRPC_RDMA_ENDPOINT_H +#define BRPC_RDMA_ENDPOINT_H + +#if BRPC_WITH_RDMA +#include <cstring> +#include <iostream> +#include <string> +#include <vector> +#include <infiniband/verbs.h> +#include "butil/atomicops.h" +#include "butil/iobuf.h" +#include "butil/macros.h" +#include "brpc/socket.h" + + +namespace brpc { +class Socket; +namespace rdma { + +class RdmaConnect : public AppConnect { +public: + void StartConnect(const Socket* socket, + void (*done)(int err, void* data), void* data) override; + void StopConnect(Socket*) override; + struct RunGuard { + RunGuard(RdmaConnect* rc) { this_rc = rc; } + ~RunGuard() { if (this_rc) this_rc->Run(); } + RdmaConnect* this_rc; + }; + +private: + void Run(); + void (*_done)(int, void*); + void* _data; +}; + +struct RdmaResource { + ibv_qp* qp; + ibv_cq* cq; + ibv_comp_channel* comp_channel; + RdmaResource* next; + RdmaResource(); + ~RdmaResource(); +}; + +class BAIDU_CACHELINE_ALIGNMENT RdmaEndpoint : public SocketUser { +friend class RdmaConnect; +friend class brpc::Socket; +public: + RdmaEndpoint(Socket* s); + ~RdmaEndpoint(); + + // Global initialization + // Return 0 if success, -1 if failed and errno set + static int GlobalInitialize(); + + static void GlobalRelease(); + + // Reset the endpoint (for next use) + void Reset(); + + // Cut data from the given IOBuf list and use RDMA to send + // Return bytes cut if success, -1 if failed and errno set + ssize_t CutFromIOBufList(butil::IOBuf** data, size_t ndata); + + // Whether the endpoint can send more data + bool IsWritable() const; + + // For debug + void DebugInfo(std::ostream& os) const; + + // Callback when there is new epollin event on TCP fd + static void OnNewDataFromTcp(Socket* m); + +private: + enum State { + UNINIT = 0x0, + C_ALLOC_QPCQ = 0x1, + C_HELLO_SEND = 0x2, + C_HELLO_WAIT = 0x3, + C_BRINGUP_QP = 0x4, + C_ACK_SEND = 0x5, + S_HELLO_WAIT = 0x11, + S_ALLOC_QPCQ = 0x12, + S_BRINGUP_QP = 0x13, + S_HELLO_SEND = 0x14, + S_ACK_WAIT = 0x15, + ESTABLISHED = 0x100, + FALLBACK_TCP = 0x200, + FAILED = 0x300 + }; + + // Process handshake at the client + static void* ProcessHandshakeAtClient(void* arg); + + // Process handshake at the server + static void* ProcessHandshakeAtServer(void* arg); + + // Allocate resources + // Return 0 if success, -1 if failed and errno set + int AllocateResources(); + + // Release resources + void DeallocateResources(); + + // Send Imm data to the remote side + // Arguments: + // imm: imm data in the WR + // Return: + // 0: success + // -1: failed, errno set + int SendImm(uint32_t imm); + + // Try to send pure ACK to the remote side + // Arguments: + // num: the number of rq entry received + // Return: + // 0: success + // -1: failed, errno set + int SendAck(int num); + + // Handle CQE + // If wc is not RDMA RECV event: + // return 0 if success, -1 if failed and errno set + // If wc is RDMA RECV event: + // return bytes appended if success, -1 if failed and errno set + ssize_t HandleCompletion(ibv_wc& wc); + + // Post a given number of WRs to Recv Queue + // If zerocopy is true, reallocate block. + // Return 0 if success, -1 if failed and errno set + int PostRecv(uint32_t num, bool zerocopy); + + // Post a WR pointing to the block to the local Recv Queue + // Arguments: + // block: the addr to receive data (ibv_sge.addr) + // block_size: the maximum length can be received (ibv_sge.length) + // Return: + // 0: success + // -1: failed, errno set + int DoPostRecv(void* block, size_t block_size); + + // Read at most len bytes from fd in _socket to data + // wait for _read_butex if encounter EAGAIN + // return -1 if encounter other errno (including EOF) + int ReadFromFd(void* data, size_t len); + + + // Wrute at most len bytes from data to fd in _socket Review Comment: Wrute 拼写错误 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
