transport/tcp/device.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 <array>
0012 #include <atomic>
0013 #include <condition_variable>
0014 #include <memory>
0015 #include <mutex>
0016 #include <thread>
0017
0018 #include <gloo/transport/device.h>
0019 #include <gloo/transport/tcp/attr.h>
0020 #include <gloo/transport/tcp/error.h>
0021 #include <gloo/transport/tcp/listener.h>
0022 #include <gloo/transport/tcp/loop.h>
0023 #include <gloo/transport/tcp/socket.h>
0024
0025 namespace gloo {
0026 namespace transport {
0027 namespace tcp {
0028
0029 struct attr CreateDeviceAttr(const struct attr& src);
0030
0031 std::shared_ptr<::gloo::transport::Device> CreateDevice(
0032     const struct attr&);
0033
0034 // Forward declarations
0035 class Pair;
0036 class Buffer;
0037
0038 class Device : public ::gloo::transport::Device,
0039                public std::enable_shared_from_this<Device> {
0040  public:
0041   explicit Device(const struct attr& attr);
0042   virtual ~Device();
0043
0044   virtual std::string str() const override;
0045
0046   virtual const std::string& getPCIBusID() const override;
0047
0048   virtual int getInterfaceSpeed() const override;
0049
0050   virtual std::shared_ptr<::gloo::transport::Context> createContext(
0051       int rank, int size) override;
0052
0053   void registerDescriptor(int fd, int events, Handler* h);
0054   void unregisterDescriptor(int fd, Handler* h);
0055
0056   // TCP is bidirectional so when we connect two ends of a pair,
0057   // one side is the connection initiator and the other is the listener.
0058   bool isInitiator(
0059       const Address& local,
0060       const Address& remote) const;
0061
0062  protected:
0063   const struct attr attr_;
0064
0065   // Return a new `Address` instance.
0066   //
0067   // This is called by the constructor of the `Pair` class. It gives
0068   // the pair a uniquely identifying address even though the device
0069   // uses a shared listening socket.
0070   //
0071   Address nextAddress();
0072
0073   // Connect a pair to a remote.
0074   //
0075   // This is performed by the device instance because we use a single
0076   // listening socket for all inbound pair connections.
0077   //
0078   // Matching these connections with pairs is done with a handshake.
0079   // The remote side of the connection writes a sequence number (see
0080   // `Address::sequence_t`) to the stream that identifies the pair
0081   // it wants to connect to. On the local side, this sequence number
0082   // is read and used as key in a map with callbacks. If the callback
0083   // is found, it is called. If the callback is not found, the
0084   // connection is cached in a map, using the sequence number.
0085   //
0086   using connect_callback_t =
0087       std::function<void(std::shared_ptr<Socket> socket, Error error)>;
0088
0089   void connect(
0090       const Address& local,
0091       const Address& remote,
0092       std::chrono::milliseconds timeout,
0093       connect_callback_t fn);
0094
0095   void connectAsListener(
0096       const Address& local,
0097       std::chrono::milliseconds timeout,
0098       connect_callback_t fn);
0099
0100   void connectAsInitiator(
0101       const Address& remote,
0102       std::chrono::milliseconds timeout,
0103       connect_callback_t fn);
0104
0105   friend class Pair;
0106   friend class Buffer;
0107
0108  private:
0109   std::shared_ptr<Loop> loop_;
0110   std::shared_ptr<Listener> listener_;
0111
0112   std::string interfaceName_;
0113   int interfaceSpeedMbps_;
0114   std::string pciBusID_;
0115 };
0116
0117 } // namespace tcp
0118 } // namespace transport
0119 } // namespace gloo