transport/tcp/listener.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 <list>
0012 #include <unordered_map>
0013
0014 #include <gloo/transport/tcp/address.h>
0015 #include <gloo/transport/tcp/attr.h>
0016 #include <gloo/transport/tcp/error.h>
0017 #include <gloo/transport/tcp/loop.h>
0018 #include <gloo/transport/tcp/socket.h>
0019
0020 namespace gloo {
0021 namespace transport {
0022 namespace tcp {
0023
0024 // Listener deals with incoming connections. Incoming connections
0025 // write a few bytes containing a sequence number. This sequence
0026 // number is read off the socket and matched to a local sequence
0027 // number. If there is a match, the socket is passed to the
0028 // corresponding pair. If it can't be matched, it is stashed until a
0029 // pair with the sequence number calls `waitForConnection`.
0030 class Listener final : public Handler {
0031  public:
0032   using connect_callback_t =
0033       std::function<void(std::shared_ptr<Socket> socket, Error error)>;
0034
0035   static constexpr int kBacklog = -1;  // allow somaxconn
0036
0037   Listener(std::shared_ptr<Loop> loop, const attr& attr);
0038
0039   ~Listener() override;
0040
0041   void handleEvents(int events) override;
0042
0043   Address nextAddress();
0044
0045   // Wait for connection with sequence number `seq`. The callback is
0046   // always called from a different thread (the event loop thread),
0047   // even if the connection is already available.
0048   void waitForConnection(sequence_number_t seq, connect_callback_t fn);
0049
0050  private:
0051   std::mutex mutex_;
0052   std::shared_ptr<Loop> loop_;
0053   std::shared_ptr<Socket> listener_;
0054
0055   // Address of this listener and the sequence number for the next
0056   // connection. Sequence numbers are written by a peer right after
0057   // establishing a new connection and used locally to match a new
0058   // connection to a pair instance.
0059   Address addr_;
0060   sequence_number_t seq_{0};
0061
0062   // Called when we've read a sequence number from a new socket.
0063   void haveConnection(std::shared_ptr<Socket> socket, sequence_number_t seq);
0064
0065   // Callbacks by sequence number (while waiting for a connection).
0066   std::unordered_map<sequence_number_t, connect_callback_t> seqToCallback_;
0067
0068   // Sockets by sequence number (while waiting for a pair to call).
0069   std::unordered_map<sequence_number_t, std::shared_ptr<Socket>> seqToSocket_;
0070 };
0071
0072 } // namespace tcp
0073 } // namespace transport
0074 } // namespace gloo