transport/tcp/buffer.h

0001 /**
0002  * Copyright (c) 2017-present, Facebook, Inc.
0003  * All rights reserved.
0004  *
0005  * This source code is licensed under the BSD-style license found in the
0006  * LICENSE file in the root directory of this source tree.
0007  */
0008
0009 #pragma once
0010
0011 #include <atomic>
0012 #include <condition_variable>
0013 #include <exception>
0014 #include <map>
0015 #include <mutex>
0016 #include <queue>
0017
0018 #include "gloo/transport/buffer.h"
0019 #include "gloo/transport/tcp/device.h"
0020 #include "gloo/transport/tcp/pair.h"
0021
0022 namespace gloo {
0023 namespace transport {
0024 namespace tcp {
0025
0026 class Buffer : public ::gloo::transport::Buffer {
0027  public:
0028   virtual ~Buffer();
0029
0030   virtual void send(size_t offset, size_t length, size_t roffset = 0) override;
0031
0032   virtual void waitRecv() override;
0033   virtual void waitSend() override;
0034
0035   void handleRecvCompletion();
0036   void handleSendCompletion();
0037
0038  protected:
0039   // May only be constructed from helper function in pair.cc
0040   Buffer(Pair* pair, int slot, void* ptr, size_t size);
0041
0042   Pair* pair_;
0043
0044   std::mutex m_;
0045   std::condition_variable recvCv_;
0046   std::condition_variable sendCv_;
0047
0048   int recvCompletions_;
0049   int sendCompletions_;
0050   std::atomic<int> sendPending_;
0051
0052   std::exception_ptr ex_;
0053
0054   // Throws if an exception if set.
0055   void throwIfException();
0056
0057   // Set exception and wake up any waitRecv/waitSend threads.
0058   void signalException(std::exception_ptr);
0059
0060   friend class Pair;
0061 };
0062
0063 } // namespace tcp
0064 } // namespace transport
0065 } // namespace gloo