2 * Copyright 2014-present Facebook, Inc.
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
8 * http://www.apache.org/licenses/LICENSE-2.0
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.
18 #include <folly/Optional.h>
19 #include <folly/SocketAddress.h>
20 #include <folly/detail/SocketFastOpen.h>
21 #include <folly/io/IOBuf.h>
22 #include <folly/io/ShutdownSocketSet.h>
23 #include <folly/io/async/AsyncSocketException.h>
24 #include <folly/io/async/AsyncTimeout.h>
25 #include <folly/io/async/AsyncTransport.h>
26 #include <folly/io/async/DelayedDestruction.h>
27 #include <folly/io/async/EventHandler.h>
28 #include <folly/portability/Sockets.h>
30 #include <sys/types.h>
39 * A class for performing asynchronous I/O on a socket.
41 * AsyncSocket allows users to asynchronously wait for data on a socket, and
42 * to asynchronously send data.
44 * The APIs for reading and writing are intentionally asymmetric. Waiting for
45 * data to read is a persistent API: a callback is installed, and is notified
46 * whenever new data is available. It continues to be notified of new events
47 * until it is uninstalled.
49 * AsyncSocket does not provide read timeout functionality, because it
50 * typically cannot determine when the timeout should be active. Generally, a
51 * timeout should only be enabled when processing is blocked waiting on data
52 * from the remote endpoint. For server sockets, the timeout should not be
53 * active if the server is currently processing one or more outstanding
54 * requests for this socket. For client sockets, the timeout should not be
55 * active if there are no requests pending on the socket. Additionally, if a
56 * client has multiple pending requests, it will ususally want a separate
57 * timeout for each request, rather than a single read timeout.
59 * The write API is fairly intuitive: a user can request to send a block of
60 * data, and a callback will be informed once the entire block has been
61 * transferred to the kernel, or on error. AsyncSocket does provide a send
62 * timeout, since most callers want to give up if the remote end stops
63 * responding and no further progress can be made sending the data.
66 #if defined __linux__ && !defined SO_NO_TRANSPARENT_TLS
67 #define SO_NO_TRANSPARENT_TLS 200
70 #if defined __linux__ && !defined SO_NO_TSOCKS
71 #define SO_NO_TSOCKS 201
75 // We do a dynamic_cast on this, in
76 // AsyncTransportWrapper::getUnderlyingTransport so be safe and
77 // force displacements for it. See:
78 // https://msdn.microsoft.com/en-us/library/7sf3txa8.aspx
79 #pragma vtordisp(push, 2)
81 class AsyncSocket : virtual public AsyncTransportWrapper {
83 typedef std::unique_ptr<AsyncSocket, Destructor> UniquePtr;
85 class ConnectCallback {
87 virtual ~ConnectCallback() = default;
90 * connectSuccess() will be invoked when the connection has been
91 * successfully established.
93 virtual void connectSuccess() noexcept = 0;
96 * connectErr() will be invoked if the connection attempt fails.
98 * @param ex An exception describing the error that occurred.
100 virtual void connectErr(const AsyncSocketException& ex)
104 class EvbChangeCallback {
106 virtual ~EvbChangeCallback() = default;
108 // Called when the socket has been attached to a new EVB
109 // and is called from within that EVB thread
110 virtual void evbAttached(AsyncSocket* socket) = 0;
112 // Called when the socket is detached from an EVB and
113 // is called from the EVB thread being detached
114 virtual void evbDetached(AsyncSocket* socket) = 0;
118 * This interface is implemented only for platforms supporting
119 * per-socket error queues.
121 class ErrMessageCallback {
123 virtual ~ErrMessageCallback() = default;
126 * errMessage() will be invoked when kernel puts a message to
127 * the error queue associated with the socket.
129 * @param cmsg Reference to cmsghdr structure describing
130 * a message read from error queue associated
134 errMessage(const cmsghdr& cmsg) noexcept = 0;
137 * errMessageError() will be invoked if an error occurs reading a message
138 * from the socket error stream.
140 * @param ex An exception describing the error that occurred.
142 virtual void errMessageError(const AsyncSocketException& ex) noexcept = 0;
145 class SendMsgParamsCallback {
147 virtual ~SendMsgParamsCallback() = default;
150 * getFlags() will be invoked to retrieve the desired flags to be passed
151 * to ::sendmsg() system call. This method was intentionally declared
152 * non-virtual, so there is no way to override it. Instead feel free to
153 * override getFlagsImpl(flags, defaultFlags) method instead, and enjoy
154 * the convenience of defaultFlags passed there.
156 * @param flags Write flags requested for the given write operation
158 int getFlags(folly::WriteFlags flags, bool zeroCopyEnabled) noexcept {
159 return getFlagsImpl(flags, getDefaultFlags(flags, zeroCopyEnabled));
163 * getAncillaryData() will be invoked to initialize ancillary data
164 * buffer referred by "msg_control" field of msghdr structure passed to
165 * ::sendmsg() system call. The function assumes that the size of buffer
166 * is not smaller than the value returned by getAncillaryDataSize() method
167 * for the same combination of flags.
169 * @param flags Write flags requested for the given write operation
170 * @param data Pointer to ancillary data buffer to initialize.
172 virtual void getAncillaryData(
173 folly::WriteFlags /*flags*/,
174 void* /*data*/) noexcept {}
177 * getAncillaryDataSize() will be invoked to retrieve the size of
178 * ancillary data buffer which should be passed to ::sendmsg() system call
180 * @param flags Write flags requested for the given write operation
182 virtual uint32_t getAncillaryDataSize(folly::WriteFlags /*flags*/)
187 static const size_t maxAncillaryDataSize{0x5000};
191 * getFlagsImpl() will be invoked by getFlags(folly::WriteFlags flags)
192 * method to retrieve the flags to be passed to ::sendmsg() system call.
193 * SendMsgParamsCallback::getFlags() is calling this method, and returns
194 * its results directly to the caller in AsyncSocket.
195 * Classes inheriting from SendMsgParamsCallback are welcome to override
196 * this method to force SendMsgParamsCallback to return its own set
199 * @param flags Write flags requested for the given write operation
200 * @param defaultflags A set of message flags returned by getDefaultFlags()
201 * method for the given "flags" mask.
203 virtual int getFlagsImpl(folly::WriteFlags /*flags*/, int defaultFlags) {
208 * getDefaultFlags() will be invoked by getFlags(folly::WriteFlags flags)
209 * to retrieve the default set of flags, and pass them to getFlagsImpl(...)
211 * @param flags Write flags requested for the given write operation
213 int getDefaultFlags(folly::WriteFlags flags, bool zeroCopyEnabled) noexcept;
216 explicit AsyncSocket();
218 * Create a new unconnected AsyncSocket.
220 * connect() must later be called on this socket to establish a connection.
222 explicit AsyncSocket(EventBase* evb);
224 void setShutdownSocketSet(const std::weak_ptr<ShutdownSocketSet>& wSS);
227 * Create a new AsyncSocket and begin the connection process.
229 * @param evb EventBase that will manage this socket.
230 * @param address The address to connect to.
231 * @param connectTimeout Optional timeout in milliseconds for the connection
234 AsyncSocket(EventBase* evb,
235 const folly::SocketAddress& address,
236 uint32_t connectTimeout = 0);
239 * Create a new AsyncSocket and begin the connection process.
241 * @param evb EventBase that will manage this socket.
242 * @param ip IP address to connect to (dotted-quad).
243 * @param port Destination port in host byte order.
244 * @param connectTimeout Optional timeout in milliseconds for the connection
247 AsyncSocket(EventBase* evb,
248 const std::string& ip,
250 uint32_t connectTimeout = 0);
253 * Create a AsyncSocket from an already connected socket file descriptor.
255 * Note that while AsyncSocket enables TCP_NODELAY for sockets it creates
256 * when connecting, it does not change the socket options when given an
257 * existing file descriptor. If callers want TCP_NODELAY enabled when using
258 * this version of the constructor, they need to explicitly call
259 * setNoDelay(true) after the constructor returns.
261 * @param evb EventBase that will manage this socket.
262 * @param fd File descriptor to take over (should be a connected socket).
263 * @param zeroCopyBufId Zerocopy buf id to start with.
265 AsyncSocket(EventBase* evb, int fd, uint32_t zeroCopyBufId = 0);
268 * Create an AsyncSocket from a different, already connected AsyncSocket.
270 * Similar to AsyncSocket(evb, fd) when fd was previously owned by an
273 explicit AsyncSocket(AsyncSocket::UniquePtr);
276 * Helper function to create a shared_ptr<AsyncSocket>.
278 * This passes in the correct destructor object, since AsyncSocket's
279 * destructor is protected and cannot be invoked directly.
281 static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb) {
282 return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb),
287 * Helper function to create a shared_ptr<AsyncSocket>.
289 static std::shared_ptr<AsyncSocket> newSocket(
291 const folly::SocketAddress& address,
292 uint32_t connectTimeout = 0) {
293 return std::shared_ptr<AsyncSocket>(
294 new AsyncSocket(evb, address, connectTimeout),
299 * Helper function to create a shared_ptr<AsyncSocket>.
301 static std::shared_ptr<AsyncSocket> newSocket(
303 const std::string& ip,
305 uint32_t connectTimeout = 0) {
306 return std::shared_ptr<AsyncSocket>(
307 new AsyncSocket(evb, ip, port, connectTimeout),
312 * Helper function to create a shared_ptr<AsyncSocket>.
314 static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb, int fd) {
315 return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb, fd),
320 * Destroy the socket.
322 * AsyncSocket::destroy() must be called to destroy the socket.
323 * The normal destructor is private, and should not be invoked directly.
324 * This prevents callers from deleting a AsyncSocket while it is invoking a
327 void destroy() override;
330 * Get the EventBase used by this socket.
332 EventBase* getEventBase() const override {
337 * Get the file descriptor used by the AsyncSocket.
339 virtual int getFd() const {
344 * Extract the file descriptor from the AsyncSocket.
346 * This will immediately cause any installed callbacks to be invoked with an
347 * error. The AsyncSocket may no longer be used after the file descriptor
348 * has been extracted.
350 * This method should be used with care as the resulting fd is not guaranteed
351 * to perfectly reflect the state of the AsyncSocket (security state,
352 * pre-received data, etc.).
354 * Returns the file descriptor. The caller assumes ownership of the
355 * descriptor, and it will not be closed when the AsyncSocket is destroyed.
357 virtual int detachFd();
360 * Uniquely identifies a handle to a socket option value. Each
361 * combination of level and option name corresponds to one socket
366 bool operator<(const OptionKey& other) const {
367 if (level == other.level) {
368 return optname < other.optname;
370 return level < other.level;
372 int apply(int fd, int val) const {
373 return setsockopt(fd, level, optname, &val, sizeof(val));
379 // Maps from a socket option key to its value
380 typedef std::map<OptionKey, int> OptionMap;
382 static const OptionMap emptyOptionMap;
383 static const folly::SocketAddress& anyAddress();
386 * Initiate a connection.
388 * @param callback The callback to inform when the connection attempt
390 * @param address The address to connect to.
391 * @param timeout A timeout value, in milliseconds. If the connection
392 * does not succeed within this period,
393 * callback->connectError() will be invoked.
395 virtual void connect(
396 ConnectCallback* callback,
397 const folly::SocketAddress& address,
399 const OptionMap& options = emptyOptionMap,
400 const folly::SocketAddress& bindAddr = anyAddress()) noexcept;
403 ConnectCallback* callback,
404 const std::string& ip,
407 const OptionMap& options = emptyOptionMap) noexcept;
410 * If a connect request is in-flight, cancels it and closes the socket
411 * immediately. Otherwise, this is a no-op.
413 * This does not invoke any connection related callbacks. Call this to
414 * prevent any connect callback while cleaning up, etc.
416 void cancelConnect();
419 * Set the send timeout.
421 * If write requests do not make any progress for more than the specified
422 * number of milliseconds, fail all pending writes and close the socket.
424 * If write requests are currently pending when setSendTimeout() is called,
425 * the timeout interval is immediately restarted using the new value.
427 * (See the comments for AsyncSocket for an explanation of why AsyncSocket
428 * provides setSendTimeout() but not setRecvTimeout().)
430 * @param milliseconds The timeout duration, in milliseconds. If 0, no
431 * timeout will be used.
433 void setSendTimeout(uint32_t milliseconds) override;
436 * Get the send timeout.
438 * @return Returns the current send timeout, in milliseconds. A return value
439 * of 0 indicates that no timeout is set.
441 uint32_t getSendTimeout() const override {
446 * Set the maximum number of reads to execute from the underlying
447 * socket each time the EventBase detects that new ingress data is
448 * available. The default is unlimited, but callers can use this method
449 * to limit the amount of data read from the socket per event loop
452 * @param maxReads Maximum number of reads per data-available event;
453 * a value of zero means unlimited.
455 void setMaxReadsPerEvent(uint16_t maxReads) {
456 maxReadsPerEvent_ = maxReads;
460 * Get the maximum number of reads this object will execute from
461 * the underlying socket each time the EventBase detects that new
462 * ingress data is available.
464 * @returns Maximum number of reads per data-available event; a value
465 * of zero means unlimited.
467 uint16_t getMaxReadsPerEvent() const {
468 return maxReadsPerEvent_;
472 * Set a pointer to ErrMessageCallback implementation which will be
473 * receiving notifications for messages posted to the error queue
474 * associated with the socket.
475 * ErrMessageCallback is implemented only for platforms with
476 * per-socket error message queus support (recvmsg() system call must
480 virtual void setErrMessageCB(ErrMessageCallback* callback);
483 * Get a pointer to ErrMessageCallback implementation currently
484 * registered with this socket.
487 virtual ErrMessageCallback* getErrMessageCallback() const;
490 * Set a pointer to SendMsgParamsCallback implementation which
491 * will be used to form ::sendmsg() system call parameters
494 virtual void setSendMsgParamCB(SendMsgParamsCallback* callback);
497 * Get a pointer to SendMsgParamsCallback implementation currently
498 * registered with this socket.
501 virtual SendMsgParamsCallback* getSendMsgParamsCB() const;
503 // Read and write methods
504 void setReadCB(ReadCallback* callback) override;
505 ReadCallback* getReadCallback() const override;
507 bool setZeroCopy(bool enable);
508 bool getZeroCopy() const {
509 return zeroCopyEnabled_;
512 uint32_t getZeroCopyBufId() const {
513 return zeroCopyBufId_;
516 void write(WriteCallback* callback, const void* buf, size_t bytes,
517 WriteFlags flags = WriteFlags::NONE) override;
518 void writev(WriteCallback* callback, const iovec* vec, size_t count,
519 WriteFlags flags = WriteFlags::NONE) override;
520 void writeChain(WriteCallback* callback,
521 std::unique_ptr<folly::IOBuf>&& buf,
522 WriteFlags flags = WriteFlags::NONE) override;
525 virtual void writeRequest(WriteRequest* req);
526 void writeRequestReady() {
530 // Methods inherited from AsyncTransport
531 void close() override;
532 void closeNow() override;
533 void closeWithReset() override;
534 void shutdownWrite() override;
535 void shutdownWriteNow() override;
537 bool readable() const override;
538 bool writable() const override;
539 bool isPending() const override;
540 virtual bool hangup() const;
541 bool good() const override;
542 bool error() const override;
543 void attachEventBase(EventBase* eventBase) override;
544 void detachEventBase() override;
545 bool isDetachable() const override;
547 void getLocalAddress(
548 folly::SocketAddress* address) const override;
550 folly::SocketAddress* address) const override;
552 bool isEorTrackingEnabled() const override {
556 void setEorTracking(bool track) override {
560 bool connecting() const override {
561 return (state_ == StateEnum::CONNECTING);
564 virtual bool isClosedByPeer() const {
565 return (state_ == StateEnum::CLOSED &&
566 (readErr_ == READ_EOF || readErr_ == READ_ERROR));
569 virtual bool isClosedBySelf() const {
570 return (state_ == StateEnum::CLOSED &&
571 (readErr_ != READ_EOF && readErr_ != READ_ERROR));
574 size_t getAppBytesWritten() const override {
575 return appBytesWritten_;
578 size_t getRawBytesWritten() const override {
579 return getAppBytesWritten();
582 size_t getAppBytesReceived() const override {
583 return appBytesReceived_;
586 size_t getRawBytesReceived() const override {
587 return getAppBytesReceived();
590 std::chrono::nanoseconds getConnectTime() const {
591 return connectEndTime_ - connectStartTime_;
594 std::chrono::milliseconds getConnectTimeout() const {
595 return connectTimeout_;
598 bool getTFOAttempted() const {
599 return tfoAttempted_;
603 * Returns whether or not the attempt to use TFO
604 * finished successfully. This does not necessarily
605 * mean TFO worked, just that trying to use TFO
608 bool getTFOFinished() const {
613 * Returns whether or not TFO attempt succeded on this
615 * For servers this is pretty straightforward API and can
616 * be invoked right after the connection is accepted. This API
617 * will perform one syscall.
618 * This API is a bit tricky to use for clients, since clients
619 * only know this for sure after the SYN-ACK is returned. So it's
620 * appropriate to call this only after the first application
621 * data is read from the socket when the caller knows that
622 * the SYN has been ACKed by the server.
624 bool getTFOSucceded() const;
626 // Methods controlling socket options
629 * Force writes to be transmitted immediately.
631 * This controls the TCP_NODELAY socket option. When enabled, TCP segments
632 * are sent as soon as possible, even if it is not a full frame of data.
633 * When disabled, the data may be buffered briefly to try and wait for a full
636 * By default, TCP_NODELAY is enabled for AsyncSocket objects.
638 * This method will fail if the socket is not currently open.
640 * @return Returns 0 if the TCP_NODELAY flag was successfully updated,
641 * or a non-zero errno value on error.
643 int setNoDelay(bool noDelay);
647 * Set the FD_CLOEXEC flag so that the socket will be closed if the program
648 * later forks and execs.
650 void setCloseOnExec();
653 * Set the Flavor of Congestion Control to be used for this Socket
654 * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
655 * first to make sure the module is available for plugging in
656 * Alternatively you can choose from net.ipv4.tcp_allowed_congestion_control
658 int setCongestionFlavor(const std::string &cname);
661 * Forces ACKs to be sent immediately
663 * @return Returns 0 if the TCP_QUICKACK flag was successfully updated,
664 * or a non-zero errno value on error.
666 int setQuickAck(bool quickack);
669 * Set the send bufsize
671 int setSendBufSize(size_t bufsize);
674 * Set the recv bufsize
676 int setRecvBufSize(size_t bufsize);
679 * Sets a specific tcp personality
680 * Available only on kernels 3.2 and greater
682 #define SO_SET_NAMESPACE 41
683 int setTCPProfile(int profd);
686 * Generic API for reading a socket option.
688 * @param level same as the "level" parameter in getsockopt().
689 * @param optname same as the "optname" parameter in getsockopt().
690 * @param optval pointer to the variable in which the option value should
692 * @param optlen value-result argument, initially containing the size of
693 * the buffer pointed to by optval, and modified on return
694 * to indicate the actual size of the value returned.
695 * @return same as the return value of getsockopt().
697 template <typename T>
698 int getSockOpt(int level, int optname, T* optval, socklen_t* optlen) {
699 return getsockopt(fd_, level, optname, (void*) optval, optlen);
703 * Generic API for setting a socket option.
705 * @param level same as the "level" parameter in getsockopt().
706 * @param optname same as the "optname" parameter in getsockopt().
707 * @param optval the option value to set.
708 * @return same as the return value of setsockopt().
710 template <typename T>
711 int setSockOpt(int level, int optname, const T *optval) {
712 return setsockopt(fd_, level, optname, optval, sizeof(T));
716 * Virtual method for reading a socket option returning integer
717 * value, which is the most typical case. Convenient for overriding
720 * @param level same as the "level" parameter in getsockopt().
721 * @param optname same as the "optname" parameter in getsockopt().
722 * @param optval same as "optval" parameter in getsockopt().
723 * @param optlen same as "optlen" parameter in getsockopt().
724 * @return same as the return value of getsockopt().
727 getSockOptVirtual(int level, int optname, void* optval, socklen_t* optlen) {
728 return getsockopt(fd_, level, optname, optval, optlen);
732 * Virtual method for setting a socket option accepting integer
733 * value, which is the most typical case. Convenient for overriding
736 * @param level same as the "level" parameter in setsockopt().
737 * @param optname same as the "optname" parameter in setsockopt().
738 * @param optval same as "optval" parameter in setsockopt().
739 * @param optlen same as "optlen" parameter in setsockopt().
740 * @return same as the return value of setsockopt().
742 virtual int setSockOptVirtual(
747 return setsockopt(fd_, level, optname, optval, optlen);
751 * Set pre-received data, to be returned to read callback before any data
754 virtual void setPreReceivedData(std::unique_ptr<IOBuf> data) {
755 if (preReceivedData_) {
756 preReceivedData_->prependChain(std::move(data));
758 preReceivedData_ = std::move(data);
763 * Enables TFO behavior on the AsyncSocket if FOLLY_ALLOW_TFO
767 // No-op if folly does not allow tfo
773 void disableTransparentTls() {
774 noTransparentTls_ = true;
777 void disableTSocks() {
781 enum class StateEnum : uint8_t {
790 void setBufferCallback(BufferCallback* cb);
792 // Callers should set this prior to connecting the socket for the safest
794 void setEvbChangedCallback(std::unique_ptr<EvbChangeCallback> cb) {
795 evbChangeCb_ = std::move(cb);
799 * Attempt to cache the current local and peer addresses (if not already
800 * cached) so that they are available from getPeerAddress() and
801 * getLocalAddress() even after the socket is closed.
803 void cacheAddresses();
806 * Returns true if there is any zero copy write in progress
807 * Needs to be called from within the socket's EVB thread
809 bool isZeroCopyWriteInProgress() const noexcept;
812 * Tries to process the msg error queue
813 * And returns true if there are no more zero copy writes in progress
815 bool processZeroCopyWriteInProgress() noexcept;
818 * writeReturn is the total number of bytes written, or WRITE_ERROR on error.
819 * If no data has been written, 0 is returned.
820 * exception is a more specific exception that cause a write error.
821 * Not all writes have exceptions associated with them thus writeReturn
822 * should be checked to determine whether the operation resulted in an error.
825 explicit WriteResult(ssize_t ret) : writeReturn(ret) {}
827 WriteResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
828 : writeReturn(ret), exception(std::move(e)) {}
831 std::unique_ptr<const AsyncSocketException> exception;
835 * readReturn is the number of bytes read, or READ_EOF on EOF, or
836 * READ_ERROR on error, or READ_BLOCKING if the operation will
838 * exception is a more specific exception that may have caused a read error.
839 * Not all read errors have exceptions associated with them thus readReturn
840 * should be checked to determine whether the operation resulted in an error.
843 explicit ReadResult(ssize_t ret) : readReturn(ret) {}
845 ReadResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
846 : readReturn(ret), exception(std::move(e)) {}
849 std::unique_ptr<const AsyncSocketException> exception;
853 * A WriteRequest object tracks information about a pending write operation.
857 WriteRequest(AsyncSocket* socket, WriteCallback* callback) :
858 socket_(socket), callback_(callback) {}
860 virtual void start() {}
862 virtual void destroy() = 0;
864 virtual WriteResult performWrite() = 0;
866 virtual void consume() = 0;
868 virtual bool isComplete() = 0;
870 WriteRequest* getNext() const {
874 WriteCallback* getCallback() const {
878 uint32_t getTotalBytesWritten() const {
879 return totalBytesWritten_;
882 void append(WriteRequest* next) {
883 assert(next_ == nullptr);
887 void fail(const char* fn, const AsyncSocketException& ex) {
888 socket_->failWrite(fn, ex);
891 void bytesWritten(size_t count) {
892 totalBytesWritten_ += uint32_t(count);
893 socket_->appBytesWritten_ += count;
897 // protected destructor, to ensure callers use destroy()
898 virtual ~WriteRequest() {}
900 AsyncSocket* socket_; ///< parent socket
901 WriteRequest* next_{nullptr}; ///< pointer to next WriteRequest
902 WriteCallback* callback_; ///< completion callback
903 uint32_t totalBytesWritten_{0}; ///< total bytes written
907 enum ReadResultEnum {
914 enum WriteResultEnum {
919 * Protected destructor.
921 * Users of AsyncSocket must never delete it directly. Instead, invoke
922 * destroy() instead. (See the documentation in DelayedDestruction.h for
925 ~AsyncSocket() override;
927 friend std::ostream& operator << (std::ostream& os, const StateEnum& state);
930 /// shutdownWrite() called, but we are still waiting on writes to drain
931 SHUT_WRITE_PENDING = 0x01,
932 /// writes have been completely shut down
935 * Reads have been shutdown.
937 * At the moment we don't distinguish between remote read shutdown
938 * (received EOF from the remote end) and local read shutdown. We can
939 * only receive EOF when a read callback is set, and we immediately inform
940 * it of the EOF. Therefore there doesn't seem to be any reason to have a
941 * separate state of "received EOF but the local side may still want to
944 * We also don't currently provide any API for only shutting down the read
945 * side of a socket. (This is a no-op as far as TCP is concerned, anyway.)
950 class BytesWriteRequest;
952 class WriteTimeout : public AsyncTimeout {
954 WriteTimeout(AsyncSocket* socket, EventBase* eventBase)
955 : AsyncTimeout(eventBase)
958 void timeoutExpired() noexcept override {
959 socket_->timeoutExpired();
963 AsyncSocket* socket_;
966 class IoHandler : public EventHandler {
968 IoHandler(AsyncSocket* socket, EventBase* eventBase)
969 : EventHandler(eventBase, -1)
971 IoHandler(AsyncSocket* socket, EventBase* eventBase, int fd)
972 : EventHandler(eventBase, fd)
975 void handlerReady(uint16_t events) noexcept override {
976 socket_->ioReady(events);
980 AsyncSocket* socket_;
985 class ImmediateReadCB : public folly::EventBase::LoopCallback {
987 explicit ImmediateReadCB(AsyncSocket* socket) : socket_(socket) {}
988 void runLoopCallback() noexcept override {
989 DestructorGuard dg(socket_);
990 socket_->checkForImmediateRead();
993 AsyncSocket* socket_;
997 * Schedule checkForImmediateRead to be executed in the next loop
1000 void scheduleImmediateRead() noexcept {
1002 eventBase_->runInLoop(&immediateReadHandler_);
1007 * Schedule handleInitalReadWrite to run in the next iteration.
1009 void scheduleInitialReadWrite() noexcept {
1011 DestructorGuard dg(this);
1012 eventBase_->runInLoop([this, dg] {
1014 handleInitialReadWrite();
1020 // event notification methods
1021 void ioReady(uint16_t events) noexcept;
1022 virtual void checkForImmediateRead() noexcept;
1023 virtual void handleInitialReadWrite() noexcept;
1024 virtual void prepareReadBuffer(void** buf, size_t* buflen);
1025 virtual size_t handleErrMessages() noexcept;
1026 virtual void handleRead() noexcept;
1027 virtual void handleWrite() noexcept;
1028 virtual void handleConnect() noexcept;
1029 void timeoutExpired() noexcept;
1032 * Attempt to read from the socket.
1034 * @param buf The buffer to read data into.
1035 * @param buflen The length of the buffer.
1037 * @return Returns a read result. See read result for details.
1039 virtual ReadResult performRead(void** buf, size_t* buflen, size_t* offset);
1042 * Populate an iovec array from an IOBuf and attempt to write it.
1044 * @param callback Write completion/error callback.
1045 * @param vec Target iovec array; caller retains ownership.
1046 * @param count Number of IOBufs to write, beginning at start of buf.
1047 * @param buf Chain of iovecs.
1048 * @param flags set of flags for the underlying write calls, like cork
1050 void writeChainImpl(WriteCallback* callback, iovec* vec,
1051 size_t count, std::unique_ptr<folly::IOBuf>&& buf,
1055 * Write as much data as possible to the socket without blocking,
1056 * and queue up any leftover data to send when the socket can
1057 * handle writes again.
1059 * @param callback The callback to invoke when the write is completed.
1060 * @param vec Array of buffers to write; this method will make a
1061 * copy of the vector (but not the buffers themselves)
1062 * if the write has to be completed asynchronously.
1063 * @param count Number of elements in vec.
1064 * @param buf The IOBuf that manages the buffers referenced by
1065 * vec, or a pointer to nullptr if the buffers are not
1066 * associated with an IOBuf. Note that ownership of
1067 * the IOBuf is transferred here; upon completion of
1068 * the write, the AsyncSocket deletes the IOBuf.
1069 * @param flags Set of write flags.
1071 void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
1072 std::unique_ptr<folly::IOBuf>&& buf,
1073 WriteFlags flags = WriteFlags::NONE);
1076 * Attempt to write to the socket.
1078 * @param vec The iovec array pointing to the buffers to write.
1079 * @param count The length of the iovec array.
1080 * @param flags Set of write flags.
1081 * @param countWritten On return, the value pointed to by this parameter
1082 * will contain the number of iovec entries that were
1084 * @param partialWritten On return, the value pointed to by this parameter
1085 * will contain the number of bytes written in the
1086 * partially written iovec entry.
1088 * @return Returns a WriteResult. See WriteResult for more details.
1090 virtual WriteResult performWrite(
1094 uint32_t* countWritten,
1095 uint32_t* partialWritten);
1098 * Sends the message over the socket using sendmsg
1100 * @param msg Message to send
1101 * @param msg_flags Flags to pass to sendmsg
1103 AsyncSocket::WriteResult
1104 sendSocketMessage(int fd, struct msghdr* msg, int msg_flags);
1106 virtual ssize_t tfoSendMsg(int fd, struct msghdr* msg, int msg_flags);
1108 int socketConnect(const struct sockaddr* addr, socklen_t len);
1110 virtual void scheduleConnectTimeout();
1111 void registerForConnectEvents();
1113 bool updateEventRegistration();
1116 * Update event registration.
1118 * @param enable Flags of events to enable. Set it to 0 if no events
1119 * need to be enabled in this call.
1120 * @param disable Flags of events
1121 * to disable. Set it to 0 if no events need to be disabled in this
1124 * @return true iff the update is successful.
1126 bool updateEventRegistration(uint16_t enable, uint16_t disable);
1128 // Actually close the file descriptor and set it to -1 so we don't
1129 // accidentally close it again.
1132 // error handling methods
1135 void finishFail(const AsyncSocketException& ex);
1136 void invokeAllErrors(const AsyncSocketException& ex);
1137 void fail(const char* fn, const AsyncSocketException& ex);
1138 void failConnect(const char* fn, const AsyncSocketException& ex);
1139 void failRead(const char* fn, const AsyncSocketException& ex);
1140 void failErrMessageRead(const char* fn, const AsyncSocketException& ex);
1141 void failWrite(const char* fn, WriteCallback* callback, size_t bytesWritten,
1142 const AsyncSocketException& ex);
1143 void failWrite(const char* fn, const AsyncSocketException& ex);
1144 void failAllWrites(const AsyncSocketException& ex);
1145 virtual void invokeConnectErr(const AsyncSocketException& ex);
1146 virtual void invokeConnectSuccess();
1147 void invalidState(ConnectCallback* callback);
1148 void invalidState(ErrMessageCallback* callback);
1149 void invalidState(ReadCallback* callback);
1150 void invalidState(WriteCallback* callback);
1152 std::string withAddr(const std::string& s);
1154 void cacheLocalAddress() const;
1155 void cachePeerAddress() const;
1157 bool isZeroCopyRequest(WriteFlags flags);
1159 bool isZeroCopyMsg(const cmsghdr& cmsg) const;
1160 void processZeroCopyMsg(const cmsghdr& cmsg);
1162 uint32_t getNextZeroCopyBufId() {
1163 return zeroCopyBufId_++;
1165 void adjustZeroCopyFlags(folly::WriteFlags& flags);
1166 void addZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf);
1167 void addZeroCopyBuf(folly::IOBuf* ptr);
1168 void setZeroCopyBuf(std::unique_ptr<folly::IOBuf>&& buf);
1169 bool containsZeroCopyBuf(folly::IOBuf* ptr);
1170 void releaseZeroCopyBuf(uint32_t id);
1172 // a folly::IOBuf can be used in multiple partial requests
1173 // there is a that maps a buffer id to a raw folly::IOBuf ptr
1174 // and another one that adds a ref count for a folly::IOBuf that is either
1175 // the original ptr or nullptr
1176 uint32_t zeroCopyBufId_{0};
1180 std::unique_ptr<folly::IOBuf> buf_;
1183 std::unordered_map<uint32_t, folly::IOBuf*> idZeroCopyBufPtrMap_;
1184 std::unordered_map<folly::IOBuf*, IOBufInfo> idZeroCopyBufInfoMap_;
1186 StateEnum state_; ///< StateEnum describing current state
1187 uint8_t shutdownFlags_; ///< Shutdown state (ShutdownFlags)
1188 uint16_t eventFlags_; ///< EventBase::HandlerFlags settings
1189 int fd_; ///< The socket file descriptor
1190 mutable folly::SocketAddress addr_; ///< The address we tried to connect to
1191 mutable folly::SocketAddress localAddr_;
1192 ///< The address we are connecting from
1193 uint32_t sendTimeout_; ///< The send timeout, in milliseconds
1194 uint16_t maxReadsPerEvent_; ///< Max reads per event loop iteration
1196 bool isBufferMovable_{false};
1198 int8_t readErr_{READ_NO_ERROR}; ///< The read error encountered, if any
1200 EventBase* eventBase_; ///< The EventBase
1201 WriteTimeout writeTimeout_; ///< A timeout for connect and write
1202 IoHandler ioHandler_; ///< A EventHandler to monitor the fd
1203 ImmediateReadCB immediateReadHandler_; ///< LoopCallback for checking read
1205 ConnectCallback* connectCallback_; ///< ConnectCallback
1206 ErrMessageCallback* errMessageCallback_; ///< TimestampCallback
1207 SendMsgParamsCallback* ///< Callback for retrieving
1208 sendMsgParamCallback_; ///< ::sendmsg() parameters
1209 ReadCallback* readCallback_; ///< ReadCallback
1210 WriteRequest* writeReqHead_; ///< Chain of WriteRequests
1211 WriteRequest* writeReqTail_; ///< End of WriteRequest chain
1212 std::weak_ptr<ShutdownSocketSet> wShutdownSocketSet_;
1213 size_t appBytesReceived_; ///< Num of bytes received from socket
1214 size_t appBytesWritten_; ///< Num of bytes written to socket
1216 // Pre-received data, to be returned to read callback before any data from the
1218 std::unique_ptr<IOBuf> preReceivedData_;
1220 std::chrono::steady_clock::time_point connectStartTime_;
1221 std::chrono::steady_clock::time_point connectEndTime_;
1223 std::chrono::milliseconds connectTimeout_{0};
1225 std::unique_ptr<EvbChangeCallback> evbChangeCb_{nullptr};
1227 BufferCallback* bufferCallback_{nullptr};
1228 bool tfoEnabled_{false};
1229 bool tfoAttempted_{false};
1230 bool tfoFinished_{false};
1231 bool noTransparentTls_{false};
1232 bool noTSocks_{false};
1233 // Whether to track EOR or not.
1234 bool trackEor_{false};
1235 bool zeroCopyEnabled_{false};
1236 bool zeroCopyVal_{false};
1239 #pragma vtordisp(pop)
1242 } // namespace folly