transport/tcp/loop.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 <functional>
0014 #include <list>
0015 #include <memory>
0016 #include <mutex>
0017 #include <thread>
0018
0019 #include <sys/epoll.h>
0020
0021 namespace gloo {
0022 namespace transport {
0023 namespace tcp {
0024
0025 // Handler abstract base class called by the epoll(2) event loop.
0026 // Dispatch to multiple types is needed because we must deal with a
0027 // single listening socket on the device instance and I/O for all pair
0028 // instances. Before this approach, we'd exclusively deal with `Pair`
0029 // instances and didn't need to dispatch events to different types.
0030 class Handler {
0031  public:
0032   virtual ~Handler() = default;
0033
0034   virtual void handleEvents(int events) = 0;
0035 };
0036
0037 class Loop;
0038
0039 // Functions can be deferred to the epoll(2) thread through the this
0040 // class. It uses readability of a pipe to wake up the event loop.
0041 class Deferrables final : public Handler {
0042 public:
0043   using function_t = std::function<void()>;
0044
0045   Deferrables();
0046
0047   ~Deferrables() override;
0048
0049   void defer(function_t fn);
0050
0051   void handleEvents(int events) override;
0052
0053 private:
0054   int rfd_;
0055   int wfd_;
0056
0057   std::mutex mutex_;
0058   std::list<function_t> functions_;
0059   bool triggered_{false};
0060
0061   friend class Loop;
0062 };
0063
0064 class Loop final : public std::enable_shared_from_this<Loop> {
0065  public:
0066   explicit Loop();
0067
0068   ~Loop();
0069
0070   void registerDescriptor(int fd, int events, Handler* h);
0071
0072   void unregisterDescriptor(int fd, Handler *h);
0073
0074   void defer(std::function<void()> fn);
0075
0076   void run();
0077
0078  private:
0079   static constexpr auto capacity_ = 64;
0080
0081   int fd_{-1};
0082   std::atomic<bool> done_{false};
0083   Deferrables deferrables_;
0084   std::unique_ptr<std::thread> loop_;
0085
0086   std::mutex m_;
0087   std::condition_variable cv_;
0088 };
0089
0090 } // namespace tcp
0091 } // namespace transport
0092 } // namespace gloo