2 * Copyright 2017 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.
19 #include <folly/Optional.h>
20 #include <folly/SocketAddress.h>
21 #include <folly/detail/SocketFastOpen.h>
22 #include <folly/io/IOBuf.h>
23 #include <folly/io/ShutdownSocketSet.h>
24 #include <folly/io/async/AsyncSocketException.h>
25 #include <folly/io/async/AsyncTimeout.h>
26 #include <folly/io/async/AsyncTransport.h>
27 #include <folly/io/async/DelayedDestruction.h>
28 #include <folly/io/async/EventHandler.h>
29 #include <folly/portability/Sockets.h>
31 #include <sys/types.h>
40 * A class for performing asynchronous I/O on a socket.
42 * AsyncSocket allows users to asynchronously wait for data on a socket, and
43 * to asynchronously send data.
45 * The APIs for reading and writing are intentionally asymmetric. Waiting for
46 * data to read is a persistent API: a callback is installed, and is notified
47 * whenever new data is available. It continues to be notified of new events
48 * until it is uninstalled.
50 * AsyncSocket does not provide read timeout functionality, because it
51 * typically cannot determine when the timeout should be active. Generally, a
52 * timeout should only be enabled when processing is blocked waiting on data
53 * from the remote endpoint. For server sockets, the timeout should not be
54 * active if the server is currently processing one or more outstanding
55 * requests for this socket. For client sockets, the timeout should not be
56 * active if there are no requests pending on the socket. Additionally, if a
57 * client has multiple pending requests, it will ususally want a separate
58 * timeout for each request, rather than a single read timeout.
60 * The write API is fairly intuitive: a user can request to send a block of
61 * data, and a callback will be informed once the entire block has been
62 * transferred to the kernel, or on error. AsyncSocket does provide a send
63 * timeout, since most callers want to give up if the remote end stops
64 * responding and no further progress can be made sending the data.
67 #if defined __linux__ && !defined SO_NO_TRANSPARENT_TLS
68 #define SO_NO_TRANSPARENT_TLS 200
72 // We do a dynamic_cast on this, in
73 // AsyncTransportWrapper::getUnderlyingTransport so be safe and
74 // force displacements for it. See:
75 // https://msdn.microsoft.com/en-us/library/7sf3txa8.aspx
76 #pragma vtordisp(push, 2)
78 class AsyncSocket : virtual public AsyncTransportWrapper {
80 typedef std::unique_ptr<AsyncSocket, Destructor> UniquePtr;
82 class ConnectCallback {
84 virtual ~ConnectCallback() = default;
87 * connectSuccess() will be invoked when the connection has been
88 * successfully established.
90 virtual void connectSuccess() noexcept = 0;
93 * connectErr() will be invoked if the connection attempt fails.
95 * @param ex An exception describing the error that occurred.
97 virtual void connectErr(const AsyncSocketException& ex)
101 class EvbChangeCallback {
103 virtual ~EvbChangeCallback() = default;
105 // Called when the socket has been attached to a new EVB
106 // and is called from within that EVB thread
107 virtual void evbAttached(AsyncSocket* socket) = 0;
109 // Called when the socket is detached from an EVB and
110 // is called from the EVB thread being detached
111 virtual void evbDetached(AsyncSocket* socket) = 0;
115 * This interface is implemented only for platforms supporting
116 * per-socket error queues.
118 class ErrMessageCallback {
120 virtual ~ErrMessageCallback() = default;
123 * errMessage() will be invoked when kernel puts a message to
124 * the error queue associated with the socket.
126 * @param cmsg Reference to cmsghdr structure describing
127 * a message read from error queue associated
131 errMessage(const cmsghdr& cmsg) noexcept = 0;
134 * errMessageError() will be invoked if an error occurs reading a message
135 * from the socket error stream.
137 * @param ex An exception describing the error that occurred.
139 virtual void errMessageError(const AsyncSocketException& ex) noexcept = 0;
142 class SendMsgParamsCallback {
144 virtual ~SendMsgParamsCallback() = default;
147 * getFlags() will be invoked to retrieve the desired flags to be passed
148 * to ::sendmsg() system call. This method was intentionally declared
149 * non-virtual, so there is no way to override it. Instead feel free to
150 * override getFlagsImpl(flags, defaultFlags) method instead, and enjoy
151 * the convenience of defaultFlags passed there.
153 * @param flags Write flags requested for the given write operation
155 int getFlags(folly::WriteFlags flags) noexcept {
156 return getFlagsImpl(flags, getDefaultFlags(flags));
160 * getAncillaryData() will be invoked to initialize ancillary data
161 * buffer referred by "msg_control" field of msghdr structure passed to
162 * ::sendmsg() system call. The function assumes that the size of buffer
163 * is not smaller than the value returned by getAncillaryDataSize() method
164 * for the same combination of flags.
166 * @param flags Write flags requested for the given write operation
167 * @param data Pointer to ancillary data buffer to initialize.
169 virtual void getAncillaryData(
170 folly::WriteFlags /*flags*/,
171 void* /*data*/) noexcept {}
174 * getAncillaryDataSize() will be invoked to retrieve the size of
175 * ancillary data buffer which should be passed to ::sendmsg() system call
177 * @param flags Write flags requested for the given write operation
179 virtual uint32_t getAncillaryDataSize(folly::WriteFlags /*flags*/)
184 static const size_t maxAncillaryDataSize{0x5000};
188 * getFlagsImpl() will be invoked by getFlags(folly::WriteFlags flags)
189 * method to retrieve the flags to be passed to ::sendmsg() system call.
190 * SendMsgParamsCallback::getFlags() is calling this method, and returns
191 * its results directly to the caller in AsyncSocket.
192 * Classes inheriting from SendMsgParamsCallback are welcome to override
193 * this method to force SendMsgParamsCallback to return its own set
196 * @param flags Write flags requested for the given write operation
197 * @param defaultflags A set of message flags returned by getDefaultFlags()
198 * method for the given "flags" mask.
200 virtual int getFlagsImpl(folly::WriteFlags /*flags*/, int defaultFlags) {
205 * getDefaultFlags() will be invoked by getFlags(folly::WriteFlags flags)
206 * to retrieve the default set of flags, and pass them to getFlagsImpl(...)
208 * @param flags Write flags requested for the given write operation
210 int getDefaultFlags(folly::WriteFlags flags) noexcept;
213 explicit AsyncSocket();
215 * Create a new unconnected AsyncSocket.
217 * connect() must later be called on this socket to establish a connection.
219 explicit AsyncSocket(EventBase* evb);
221 void setShutdownSocketSet(ShutdownSocketSet* ss);
224 * Create a new AsyncSocket and begin the connection process.
226 * @param evb EventBase that will manage this socket.
227 * @param address The address to connect to.
228 * @param connectTimeout Optional timeout in milliseconds for the connection
231 AsyncSocket(EventBase* evb,
232 const folly::SocketAddress& address,
233 uint32_t connectTimeout = 0);
236 * Create a new AsyncSocket and begin the connection process.
238 * @param evb EventBase that will manage this socket.
239 * @param ip IP address to connect to (dotted-quad).
240 * @param port Destination port in host byte order.
241 * @param connectTimeout Optional timeout in milliseconds for the connection
244 AsyncSocket(EventBase* evb,
245 const std::string& ip,
247 uint32_t connectTimeout = 0);
250 * Create a AsyncSocket from an already connected socket file descriptor.
252 * Note that while AsyncSocket enables TCP_NODELAY for sockets it creates
253 * when connecting, it does not change the socket options when given an
254 * existing file descriptor. If callers want TCP_NODELAY enabled when using
255 * this version of the constructor, they need to explicitly call
256 * setNoDelay(true) after the constructor returns.
258 * @param evb EventBase that will manage this socket.
259 * @param fd File descriptor to take over (should be a connected socket).
261 AsyncSocket(EventBase* evb, int fd);
264 * Create an AsyncSocket from a different, already connected AsyncSocket.
266 * Similar to AsyncSocket(evb, fd) when fd was previously owned by an
269 explicit AsyncSocket(AsyncSocket::UniquePtr);
272 * Helper function to create a shared_ptr<AsyncSocket>.
274 * This passes in the correct destructor object, since AsyncSocket's
275 * destructor is protected and cannot be invoked directly.
277 static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb) {
278 return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb),
283 * Helper function to create a shared_ptr<AsyncSocket>.
285 static std::shared_ptr<AsyncSocket> newSocket(
287 const folly::SocketAddress& address,
288 uint32_t connectTimeout = 0) {
289 return std::shared_ptr<AsyncSocket>(
290 new AsyncSocket(evb, address, connectTimeout),
295 * Helper function to create a shared_ptr<AsyncSocket>.
297 static std::shared_ptr<AsyncSocket> newSocket(
299 const std::string& ip,
301 uint32_t connectTimeout = 0) {
302 return std::shared_ptr<AsyncSocket>(
303 new AsyncSocket(evb, ip, port, connectTimeout),
308 * Helper function to create a shared_ptr<AsyncSocket>.
310 static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb, int fd) {
311 return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb, fd),
316 * Destroy the socket.
318 * AsyncSocket::destroy() must be called to destroy the socket.
319 * The normal destructor is private, and should not be invoked directly.
320 * This prevents callers from deleting a AsyncSocket while it is invoking a
323 virtual void destroy() override;
326 * Get the EventBase used by this socket.
328 EventBase* getEventBase() const override {
333 * Get the file descriptor used by the AsyncSocket.
335 virtual int getFd() const {
340 * Extract the file descriptor from the AsyncSocket.
342 * This will immediately cause any installed callbacks to be invoked with an
343 * error. The AsyncSocket may no longer be used after the file descriptor
344 * has been extracted.
346 * This method should be used with care as the resulting fd is not guaranteed
347 * to perfectly reflect the state of the AsyncSocket (security state,
348 * pre-received data, etc.).
350 * Returns the file descriptor. The caller assumes ownership of the
351 * descriptor, and it will not be closed when the AsyncSocket is destroyed.
353 virtual int detachFd();
356 * Uniquely identifies a handle to a socket option value. Each
357 * combination of level and option name corresponds to one socket
362 bool operator<(const OptionKey& other) const {
363 if (level == other.level) {
364 return optname < other.optname;
366 return level < other.level;
368 int apply(int fd, int val) const {
369 return setsockopt(fd, level, optname, &val, sizeof(val));
375 // Maps from a socket option key to its value
376 typedef std::map<OptionKey, int> OptionMap;
378 static const OptionMap emptyOptionMap;
379 static const folly::SocketAddress& anyAddress();
382 * Initiate a connection.
384 * @param callback The callback to inform when the connection attempt
386 * @param address The address to connect to.
387 * @param timeout A timeout value, in milliseconds. If the connection
388 * does not succeed within this period,
389 * callback->connectError() will be invoked.
391 virtual void connect(
392 ConnectCallback* callback,
393 const folly::SocketAddress& address,
395 const OptionMap& options = emptyOptionMap,
396 const folly::SocketAddress& bindAddr = anyAddress()) noexcept;
399 ConnectCallback* callback,
400 const std::string& ip,
403 const OptionMap& options = emptyOptionMap) noexcept;
406 * If a connect request is in-flight, cancels it and closes the socket
407 * immediately. Otherwise, this is a no-op.
409 * This does not invoke any connection related callbacks. Call this to
410 * prevent any connect callback while cleaning up, etc.
412 void cancelConnect();
415 * Set the send timeout.
417 * If write requests do not make any progress for more than the specified
418 * number of milliseconds, fail all pending writes and close the socket.
420 * If write requests are currently pending when setSendTimeout() is called,
421 * the timeout interval is immediately restarted using the new value.
423 * (See the comments for AsyncSocket for an explanation of why AsyncSocket
424 * provides setSendTimeout() but not setRecvTimeout().)
426 * @param milliseconds The timeout duration, in milliseconds. If 0, no
427 * timeout will be used.
429 void setSendTimeout(uint32_t milliseconds) override;
432 * Get the send timeout.
434 * @return Returns the current send timeout, in milliseconds. A return value
435 * of 0 indicates that no timeout is set.
437 uint32_t getSendTimeout() const override {
442 * Set the maximum number of reads to execute from the underlying
443 * socket each time the EventBase detects that new ingress data is
444 * available. The default is unlimited, but callers can use this method
445 * to limit the amount of data read from the socket per event loop
448 * @param maxReads Maximum number of reads per data-available event;
449 * a value of zero means unlimited.
451 void setMaxReadsPerEvent(uint16_t maxReads) {
452 maxReadsPerEvent_ = maxReads;
456 * Get the maximum number of reads this object will execute from
457 * the underlying socket each time the EventBase detects that new
458 * ingress data is available.
460 * @returns Maximum number of reads per data-available event; a value
461 * of zero means unlimited.
463 uint16_t getMaxReadsPerEvent() const {
464 return maxReadsPerEvent_;
468 * Set a pointer to ErrMessageCallback implementation which will be
469 * receiving notifications for messages posted to the error queue
470 * associated with the socket.
471 * ErrMessageCallback is implemented only for platforms with
472 * per-socket error message queus support (recvmsg() system call must
476 void setErrMessageCB(ErrMessageCallback* callback);
479 * Get a pointer to ErrMessageCallback implementation currently
480 * registered with this socket.
483 ErrMessageCallback* getErrMessageCallback() const;
486 * Set a pointer to SendMsgParamsCallback implementation which
487 * will be used to form ::sendmsg() system call parameters
490 void setSendMsgParamCB(SendMsgParamsCallback* callback);
493 * Get a pointer to SendMsgParamsCallback implementation currently
494 * registered with this socket.
497 SendMsgParamsCallback* getSendMsgParamsCB() const;
499 // Read and write methods
500 void setReadCB(ReadCallback* callback) override;
501 ReadCallback* getReadCallback() const override;
503 void write(WriteCallback* callback, const void* buf, size_t bytes,
504 WriteFlags flags = WriteFlags::NONE) override;
505 void writev(WriteCallback* callback, const iovec* vec, size_t count,
506 WriteFlags flags = WriteFlags::NONE) override;
507 void writeChain(WriteCallback* callback,
508 std::unique_ptr<folly::IOBuf>&& buf,
509 WriteFlags flags = WriteFlags::NONE) override;
512 virtual void writeRequest(WriteRequest* req);
513 void writeRequestReady() {
517 // Methods inherited from AsyncTransport
518 void close() override;
519 void closeNow() override;
520 void closeWithReset() override;
521 void shutdownWrite() override;
522 void shutdownWriteNow() override;
524 bool readable() const override;
525 bool isPending() const override;
526 virtual bool hangup() const;
527 bool good() const override;
528 bool error() const override;
529 void attachEventBase(EventBase* eventBase) override;
530 void detachEventBase() override;
531 bool isDetachable() const override;
533 void getLocalAddress(
534 folly::SocketAddress* address) const override;
536 folly::SocketAddress* address) const override;
538 bool isEorTrackingEnabled() const override {
542 void setEorTracking(bool track) override {
546 bool connecting() const override {
547 return (state_ == StateEnum::CONNECTING);
550 virtual bool isClosedByPeer() const {
551 return (state_ == StateEnum::CLOSED &&
552 (readErr_ == READ_EOF || readErr_ == READ_ERROR));
555 virtual bool isClosedBySelf() const {
556 return (state_ == StateEnum::CLOSED &&
557 (readErr_ != READ_EOF && readErr_ != READ_ERROR));
560 size_t getAppBytesWritten() const override {
561 return appBytesWritten_;
564 size_t getRawBytesWritten() const override {
565 return getAppBytesWritten();
568 size_t getAppBytesReceived() const override {
569 return appBytesReceived_;
572 size_t getRawBytesReceived() const override {
573 return getAppBytesReceived();
576 std::chrono::nanoseconds getConnectTime() const {
577 return connectEndTime_ - connectStartTime_;
580 std::chrono::milliseconds getConnectTimeout() const {
581 return connectTimeout_;
584 bool getTFOAttempted() const {
585 return tfoAttempted_;
589 * Returns whether or not the attempt to use TFO
590 * finished successfully. This does not necessarily
591 * mean TFO worked, just that trying to use TFO
594 bool getTFOFinished() const {
599 * Returns whether or not TFO attempt succeded on this
601 * For servers this is pretty straightforward API and can
602 * be invoked right after the connection is accepted. This API
603 * will perform one syscall.
604 * This API is a bit tricky to use for clients, since clients
605 * only know this for sure after the SYN-ACK is returned. So it's
606 * appropriate to call this only after the first application
607 * data is read from the socket when the caller knows that
608 * the SYN has been ACKed by the server.
610 bool getTFOSucceded() const;
612 // Methods controlling socket options
615 * Force writes to be transmitted immediately.
617 * This controls the TCP_NODELAY socket option. When enabled, TCP segments
618 * are sent as soon as possible, even if it is not a full frame of data.
619 * When disabled, the data may be buffered briefly to try and wait for a full
622 * By default, TCP_NODELAY is enabled for AsyncSocket objects.
624 * This method will fail if the socket is not currently open.
626 * @return Returns 0 if the TCP_NODELAY flag was successfully updated,
627 * or a non-zero errno value on error.
629 int setNoDelay(bool noDelay);
633 * Set the FD_CLOEXEC flag so that the socket will be closed if the program
634 * later forks and execs.
636 void setCloseOnExec();
639 * Set the Flavor of Congestion Control to be used for this Socket
640 * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
641 * first to make sure the module is available for plugging in
642 * Alternatively you can choose from net.ipv4.tcp_allowed_congestion_control
644 int setCongestionFlavor(const std::string &cname);
647 * Forces ACKs to be sent immediately
649 * @return Returns 0 if the TCP_QUICKACK flag was successfully updated,
650 * or a non-zero errno value on error.
652 int setQuickAck(bool quickack);
655 * Set the send bufsize
657 int setSendBufSize(size_t bufsize);
660 * Set the recv bufsize
662 int setRecvBufSize(size_t bufsize);
665 * Sets a specific tcp personality
666 * Available only on kernels 3.2 and greater
668 #define SO_SET_NAMESPACE 41
669 int setTCPProfile(int profd);
672 * Generic API for reading a socket option.
674 * @param level same as the "level" parameter in getsockopt().
675 * @param optname same as the "optname" parameter in getsockopt().
676 * @param optval pointer to the variable in which the option value should
678 * @param optlen value-result argument, initially containing the size of
679 * the buffer pointed to by optval, and modified on return
680 * to indicate the actual size of the value returned.
681 * @return same as the return value of getsockopt().
683 template <typename T>
684 int getSockOpt(int level, int optname, T* optval, socklen_t* optlen) {
685 return getsockopt(fd_, level, optname, (void*) optval, optlen);
689 * Generic API for setting a socket option.
691 * @param level same as the "level" parameter in getsockopt().
692 * @param optname same as the "optname" parameter in getsockopt().
693 * @param optval the option value to set.
694 * @return same as the return value of setsockopt().
696 template <typename T>
697 int setSockOpt(int level, int optname, const T *optval) {
698 return setsockopt(fd_, level, optname, optval, sizeof(T));
702 * Set pre-received data, to be returned to read callback before any data
705 virtual void setPreReceivedData(std::unique_ptr<IOBuf> data) {
706 if (preReceivedData_) {
707 preReceivedData_->prependChain(std::move(data));
709 preReceivedData_ = std::move(data);
714 * Enables TFO behavior on the AsyncSocket if FOLLY_ALLOW_TFO
718 // No-op if folly does not allow tfo
724 void disableTransparentTls() {
725 noTransparentTls_ = true;
728 enum class StateEnum : uint8_t {
737 void setBufferCallback(BufferCallback* cb);
739 // Callers should set this prior to connecting the socket for the safest
741 void setEvbChangedCallback(std::unique_ptr<EvbChangeCallback> cb) {
742 evbChangeCb_ = std::move(cb);
746 * writeReturn is the total number of bytes written, or WRITE_ERROR on error.
747 * If no data has been written, 0 is returned.
748 * exception is a more specific exception that cause a write error.
749 * Not all writes have exceptions associated with them thus writeReturn
750 * should be checked to determine whether the operation resulted in an error.
753 explicit WriteResult(ssize_t ret) : writeReturn(ret) {}
755 WriteResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
756 : writeReturn(ret), exception(std::move(e)) {}
759 std::unique_ptr<const AsyncSocketException> exception;
763 * readReturn is the number of bytes read, or READ_EOF on EOF, or
764 * READ_ERROR on error, or READ_BLOCKING if the operation will
766 * exception is a more specific exception that may have caused a read error.
767 * Not all read errors have exceptions associated with them thus readReturn
768 * should be checked to determine whether the operation resulted in an error.
771 explicit ReadResult(ssize_t ret) : readReturn(ret) {}
773 ReadResult(ssize_t ret, std::unique_ptr<const AsyncSocketException> e)
774 : readReturn(ret), exception(std::move(e)) {}
777 std::unique_ptr<const AsyncSocketException> exception;
781 * A WriteRequest object tracks information about a pending write operation.
785 WriteRequest(AsyncSocket* socket, WriteCallback* callback) :
786 socket_(socket), callback_(callback) {}
788 virtual void start() {}
790 virtual void destroy() = 0;
792 virtual WriteResult performWrite() = 0;
794 virtual void consume() = 0;
796 virtual bool isComplete() = 0;
798 WriteRequest* getNext() const {
802 WriteCallback* getCallback() const {
806 uint32_t getTotalBytesWritten() const {
807 return totalBytesWritten_;
810 void append(WriteRequest* next) {
811 assert(next_ == nullptr);
815 void fail(const char* fn, const AsyncSocketException& ex) {
816 socket_->failWrite(fn, ex);
819 void bytesWritten(size_t count) {
820 totalBytesWritten_ += uint32_t(count);
821 socket_->appBytesWritten_ += count;
825 // protected destructor, to ensure callers use destroy()
826 virtual ~WriteRequest() {}
828 AsyncSocket* socket_; ///< parent socket
829 WriteRequest* next_{nullptr}; ///< pointer to next WriteRequest
830 WriteCallback* callback_; ///< completion callback
831 uint32_t totalBytesWritten_{0}; ///< total bytes written
835 enum ReadResultEnum {
842 enum WriteResultEnum {
847 * Protected destructor.
849 * Users of AsyncSocket must never delete it directly. Instead, invoke
850 * destroy() instead. (See the documentation in DelayedDestruction.h for
855 friend std::ostream& operator << (std::ostream& os, const StateEnum& state);
858 /// shutdownWrite() called, but we are still waiting on writes to drain
859 SHUT_WRITE_PENDING = 0x01,
860 /// writes have been completely shut down
863 * Reads have been shutdown.
865 * At the moment we don't distinguish between remote read shutdown
866 * (received EOF from the remote end) and local read shutdown. We can
867 * only receive EOF when a read callback is set, and we immediately inform
868 * it of the EOF. Therefore there doesn't seem to be any reason to have a
869 * separate state of "received EOF but the local side may still want to
872 * We also don't currently provide any API for only shutting down the read
873 * side of a socket. (This is a no-op as far as TCP is concerned, anyway.)
878 class BytesWriteRequest;
880 class WriteTimeout : public AsyncTimeout {
882 WriteTimeout(AsyncSocket* socket, EventBase* eventBase)
883 : AsyncTimeout(eventBase)
886 virtual void timeoutExpired() noexcept {
887 socket_->timeoutExpired();
891 AsyncSocket* socket_;
894 class IoHandler : public EventHandler {
896 IoHandler(AsyncSocket* socket, EventBase* eventBase)
897 : EventHandler(eventBase, -1)
899 IoHandler(AsyncSocket* socket, EventBase* eventBase, int fd)
900 : EventHandler(eventBase, fd)
903 virtual void handlerReady(uint16_t events) noexcept {
904 socket_->ioReady(events);
908 AsyncSocket* socket_;
913 class ImmediateReadCB : public folly::EventBase::LoopCallback {
915 explicit ImmediateReadCB(AsyncSocket* socket) : socket_(socket) {}
916 void runLoopCallback() noexcept override {
917 DestructorGuard dg(socket_);
918 socket_->checkForImmediateRead();
921 AsyncSocket* socket_;
925 * Schedule checkForImmediateRead to be executed in the next loop
928 void scheduleImmediateRead() noexcept {
930 eventBase_->runInLoop(&immediateReadHandler_);
935 * Schedule handleInitalReadWrite to run in the next iteration.
937 void scheduleInitialReadWrite() noexcept {
939 DestructorGuard dg(this);
940 eventBase_->runInLoop([this, dg] {
942 handleInitialReadWrite();
948 // event notification methods
949 void ioReady(uint16_t events) noexcept;
950 virtual void checkForImmediateRead() noexcept;
951 virtual void handleInitialReadWrite() noexcept;
952 virtual void prepareReadBuffer(void** buf, size_t* buflen);
953 virtual void handleErrMessages() noexcept;
954 virtual void handleRead() noexcept;
955 virtual void handleWrite() noexcept;
956 virtual void handleConnect() noexcept;
957 void timeoutExpired() noexcept;
960 * Attempt to read from the socket.
962 * @param buf The buffer to read data into.
963 * @param buflen The length of the buffer.
965 * @return Returns a read result. See read result for details.
967 virtual ReadResult performRead(void** buf, size_t* buflen, size_t* offset);
970 * Populate an iovec array from an IOBuf and attempt to write it.
972 * @param callback Write completion/error callback.
973 * @param vec Target iovec array; caller retains ownership.
974 * @param count Number of IOBufs to write, beginning at start of buf.
975 * @param buf Chain of iovecs.
976 * @param flags set of flags for the underlying write calls, like cork
978 void writeChainImpl(WriteCallback* callback, iovec* vec,
979 size_t count, std::unique_ptr<folly::IOBuf>&& buf,
983 * Write as much data as possible to the socket without blocking,
984 * and queue up any leftover data to send when the socket can
985 * handle writes again.
987 * @param callback The callback to invoke when the write is completed.
988 * @param vec Array of buffers to write; this method will make a
989 * copy of the vector (but not the buffers themselves)
990 * if the write has to be completed asynchronously.
991 * @param count Number of elements in vec.
992 * @param buf The IOBuf that manages the buffers referenced by
993 * vec, or a pointer to nullptr if the buffers are not
994 * associated with an IOBuf. Note that ownership of
995 * the IOBuf is transferred here; upon completion of
996 * the write, the AsyncSocket deletes the IOBuf.
997 * @param flags Set of write flags.
999 void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
1000 std::unique_ptr<folly::IOBuf>&& buf,
1001 WriteFlags flags = WriteFlags::NONE);
1004 * Attempt to write to the socket.
1006 * @param vec The iovec array pointing to the buffers to write.
1007 * @param count The length of the iovec array.
1008 * @param flags Set of write flags.
1009 * @param countWritten On return, the value pointed to by this parameter
1010 * will contain the number of iovec entries that were
1012 * @param partialWritten On return, the value pointed to by this parameter
1013 * will contain the number of bytes written in the
1014 * partially written iovec entry.
1016 * @return Returns a WriteResult. See WriteResult for more details.
1018 virtual WriteResult performWrite(
1022 uint32_t* countWritten,
1023 uint32_t* partialWritten);
1026 * Sends the message over the socket using sendmsg
1028 * @param msg Message to send
1029 * @param msg_flags Flags to pass to sendmsg
1031 AsyncSocket::WriteResult
1032 sendSocketMessage(int fd, struct msghdr* msg, int msg_flags);
1034 virtual ssize_t tfoSendMsg(int fd, struct msghdr* msg, int msg_flags);
1036 int socketConnect(const struct sockaddr* addr, socklen_t len);
1038 virtual void scheduleConnectTimeout();
1039 void registerForConnectEvents();
1041 bool updateEventRegistration();
1044 * Update event registration.
1046 * @param enable Flags of events to enable. Set it to 0 if no events
1047 * need to be enabled in this call.
1048 * @param disable Flags of events
1049 * to disable. Set it to 0 if no events need to be disabled in this
1052 * @return true iff the update is successful.
1054 bool updateEventRegistration(uint16_t enable, uint16_t disable);
1056 // Actually close the file descriptor and set it to -1 so we don't
1057 // accidentally close it again.
1060 // error handling methods
1063 void finishFail(const AsyncSocketException& ex);
1064 void invokeAllErrors(const AsyncSocketException& ex);
1065 void fail(const char* fn, const AsyncSocketException& ex);
1066 void failConnect(const char* fn, const AsyncSocketException& ex);
1067 void failRead(const char* fn, const AsyncSocketException& ex);
1068 void failErrMessageRead(const char* fn, const AsyncSocketException& ex);
1069 void failWrite(const char* fn, WriteCallback* callback, size_t bytesWritten,
1070 const AsyncSocketException& ex);
1071 void failWrite(const char* fn, const AsyncSocketException& ex);
1072 void failAllWrites(const AsyncSocketException& ex);
1073 virtual void invokeConnectErr(const AsyncSocketException& ex);
1074 virtual void invokeConnectSuccess();
1075 void invalidState(ConnectCallback* callback);
1076 void invalidState(ErrMessageCallback* callback);
1077 void invalidState(ReadCallback* callback);
1078 void invalidState(WriteCallback* callback);
1080 std::string withAddr(const std::string& s);
1082 StateEnum state_; ///< StateEnum describing current state
1083 uint8_t shutdownFlags_; ///< Shutdown state (ShutdownFlags)
1084 uint16_t eventFlags_; ///< EventBase::HandlerFlags settings
1085 int fd_; ///< The socket file descriptor
1086 mutable folly::SocketAddress addr_; ///< The address we tried to connect to
1087 mutable folly::SocketAddress localAddr_;
1088 ///< The address we are connecting from
1089 uint32_t sendTimeout_; ///< The send timeout, in milliseconds
1090 uint16_t maxReadsPerEvent_; ///< Max reads per event loop iteration
1091 EventBase* eventBase_; ///< The EventBase
1092 WriteTimeout writeTimeout_; ///< A timeout for connect and write
1093 IoHandler ioHandler_; ///< A EventHandler to monitor the fd
1094 ImmediateReadCB immediateReadHandler_; ///< LoopCallback for checking read
1096 ConnectCallback* connectCallback_; ///< ConnectCallback
1097 ErrMessageCallback* errMessageCallback_; ///< TimestampCallback
1098 SendMsgParamsCallback* ///< Callback for retreaving
1099 sendMsgParamCallback_; ///< ::sendmsg() parameters
1100 ReadCallback* readCallback_; ///< ReadCallback
1101 WriteRequest* writeReqHead_; ///< Chain of WriteRequests
1102 WriteRequest* writeReqTail_; ///< End of WriteRequest chain
1103 ShutdownSocketSet* shutdownSocketSet_;
1104 size_t appBytesReceived_; ///< Num of bytes received from socket
1105 size_t appBytesWritten_; ///< Num of bytes written to socket
1106 bool isBufferMovable_{false};
1108 // Pre-received data, to be returned to read callback before any data from the
1110 std::unique_ptr<IOBuf> preReceivedData_;
1112 int8_t readErr_{READ_NO_ERROR}; ///< The read error encountered, if any
1114 std::chrono::steady_clock::time_point connectStartTime_;
1115 std::chrono::steady_clock::time_point connectEndTime_;
1117 std::chrono::milliseconds connectTimeout_{0};
1119 BufferCallback* bufferCallback_{nullptr};
1120 bool tfoEnabled_{false};
1121 bool tfoAttempted_{false};
1122 bool tfoFinished_{false};
1123 bool noTransparentTls_{false};
1124 // Whether to track EOR or not.
1125 bool trackEor_{false};
1127 std::unique_ptr<EvbChangeCallback> evbChangeCb_{nullptr};
1130 #pragma vtordisp(pop)