Add disableTransparentTls call
[folly.git] / folly / io / async / AsyncServerSocket.h
1 /*
2  * Copyright 2017 Facebook, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  *   http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16
17 #pragma once
18
19 #include <folly/SocketAddress.h>
20 #include <folly/io/ShutdownSocketSet.h>
21 #include <folly/io/async/AsyncSocketBase.h>
22 #include <folly/io/async/AsyncTimeout.h>
23 #include <folly/io/async/DelayedDestruction.h>
24 #include <folly/io/async/EventBase.h>
25 #include <folly/io/async/EventHandler.h>
26 #include <folly/io/async/NotificationQueue.h>
27 #include <folly/portability/Sockets.h>
28
29 #include <limits.h>
30 #include <stddef.h>
31 #include <exception>
32 #include <memory>
33 #include <vector>
34
35 // Due to the way kernel headers are included, this may or may not be defined.
36 // Number pulled from 3.10 kernel headers.
37 #ifndef SO_REUSEPORT
38 #define SO_REUSEPORT 15
39 #endif
40
41 #if defined __linux__ && !defined SO_NO_TRANSPARENT_TLS
42 #define SO_NO_TRANSPARENT_TLS 200
43 #endif
44
45 namespace folly {
46
47 /**
48  * A listening socket that asynchronously informs a callback whenever a new
49  * connection has been accepted.
50  *
51  * Unlike most async interfaces that always invoke their callback in the same
52  * EventBase thread, AsyncServerSocket is unusual in that it can distribute
53  * the callbacks across multiple EventBase threads.
54  *
55  * This supports a common use case for network servers to distribute incoming
56  * connections across a number of EventBase threads.  (Servers typically run
57  * with one EventBase thread per CPU.)
58  *
59  * Despite being able to invoke callbacks in multiple EventBase threads,
60  * AsyncServerSocket still has one "primary" EventBase.  Operations that
61  * modify the AsyncServerSocket state may only be performed from the primary
62  * EventBase thread.
63  */
64 class AsyncServerSocket : public DelayedDestruction
65                         , public AsyncSocketBase {
66  public:
67   typedef std::unique_ptr<AsyncServerSocket, Destructor> UniquePtr;
68   // Disallow copy, move, and default construction.
69   AsyncServerSocket(AsyncServerSocket&&) = delete;
70
71   /**
72    * A callback interface to get notified of client socket events.
73    *
74    * The ConnectionEventCallback implementations need to be thread-safe as the
75    * callbacks may be called from different threads.
76    */
77   class ConnectionEventCallback {
78    public:
79     virtual ~ConnectionEventCallback() = default;
80
81     /**
82      * onConnectionAccepted() is called right after a client connection
83      * is accepted using the system accept()/accept4() APIs.
84      */
85     virtual void onConnectionAccepted(const int socket,
86                                       const SocketAddress& addr) noexcept = 0;
87
88     /**
89      * onConnectionAcceptError() is called when an error occurred accepting
90      * a connection.
91      */
92     virtual void onConnectionAcceptError(const int err) noexcept = 0;
93
94     /**
95      * onConnectionDropped() is called when a connection is dropped,
96      * probably because of some error encountered.
97      */
98     virtual void onConnectionDropped(const int socket,
99                                      const SocketAddress& addr) noexcept = 0;
100
101     /**
102      * onConnectionEnqueuedForAcceptorCallback() is called when the
103      * connection is successfully enqueued for an AcceptCallback to pick up.
104      */
105     virtual void onConnectionEnqueuedForAcceptorCallback(
106         const int socket,
107         const SocketAddress& addr) noexcept = 0;
108
109     /**
110      * onConnectionDequeuedByAcceptorCallback() is called when the
111      * connection is successfully dequeued by an AcceptCallback.
112      */
113     virtual void onConnectionDequeuedByAcceptorCallback(
114         const int socket,
115         const SocketAddress& addr) noexcept = 0;
116
117     /**
118      * onBackoffStarted is called when the socket has successfully started
119      * backing off accepting new client sockets.
120      */
121     virtual void onBackoffStarted() noexcept = 0;
122
123     /**
124      * onBackoffEnded is called when the backoff period has ended and the socket
125      * has successfully resumed accepting new connections if there is any
126      * AcceptCallback registered.
127      */
128     virtual void onBackoffEnded() noexcept = 0;
129
130     /**
131      * onBackoffError is called when there is an error entering backoff
132      */
133     virtual void onBackoffError() noexcept = 0;
134   };
135
136   class AcceptCallback {
137    public:
138     virtual ~AcceptCallback() = default;
139
140     /**
141      * connectionAccepted() is called whenever a new client connection is
142      * received.
143      *
144      * The AcceptCallback will remain installed after connectionAccepted()
145      * returns.
146      *
147      * @param fd          The newly accepted client socket.  The AcceptCallback
148      *                    assumes ownership of this socket, and is responsible
149      *                    for closing it when done.  The newly accepted file
150      *                    descriptor will have already been put into
151      *                    non-blocking mode.
152      * @param clientAddr  A reference to a SocketAddress struct containing the
153      *                    client's address.  This struct is only guaranteed to
154      *                    remain valid until connectionAccepted() returns.
155      */
156     virtual void connectionAccepted(int fd,
157                                     const SocketAddress& clientAddr)
158       noexcept = 0;
159
160     /**
161      * acceptError() is called if an error occurs while accepting.
162      *
163      * The AcceptCallback will remain installed even after an accept error,
164      * as the errors are typically somewhat transient, such as being out of
165      * file descriptors.  The server socket must be explicitly stopped if you
166      * wish to stop accepting after an error.
167      *
168      * @param ex  An exception representing the error.
169      */
170     virtual void acceptError(const std::exception& ex) noexcept = 0;
171
172     /**
173      * acceptStarted() will be called in the callback's EventBase thread
174      * after this callback has been added to the AsyncServerSocket.
175      *
176      * acceptStarted() will be called before any calls to connectionAccepted()
177      * or acceptError() are made on this callback.
178      *
179      * acceptStarted() makes it easier for callbacks to perform initialization
180      * inside the callback thread.  (The call to addAcceptCallback() must
181      * always be made from the AsyncServerSocket's primary EventBase thread.
182      * acceptStarted() provides a hook that will always be invoked in the
183      * callback's thread.)
184      *
185      * Note that the call to acceptStarted() is made once the callback is
186      * added, regardless of whether or not the AsyncServerSocket is actually
187      * accepting at the moment.  acceptStarted() will be called even if the
188      * AsyncServerSocket is paused when the callback is added (including if
189      * the initial call to startAccepting() on the AsyncServerSocket has not
190      * been made yet).
191      */
192     virtual void acceptStarted() noexcept {}
193
194     /**
195      * acceptStopped() will be called when this AcceptCallback is removed from
196      * the AsyncServerSocket, or when the AsyncServerSocket is destroyed,
197      * whichever occurs first.
198      *
199      * No more calls to connectionAccepted() or acceptError() will be made
200      * after acceptStopped() is invoked.
201      */
202     virtual void acceptStopped() noexcept {}
203   };
204
205   static const uint32_t kDefaultMaxAcceptAtOnce = 30;
206   static const uint32_t kDefaultCallbackAcceptAtOnce = 5;
207   static const uint32_t kDefaultMaxMessagesInQueue = 1024;
208   /**
209    * Create a new AsyncServerSocket with the specified EventBase.
210    *
211    * @param eventBase  The EventBase to use for driving the asynchronous I/O.
212    *                   If this parameter is nullptr, attachEventBase() must be
213    *                   called before this socket can begin accepting
214    *                   connections.
215    */
216   explicit AsyncServerSocket(EventBase* eventBase = nullptr);
217
218   /**
219    * Helper function to create a shared_ptr<AsyncServerSocket>.
220    *
221    * This passes in the correct destructor object, since AsyncServerSocket's
222    * destructor is protected and cannot be invoked directly.
223    */
224   static std::shared_ptr<AsyncServerSocket>
225   newSocket(EventBase* evb = nullptr) {
226     return std::shared_ptr<AsyncServerSocket>(new AsyncServerSocket(evb),
227                                                  Destructor());
228   }
229
230   void setShutdownSocketSet(ShutdownSocketSet* newSS);
231
232   /**
233    * Destroy the socket.
234    *
235    * AsyncServerSocket::destroy() must be called to destroy the socket.
236    * The normal destructor is private, and should not be invoked directly.
237    * This prevents callers from deleting a AsyncServerSocket while it is
238    * invoking a callback.
239    *
240    * destroy() must be invoked from the socket's primary EventBase thread.
241    *
242    * If there are AcceptCallbacks still installed when destroy() is called,
243    * acceptStopped() will be called on these callbacks to notify them that
244    * accepting has stopped.  Accept callbacks being driven by other EventBase
245    * threads may continue to receive new accept callbacks for a brief period of
246    * time after destroy() returns.  They will not receive any more callback
247    * invocations once acceptStopped() is invoked.
248    */
249   virtual void destroy();
250
251   /**
252    * Attach this AsyncServerSocket to its primary EventBase.
253    *
254    * This may only be called if the AsyncServerSocket is not already attached
255    * to a EventBase.  The AsyncServerSocket must be attached to a EventBase
256    * before it can begin accepting connections.
257    */
258   void attachEventBase(EventBase *eventBase);
259
260   /**
261    * Detach the AsyncServerSocket from its primary EventBase.
262    *
263    * detachEventBase() may only be called if the AsyncServerSocket is not
264    * currently accepting connections.
265    */
266   void detachEventBase();
267
268   /**
269    * Get the EventBase used by this socket.
270    */
271   EventBase* getEventBase() const {
272     return eventBase_;
273   }
274
275   /**
276    * Create a AsyncServerSocket from an existing socket file descriptor.
277    *
278    * useExistingSocket() will cause the AsyncServerSocket to take ownership of
279    * the specified file descriptor, and use it to listen for new connections.
280    * The AsyncServerSocket will close the file descriptor when it is
281    * destroyed.
282    *
283    * useExistingSocket() must be called before bind() or listen().
284    *
285    * The supplied file descriptor will automatically be put into non-blocking
286    * mode.  The caller may have already directly called bind() and possibly
287    * listen on the file descriptor.  If so the caller should skip calling the
288    * corresponding AsyncServerSocket::bind() and listen() methods.
289    *
290    * On error a TTransportException will be thrown and the caller will retain
291    * ownership of the file descriptor.
292    */
293   void useExistingSocket(int fd);
294   void useExistingSockets(const std::vector<int>& fds);
295
296   /**
297    * Return the underlying file descriptor
298    */
299   std::vector<int> getSockets() const {
300     std::vector<int> sockets;
301     for (auto& handler : sockets_) {
302       sockets.push_back(handler.socket_);
303     }
304     return sockets;
305   }
306
307   /**
308    * Backwards compatible getSocket, warns if > 1 socket
309    */
310   int getSocket() const {
311     if (sockets_.size() > 1) {
312       VLOG(2) << "Warning: getSocket can return multiple fds, " <<
313         "but getSockets was not called, so only returning the first";
314     }
315     if (sockets_.size() == 0) {
316       return -1;
317     } else {
318       return sockets_[0].socket_;
319     }
320   }
321
322   /**
323    * Bind to the specified address.
324    *
325    * This must be called from the primary EventBase thread.
326    *
327    * Throws TTransportException on error.
328    */
329   virtual void bind(const SocketAddress& address);
330
331   /**
332    * Bind to the specified port for the specified addresses.
333    *
334    * This must be called from the primary EventBase thread.
335    *
336    * Throws TTransportException on error.
337    */
338   virtual void bind(
339       const std::vector<IPAddress>& ipAddresses,
340       uint16_t port);
341
342   /**
343    * Bind to the specified port.
344    *
345    * This must be called from the primary EventBase thread.
346    *
347    * Throws TTransportException on error.
348    */
349   virtual void bind(uint16_t port);
350
351   /**
352    * Get the local address to which the socket is bound.
353    *
354    * Throws TTransportException on error.
355    */
356   void getAddress(SocketAddress* addressReturn) const;
357
358   /**
359    * Get the local address to which the socket is bound.
360    *
361    * Throws TTransportException on error.
362    */
363   SocketAddress getAddress() const {
364     SocketAddress ret;
365     getAddress(&ret);
366     return ret;
367   }
368
369   /**
370    * Get all the local addresses to which the socket is bound.
371    *
372    * Throws TTransportException on error.
373    */
374   std::vector<SocketAddress> getAddresses() const;
375
376   /**
377    * Begin listening for connections.
378    *
379    * This calls ::listen() with the specified backlog.
380    *
381    * Once listen() is invoked the socket will actually be open so that remote
382    * clients may establish connections.  (Clients that attempt to connect
383    * before listen() is called will receive a connection refused error.)
384    *
385    * At least one callback must be set and startAccepting() must be called to
386    * actually begin notifying the accept callbacks of newly accepted
387    * connections.  The backlog parameter controls how many connections the
388    * kernel will accept and buffer internally while the accept callbacks are
389    * paused (or if accepting is enabled but the callbacks cannot keep up).
390    *
391    * bind() must be called before calling listen().
392    * listen() must be called from the primary EventBase thread.
393    *
394    * Throws TTransportException on error.
395    */
396   virtual void listen(int backlog);
397
398   /**
399    * Add an AcceptCallback.
400    *
401    * When a new socket is accepted, one of the AcceptCallbacks will be invoked
402    * with the new socket.  The AcceptCallbacks are invoked in a round-robin
403    * fashion.  This allows the accepted sockets to be distributed among a pool
404    * of threads, each running its own EventBase object.  This is a common model,
405    * since most asynchronous-style servers typically run one EventBase thread
406    * per CPU.
407    *
408    * The EventBase object associated with each AcceptCallback must be running
409    * its loop.  If the EventBase loop is not running, sockets will still be
410    * scheduled for the callback, but the callback cannot actually get invoked
411    * until the loop runs.
412    *
413    * This method must be invoked from the AsyncServerSocket's primary
414    * EventBase thread.
415    *
416    * Note that startAccepting() must be called on the AsyncServerSocket to
417    * cause it to actually start accepting sockets once callbacks have been
418    * installed.
419    *
420    * @param callback   The callback to invoke.
421    * @param eventBase  The EventBase to use to invoke the callback.  This
422    *     parameter may be nullptr, in which case the callback will be invoked in
423    *     the AsyncServerSocket's primary EventBase.
424    * @param maxAtOnce  The maximum number of connections to accept in this
425    *                   callback on a single iteration of the event base loop.
426    *                   This only takes effect when eventBase is non-nullptr.
427    *                   When using a nullptr eventBase for the callback, the
428    *                   setMaxAcceptAtOnce() method controls how many
429    *                   connections the main event base will accept at once.
430    */
431   virtual void addAcceptCallback(
432     AcceptCallback *callback,
433     EventBase *eventBase,
434     uint32_t maxAtOnce = kDefaultCallbackAcceptAtOnce);
435
436   /**
437    * Remove an AcceptCallback.
438    *
439    * This allows a single AcceptCallback to be removed from the round-robin
440    * pool.
441    *
442    * This method must be invoked from the AsyncServerSocket's primary
443    * EventBase thread.  Use EventBase::runInEventBaseThread() to schedule the
444    * operation in the correct EventBase if your code is not in the server
445    * socket's primary EventBase.
446    *
447    * Given that the accept callback is being driven by a different EventBase,
448    * the AcceptCallback may continue to be invoked for a short period of time
449    * after removeAcceptCallback() returns in this thread.  Once the other
450    * EventBase thread receives the notification to stop, it will call
451    * acceptStopped() on the callback to inform it that it is fully stopped and
452    * will not receive any new sockets.
453    *
454    * If the last accept callback is removed while the socket is accepting,
455    * the socket will implicitly pause accepting.  If a callback is later added,
456    * it will resume accepting immediately, without requiring startAccepting()
457    * to be invoked.
458    *
459    * @param callback   The callback to uninstall.
460    * @param eventBase  The EventBase associated with this callback.  This must
461    *     be the same EventBase that was used when the callback was installed
462    *     with addAcceptCallback().
463    */
464   void removeAcceptCallback(AcceptCallback *callback, EventBase *eventBase);
465
466   /**
467    * Begin accepting connctions on this socket.
468    *
469    * bind() and listen() must be called before calling startAccepting().
470    *
471    * When a AsyncServerSocket is initially created, it will not begin
472    * accepting connections until at least one callback has been added and
473    * startAccepting() has been called.  startAccepting() can also be used to
474    * resume accepting connections after a call to pauseAccepting().
475    *
476    * If startAccepting() is called when there are no accept callbacks
477    * installed, the socket will not actually begin accepting until an accept
478    * callback is added.
479    *
480    * This method may only be called from the primary EventBase thread.
481    */
482   virtual void startAccepting();
483
484   /**
485    * Pause accepting connections.
486    *
487    * startAccepting() may be called to resume accepting.
488    *
489    * This method may only be called from the primary EventBase thread.
490    * If there are AcceptCallbacks being driven by other EventBase threads they
491    * may continue to receive callbacks for a short period of time after
492    * pauseAccepting() returns.
493    *
494    * Unlike removeAcceptCallback() or destroy(), acceptStopped() will not be
495    * called on the AcceptCallback objects simply due to a temporary pause.  If
496    * the server socket is later destroyed while paused, acceptStopped() will be
497    * called all of the installed AcceptCallbacks.
498    */
499   void pauseAccepting();
500
501   /**
502    * Shutdown the listen socket and notify all callbacks that accept has
503    * stopped, but don't close the socket.  This invokes shutdown(2) with the
504    * supplied argument.  Passing -1 will close the socket now.  Otherwise, the
505    * close will be delayed until this object is destroyed.
506    *
507    * Only use this if you have reason to pass special flags to shutdown.
508    * Otherwise just destroy the socket.
509    *
510    * This method has no effect when a ShutdownSocketSet option is used.
511    *
512    * Returns the result of shutdown on sockets_[n-1]
513    */
514   int stopAccepting(int shutdownFlags = -1);
515
516   /**
517    * Get the maximum number of connections that will be accepted each time
518    * around the event loop.
519    */
520   uint32_t getMaxAcceptAtOnce() const {
521     return maxAcceptAtOnce_;
522   }
523
524   /**
525    * Set the maximum number of connections that will be accepted each time
526    * around the event loop.
527    *
528    * This provides a very coarse-grained way of controlling how fast the
529    * AsyncServerSocket will accept connections.  If you find that when your
530    * server is overloaded AsyncServerSocket accepts connections more quickly
531    * than your code can process them, you can try lowering this number so that
532    * fewer connections will be accepted each event loop iteration.
533    *
534    * For more explicit control over the accept rate, you can also use
535    * pauseAccepting() to temporarily pause accepting when your server is
536    * overloaded, and then use startAccepting() later to resume accepting.
537    */
538   void setMaxAcceptAtOnce(uint32_t numConns) {
539     maxAcceptAtOnce_ = numConns;
540   }
541
542   /**
543    * Get the maximum number of unprocessed messages which a NotificationQueue
544    * can hold.
545    */
546   uint32_t getMaxNumMessagesInQueue() const {
547     return maxNumMsgsInQueue_;
548   }
549
550   /**
551    * Set the maximum number of unprocessed messages in NotificationQueue.
552    * No new message will be sent to that NotificationQueue if there are more
553    * than such number of unprocessed messages in that queue.
554    *
555    * Only works if called before addAcceptCallback.
556    */
557   void setMaxNumMessagesInQueue(uint32_t num) {
558     maxNumMsgsInQueue_ = num;
559   }
560
561   /**
562    * Get the speed of adjusting connection accept rate.
563    */
564   double getAcceptRateAdjustSpeed() const {
565     return acceptRateAdjustSpeed_;
566   }
567
568   /**
569    * Set the speed of adjusting connection accept rate.
570    */
571   void setAcceptRateAdjustSpeed(double speed) {
572     acceptRateAdjustSpeed_ = speed;
573   }
574
575   /**
576    * Get the number of connections dropped by the AsyncServerSocket
577    */
578   uint64_t getNumDroppedConnections() const {
579     return numDroppedConnections_;
580   }
581
582   /**
583    * Get the current number of unprocessed messages in NotificationQueue.
584    *
585    * This method must be invoked from the AsyncServerSocket's primary
586    * EventBase thread.  Use EventBase::runInEventBaseThread() to schedule the
587    * operation in the correct EventBase if your code is not in the server
588    * socket's primary EventBase.
589    */
590   int64_t getNumPendingMessagesInQueue() const {
591     assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
592     int64_t numMsgs = 0;
593     for (const auto& callback : callbacks_) {
594       numMsgs += callback.consumer->getQueue()->size();
595     }
596     return numMsgs;
597   }
598
599   /**
600    * Set whether or not SO_KEEPALIVE should be enabled on the server socket
601    * (and thus on all subsequently-accepted connections). By default, keepalive
602    * is enabled.
603    *
604    * Note that TCP keepalive usually only kicks in after the connection has
605    * been idle for several hours. Applications should almost always have their
606    * own, shorter idle timeout.
607    */
608   void setKeepAliveEnabled(bool enabled) {
609     keepAliveEnabled_ = enabled;
610
611     for (auto& handler : sockets_) {
612       if (handler.socket_ < 0) {
613         continue;
614       }
615
616       int val = (enabled) ? 1 : 0;
617       if (setsockopt(handler.socket_, SOL_SOCKET,
618                      SO_KEEPALIVE, &val, sizeof(val)) != 0) {
619         LOG(ERROR) << "failed to set SO_KEEPALIVE on async server socket: %s" <<
620                 strerror(errno);
621       }
622     }
623   }
624
625   /**
626    * Get whether or not SO_KEEPALIVE is enabled on the server socket.
627    */
628   bool getKeepAliveEnabled() const {
629     return keepAliveEnabled_;
630   }
631
632   /**
633    * Set whether or not SO_REUSEPORT should be enabled on the server socket,
634    * allowing multiple binds to the same port
635    */
636   void setReusePortEnabled(bool enabled) {
637     reusePortEnabled_ = enabled;
638
639     for (auto& handler : sockets_) {
640       if (handler.socket_ < 0) {
641         continue;
642       }
643
644       int val = (enabled) ? 1 : 0;
645       if (setsockopt(handler.socket_, SOL_SOCKET,
646                      SO_REUSEPORT, &val, sizeof(val)) != 0) {
647         LOG(ERROR) <<
648           "failed to set SO_REUSEPORT on async server socket " << errno;
649         folly::throwSystemError(errno,
650                                 "failed to bind to async server socket");
651       }
652     }
653   }
654
655   /**
656    * Get whether or not SO_REUSEPORT is enabled on the server socket.
657    */
658   bool getReusePortEnabled_() const {
659     return reusePortEnabled_;
660   }
661
662   /**
663    * Set whether or not the socket should close during exec() (FD_CLOEXEC). By
664    * default, this is enabled
665    */
666   void setCloseOnExec(bool closeOnExec) {
667     closeOnExec_ = closeOnExec;
668   }
669
670   /**
671    * Get whether or not FD_CLOEXEC is enabled on the server socket.
672    */
673   bool getCloseOnExec() const {
674     return closeOnExec_;
675   }
676
677   /**
678    * Tries to enable TFO if the machine supports it.
679    */
680   void setTFOEnabled(bool enabled, uint32_t maxTFOQueueSize) {
681     tfo_ = enabled;
682     tfoMaxQueueSize_ = maxTFOQueueSize;
683   }
684
685   /**
686    * Do not attempt the transparent TLS handshake
687    */
688   void disableTransparentTls() {
689     noTransparentTls_ = true;
690   }
691
692   /**
693    * Get whether or not the socket is accepting new connections
694    */
695   bool getAccepting() const {
696     return accepting_;
697   }
698
699   /**
700    * Set the ConnectionEventCallback
701    */
702   void setConnectionEventCallback(
703       ConnectionEventCallback* const connectionEventCallback) {
704     connectionEventCallback_ = connectionEventCallback;
705   }
706
707   /**
708    * Get the ConnectionEventCallback
709    */
710   ConnectionEventCallback* getConnectionEventCallback() const {
711     return connectionEventCallback_;
712   }
713
714  protected:
715   /**
716    * Protected destructor.
717    *
718    * Invoke destroy() instead to destroy the AsyncServerSocket.
719    */
720   virtual ~AsyncServerSocket();
721
722  private:
723   enum class MessageType {
724     MSG_NEW_CONN = 0,
725     MSG_ERROR = 1
726   };
727
728   struct QueueMessage {
729     MessageType type;
730     int fd;
731     int err;
732     SocketAddress address;
733     std::string msg;
734   };
735
736   /**
737    * A class to receive notifications to invoke AcceptCallback objects
738    * in other EventBase threads.
739    *
740    * A RemoteAcceptor object is created for each AcceptCallback that
741    * is installed in a separate EventBase thread.  The RemoteAcceptor
742    * receives notification of new sockets via a NotificationQueue,
743    * and then invokes the AcceptCallback.
744    */
745   class RemoteAcceptor
746       : private NotificationQueue<QueueMessage>::Consumer {
747   public:
748     explicit RemoteAcceptor(AcceptCallback *callback,
749                             ConnectionEventCallback *connectionEventCallback)
750       : callback_(callback),
751         connectionEventCallback_(connectionEventCallback) {}
752
753     ~RemoteAcceptor() = default;
754
755     void start(EventBase *eventBase, uint32_t maxAtOnce, uint32_t maxInQueue);
756     void stop(EventBase* eventBase, AcceptCallback* callback);
757
758     virtual void messageAvailable(QueueMessage&& message);
759
760     NotificationQueue<QueueMessage>* getQueue() {
761       return &queue_;
762     }
763
764   private:
765     AcceptCallback *callback_;
766     ConnectionEventCallback* connectionEventCallback_;
767
768     NotificationQueue<QueueMessage> queue_;
769   };
770
771   /**
772    * A struct to keep track of the callbacks associated with this server
773    * socket.
774    */
775   struct CallbackInfo {
776     CallbackInfo(AcceptCallback *cb, EventBase *evb)
777       : callback(cb),
778         eventBase(evb),
779         consumer(nullptr) {}
780
781     AcceptCallback *callback;
782     EventBase *eventBase;
783
784     RemoteAcceptor* consumer;
785   };
786
787   class BackoffTimeout;
788
789   virtual void handlerReady(
790     uint16_t events, int socket, sa_family_t family) noexcept;
791
792   int createSocket(int family);
793   void setupSocket(int fd, int family);
794   void bindSocket(int fd, const SocketAddress& address, bool isExistingSocket);
795   void dispatchSocket(int socket, SocketAddress&& address);
796   void dispatchError(const char *msg, int errnoValue);
797   void enterBackoff();
798   void backoffTimeoutExpired();
799
800   CallbackInfo* nextCallback() {
801     CallbackInfo* info = &callbacks_[callbackIndex_];
802
803     ++callbackIndex_;
804     if (callbackIndex_ >= callbacks_.size()) {
805       callbackIndex_ = 0;
806     }
807
808     return info;
809   }
810
811   struct ServerEventHandler : public EventHandler {
812     ServerEventHandler(EventBase* eventBase, int socket,
813                        AsyncServerSocket* parent,
814                       sa_family_t addressFamily)
815         : EventHandler(eventBase, socket)
816         , eventBase_(eventBase)
817         , socket_(socket)
818         , parent_(parent)
819         , addressFamily_(addressFamily) {}
820
821     ServerEventHandler(const ServerEventHandler& other)
822     : EventHandler(other.eventBase_, other.socket_)
823     , eventBase_(other.eventBase_)
824     , socket_(other.socket_)
825     , parent_(other.parent_)
826     , addressFamily_(other.addressFamily_) {}
827
828     ServerEventHandler& operator=(
829         const ServerEventHandler& other) {
830       if (this != &other) {
831         eventBase_ = other.eventBase_;
832         socket_ = other.socket_;
833         parent_ = other.parent_;
834         addressFamily_ = other.addressFamily_;
835
836         detachEventBase();
837         attachEventBase(other.eventBase_);
838         changeHandlerFD(other.socket_);
839       }
840       return *this;
841     }
842
843     // Inherited from EventHandler
844     virtual void handlerReady(uint16_t events) noexcept {
845       parent_->handlerReady(events, socket_, addressFamily_);
846     }
847
848     EventBase* eventBase_;
849     int socket_;
850     AsyncServerSocket* parent_;
851     sa_family_t addressFamily_;
852   };
853
854   EventBase *eventBase_;
855   std::vector<ServerEventHandler> sockets_;
856   std::vector<int> pendingCloseSockets_;
857   bool accepting_;
858   uint32_t maxAcceptAtOnce_;
859   uint32_t maxNumMsgsInQueue_;
860   double acceptRateAdjustSpeed_;  //0 to disable auto adjust
861   double acceptRate_;
862   std::chrono::time_point<std::chrono::steady_clock> lastAccepTimestamp_;
863   uint64_t numDroppedConnections_;
864   uint32_t callbackIndex_;
865   BackoffTimeout *backoffTimeout_;
866   std::vector<CallbackInfo> callbacks_;
867   bool keepAliveEnabled_;
868   bool reusePortEnabled_{false};
869   bool closeOnExec_;
870   bool tfo_{false};
871   bool noTransparentTls_{false};
872   uint32_t tfoMaxQueueSize_{0};
873   ShutdownSocketSet* shutdownSocketSet_;
874   ConnectionEventCallback* connectionEventCallback_{nullptr};
875 };
876
877 } // folly