+++ /dev/null
-/*
- * Copyright 2014 Facebook, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#include <folly/io/async/AsyncSocket.h>
-
-#include <folly/io/async/EventBase.h>
-#include <folly/SocketAddress.h>
-#include <folly/io/IOBuf.h>
-
-#include <poll.h>
-#include <errno.h>
-#include <limits.h>
-#include <unistd.h>
-#include <fcntl.h>
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <netinet/in.h>
-#include <netinet/tcp.h>
-
-using std::string;
-using std::unique_ptr;
-
-namespace folly {
-
-// static members initializers
-const AsyncSocket::OptionMap AsyncSocket::emptyOptionMap;
-const folly::SocketAddress AsyncSocket::anyAddress =
- folly::SocketAddress("0.0.0.0", 0);
-
-const AsyncSocketException socketClosedLocallyEx(
- AsyncSocketException::END_OF_FILE, "socket closed locally");
-const AsyncSocketException socketShutdownForWritesEx(
- AsyncSocketException::END_OF_FILE, "socket shutdown for writes");
-
-// TODO: It might help performance to provide a version of WriteRequest that
-// users could derive from, so we can avoid the extra allocation for each call
-// to write()/writev(). We could templatize TFramedAsyncChannel just like the
-// protocols are currently templatized for transports.
-//
-// We would need the version for external users where they provide the iovec
-// storage space, and only our internal version would allocate it at the end of
-// the WriteRequest.
-
-/**
- * A WriteRequest object tracks information about a pending write() or writev()
- * operation.
- *
- * A new WriteRequest operation is allocated on the heap for all write
- * operations that cannot be completed immediately.
- */
-class AsyncSocket::WriteRequest {
- public:
- static WriteRequest* newRequest(WriteCallback* callback,
- const iovec* ops,
- uint32_t opCount,
- unique_ptr<IOBuf>&& ioBuf,
- WriteFlags flags) {
- assert(opCount > 0);
- // Since we put a variable size iovec array at the end
- // of each WriteRequest, we have to manually allocate the memory.
- void* buf = malloc(sizeof(WriteRequest) +
- (opCount * sizeof(struct iovec)));
- if (buf == nullptr) {
- throw std::bad_alloc();
- }
-
- return new(buf) WriteRequest(callback, ops, opCount, std::move(ioBuf),
- flags);
- }
-
- void destroy() {
- this->~WriteRequest();
- free(this);
- }
-
- bool cork() const {
- return isSet(flags_, WriteFlags::CORK);
- }
-
- WriteFlags flags() const {
- return flags_;
- }
-
- WriteRequest* getNext() const {
- return next_;
- }
-
- WriteCallback* getCallback() const {
- return callback_;
- }
-
- uint32_t getBytesWritten() const {
- return bytesWritten_;
- }
-
- const struct iovec* getOps() const {
- assert(opCount_ > opIndex_);
- return writeOps_ + opIndex_;
- }
-
- uint32_t getOpCount() const {
- assert(opCount_ > opIndex_);
- return opCount_ - opIndex_;
- }
-
- void consume(uint32_t wholeOps, uint32_t partialBytes,
- uint32_t totalBytesWritten) {
- // Advance opIndex_ forward by wholeOps
- opIndex_ += wholeOps;
- assert(opIndex_ < opCount_);
-
- // If we've finished writing any IOBufs, release them
- if (ioBuf_) {
- for (uint32_t i = wholeOps; i != 0; --i) {
- assert(ioBuf_);
- ioBuf_ = ioBuf_->pop();
- }
- }
-
- // Move partialBytes forward into the current iovec buffer
- struct iovec* currentOp = writeOps_ + opIndex_;
- assert((partialBytes < currentOp->iov_len) || (currentOp->iov_len == 0));
- currentOp->iov_base =
- reinterpret_cast<uint8_t*>(currentOp->iov_base) + partialBytes;
- currentOp->iov_len -= partialBytes;
-
- // Increment the bytesWritten_ count by totalBytesWritten
- bytesWritten_ += totalBytesWritten;
- }
-
- void append(WriteRequest* next) {
- assert(next_ == nullptr);
- next_ = next;
- }
-
- private:
- WriteRequest(WriteCallback* callback,
- const struct iovec* ops,
- uint32_t opCount,
- unique_ptr<IOBuf>&& ioBuf,
- WriteFlags flags)
- : next_(nullptr)
- , callback_(callback)
- , bytesWritten_(0)
- , opCount_(opCount)
- , opIndex_(0)
- , flags_(flags)
- , ioBuf_(std::move(ioBuf)) {
- memcpy(writeOps_, ops, sizeof(*ops) * opCount_);
- }
-
- // Private destructor, to ensure callers use destroy()
- ~WriteRequest() {}
-
- WriteRequest* next_; ///< pointer to next WriteRequest
- WriteCallback* callback_; ///< completion callback
- uint32_t bytesWritten_; ///< bytes written
- uint32_t opCount_; ///< number of entries in writeOps_
- uint32_t opIndex_; ///< current index into writeOps_
- WriteFlags flags_; ///< set for WriteFlags
- unique_ptr<IOBuf> ioBuf_; ///< underlying IOBuf, or nullptr if N/A
- struct iovec writeOps_[]; ///< write operation(s) list
-};
-
-AsyncSocket::AsyncSocket(EventBase* evb)
- : eventBase_(evb)
- , writeTimeout_(this, evb)
- , ioHandler_(this, evb) {
- VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
- init();
-}
-
-AsyncSocket::AsyncSocket(EventBase* evb,
- const folly::SocketAddress& address,
- uint32_t connectTimeout)
- : eventBase_(evb)
- , writeTimeout_(this, evb)
- , ioHandler_(this, evb) {
- VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
- init();
- connect(nullptr, address, connectTimeout);
-}
-
-AsyncSocket::AsyncSocket(EventBase* evb,
- const std::string& ip,
- uint16_t port,
- uint32_t connectTimeout)
- : eventBase_(evb)
- , writeTimeout_(this, evb)
- , ioHandler_(this, evb) {
- VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ")";
- init();
- connect(nullptr, ip, port, connectTimeout);
-}
-
-AsyncSocket::AsyncSocket(EventBase* evb, int fd)
- : eventBase_(evb)
- , writeTimeout_(this, evb)
- , ioHandler_(this, evb, fd) {
- VLOG(5) << "new AsyncSocket(" << this << ", evb=" << evb << ", fd="
- << fd << ")";
- init();
- fd_ = fd;
- state_ = StateEnum::ESTABLISHED;
-}
-
-// init() method, since constructor forwarding isn't supported in most
-// compilers yet.
-void AsyncSocket::init() {
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
- shutdownFlags_ = 0;
- state_ = StateEnum::UNINIT;
- eventFlags_ = EventHandler::NONE;
- fd_ = -1;
- sendTimeout_ = 0;
- maxReadsPerEvent_ = 0;
- connectCallback_ = nullptr;
- readCallback_ = nullptr;
- writeReqHead_ = nullptr;
- writeReqTail_ = nullptr;
- shutdownSocketSet_ = nullptr;
- appBytesWritten_ = 0;
- appBytesReceived_ = 0;
-}
-
-AsyncSocket::~AsyncSocket() {
- VLOG(7) << "actual destruction of AsyncSocket(this=" << this
- << ", evb=" << eventBase_ << ", fd=" << fd_
- << ", state=" << state_ << ")";
-}
-
-void AsyncSocket::destroy() {
- VLOG(5) << "AsyncSocket::destroy(this=" << this << ", evb=" << eventBase_
- << ", fd=" << fd_ << ", state=" << state_;
- // When destroy is called, close the socket immediately
- closeNow();
-
- // Then call DelayedDestruction::destroy() to take care of
- // whether or not we need immediate or delayed destruction
- DelayedDestruction::destroy();
-}
-
-int AsyncSocket::detachFd() {
- VLOG(6) << "AsyncSocket::detachFd(this=" << this << ", fd=" << fd_
- << ", evb=" << eventBase_ << ", state=" << state_
- << ", events=" << std::hex << eventFlags_ << ")";
- // Extract the fd, and set fd_ to -1 first, so closeNow() won't
- // actually close the descriptor.
- if (shutdownSocketSet_) {
- shutdownSocketSet_->remove(fd_);
- }
- int fd = fd_;
- fd_ = -1;
- // Call closeNow() to invoke all pending callbacks with an error.
- closeNow();
- // Update the EventHandler to stop using this fd.
- // This can only be done after closeNow() unregisters the handler.
- ioHandler_.changeHandlerFD(-1);
- return fd;
-}
-
-void AsyncSocket::setShutdownSocketSet(ShutdownSocketSet* newSS) {
- if (shutdownSocketSet_ == newSS) {
- return;
- }
- if (shutdownSocketSet_ && fd_ != -1) {
- shutdownSocketSet_->remove(fd_);
- }
- shutdownSocketSet_ = newSS;
- if (shutdownSocketSet_ && fd_ != -1) {
- shutdownSocketSet_->add(fd_);
- }
-}
-
-void AsyncSocket::connect(ConnectCallback* callback,
- const folly::SocketAddress& address,
- int timeout,
- const OptionMap &options,
- const folly::SocketAddress& bindAddr) noexcept {
- DestructorGuard dg(this);
- assert(eventBase_->isInEventBaseThread());
-
- addr_ = address;
-
- // Make sure we're in the uninitialized state
- if (state_ != StateEnum::UNINIT) {
- return invalidState(callback);
- }
-
- assert(fd_ == -1);
- state_ = StateEnum::CONNECTING;
- connectCallback_ = callback;
-
- sockaddr_storage addrStorage;
- sockaddr* saddr = reinterpret_cast<sockaddr*>(&addrStorage);
-
- try {
- // Create the socket
- // Technically the first parameter should actually be a protocol family
- // constant (PF_xxx) rather than an address family (AF_xxx), but the
- // distinction is mainly just historical. In pretty much all
- // implementations the PF_foo and AF_foo constants are identical.
- fd_ = socket(address.getFamily(), SOCK_STREAM, 0);
- if (fd_ < 0) {
- throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to create socket"), errno);
- }
- if (shutdownSocketSet_) {
- shutdownSocketSet_->add(fd_);
- }
- ioHandler_.changeHandlerFD(fd_);
-
- // Set the FD_CLOEXEC flag so that the socket will be closed if the program
- // later forks and execs.
- int rv = fcntl(fd_, F_SETFD, FD_CLOEXEC);
- if (rv != 0) {
- throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to set close-on-exec flag"),
- errno);
- }
-
- // Put the socket in non-blocking mode
- int flags = fcntl(fd_, F_GETFL, 0);
- if (flags == -1) {
- throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to get socket flags"), errno);
- }
- rv = fcntl(fd_, F_SETFL, flags | O_NONBLOCK);
- if (rv == -1) {
- throw AsyncSocketException(
- AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to put socket in non-blocking mode"),
- errno);
- }
-
-#if !defined(MSG_NOSIGNAL) && defined(F_SETNOSIGPIPE)
- // iOS and OS X don't support MSG_NOSIGNAL; set F_SETNOSIGPIPE instead
- rv = fcntl(fd_, F_SETNOSIGPIPE, 1);
- if (rv == -1) {
- throw AsyncSocketException(
- AsyncSocketException::INTERNAL_ERROR,
- "failed to enable F_SETNOSIGPIPE on socket",
- errno);
- }
-#endif
-
- // By default, turn on TCP_NODELAY
- // If setNoDelay() fails, we continue anyway; this isn't a fatal error.
- // setNoDelay() will log an error message if it fails.
- if (address.getFamily() != AF_UNIX) {
- (void)setNoDelay(true);
- }
-
- VLOG(5) << "AsyncSocket::connect(this=" << this << ", evb=" << eventBase_
- << ", fd=" << fd_ << ", host=" << address.describe().c_str();
-
- // bind the socket
- if (bindAddr != anyAddress) {
- int one = 1;
- if (::setsockopt(fd_, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one))) {
- doClose();
- throw AsyncSocketException(
- AsyncSocketException::NOT_OPEN,
- "failed to setsockopt prior to bind on " + bindAddr.describe(),
- errno);
- }
-
- bindAddr.getAddress(&addrStorage);
-
- if (::bind(fd_, saddr, bindAddr.getActualSize()) != 0) {
- doClose();
- throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
- "failed to bind to async socket: " +
- bindAddr.describe(),
- errno);
- }
- }
-
- // Apply the additional options if any.
- for (const auto& opt: options) {
- int rv = opt.first.apply(fd_, opt.second);
- if (rv != 0) {
- throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to set socket option"),
- errno);
- }
- }
-
- // Perform the connect()
- address.getAddress(&addrStorage);
-
- rv = ::connect(fd_, saddr, address.getActualSize());
- if (rv < 0) {
- if (errno == EINPROGRESS) {
- // Connection in progress.
- if (timeout > 0) {
- // Start a timer in case the connection takes too long.
- if (!writeTimeout_.scheduleTimeout(timeout)) {
- throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to schedule AsyncSocket connect timeout"));
- }
- }
-
- // Register for write events, so we'll
- // be notified when the connection finishes/fails.
- // Note that we don't register for a persistent event here.
- assert(eventFlags_ == EventHandler::NONE);
- eventFlags_ = EventHandler::WRITE;
- if (!ioHandler_.registerHandler(eventFlags_)) {
- throw AsyncSocketException(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to register AsyncSocket connect handler"));
- }
- return;
- } else {
- throw AsyncSocketException(AsyncSocketException::NOT_OPEN,
- "connect failed (immediately)", errno);
- }
- }
-
- // If we're still here the connect() succeeded immediately.
- // Fall through to call the callback outside of this try...catch block
- } catch (const AsyncSocketException& ex) {
- return failConnect(__func__, ex);
- } catch (const std::exception& ex) {
- // shouldn't happen, but handle it just in case
- VLOG(4) << "AsyncSocket::connect(this=" << this << ", fd=" << fd_
- << "): unexpected " << typeid(ex).name() << " exception: "
- << ex.what();
- AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
- withAddr(string("unexpected exception: ") +
- ex.what()));
- return failConnect(__func__, tex);
- }
-
- // The connection succeeded immediately
- // The read callback may not have been set yet, and no writes may be pending
- // yet, so we don't have to register for any events at the moment.
- VLOG(8) << "AsyncSocket::connect succeeded immediately; this=" << this;
- assert(readCallback_ == nullptr);
- assert(writeReqHead_ == nullptr);
- state_ = StateEnum::ESTABLISHED;
- if (callback) {
- connectCallback_ = nullptr;
- callback->connectSuccess();
- }
-}
-
-void AsyncSocket::connect(ConnectCallback* callback,
- const string& ip, uint16_t port,
- int timeout,
- const OptionMap &options) noexcept {
- DestructorGuard dg(this);
- try {
- connectCallback_ = callback;
- connect(callback, folly::SocketAddress(ip, port), timeout, options);
- } catch (const std::exception& ex) {
- AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
- ex.what());
- return failConnect(__func__, tex);
- }
-}
-
-void AsyncSocket::setSendTimeout(uint32_t milliseconds) {
- sendTimeout_ = milliseconds;
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
-
- // If we are currently pending on write requests, immediately update
- // writeTimeout_ with the new value.
- if ((eventFlags_ & EventHandler::WRITE) &&
- (state_ != StateEnum::CONNECTING)) {
- assert(state_ == StateEnum::ESTABLISHED);
- assert((shutdownFlags_ & SHUT_WRITE) == 0);
- if (sendTimeout_ > 0) {
- if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to reschedule send timeout in setSendTimeout"));
- return failWrite(__func__, ex);
- }
- } else {
- writeTimeout_.cancelTimeout();
- }
- }
-}
-
-void AsyncSocket::setReadCB(ReadCallback *callback) {
- VLOG(6) << "AsyncSocket::setReadCallback() this=" << this << ", fd=" << fd_
- << ", callback=" << callback << ", state=" << state_;
-
- // Short circuit if callback is the same as the existing readCallback_.
- //
- // Note that this is needed for proper functioning during some cleanup cases.
- // During cleanup we allow setReadCallback(nullptr) to be called even if the
- // read callback is already unset and we have been detached from an event
- // base. This check prevents us from asserting
- // eventBase_->isInEventBaseThread() when eventBase_ is nullptr.
- if (callback == readCallback_) {
- return;
- }
-
- if (shutdownFlags_ & SHUT_READ) {
- // Reads have already been shut down on this socket.
- //
- // Allow setReadCallback(nullptr) to be called in this case, but don't
- // allow a new callback to be set.
- //
- // For example, setReadCallback(nullptr) can happen after an error if we
- // invoke some other error callback before invoking readError(). The other
- // error callback that is invoked first may go ahead and clear the read
- // callback before we get a chance to invoke readError().
- if (callback != nullptr) {
- return invalidState(callback);
- }
- assert((eventFlags_ & EventHandler::READ) == 0);
- readCallback_ = nullptr;
- return;
- }
-
- DestructorGuard dg(this);
- assert(eventBase_->isInEventBaseThread());
-
- switch ((StateEnum)state_) {
- case StateEnum::CONNECTING:
- // For convenience, we allow the read callback to be set while we are
- // still connecting. We just store the callback for now. Once the
- // connection completes we'll register for read events.
- readCallback_ = callback;
- return;
- case StateEnum::ESTABLISHED:
- {
- readCallback_ = callback;
- uint16_t oldFlags = eventFlags_;
- if (readCallback_) {
- eventFlags_ |= EventHandler::READ;
- } else {
- eventFlags_ &= ~EventHandler::READ;
- }
-
- // Update our registration if our flags have changed
- if (eventFlags_ != oldFlags) {
- // We intentionally ignore the return value here.
- // updateEventRegistration() will move us into the error state if it
- // fails, and we don't need to do anything else here afterwards.
- (void)updateEventRegistration();
- }
-
- if (readCallback_) {
- checkForImmediateRead();
- }
- return;
- }
- case StateEnum::CLOSED:
- case StateEnum::ERROR:
- // We should never reach here. SHUT_READ should always be set
- // if we are in STATE_CLOSED or STATE_ERROR.
- assert(false);
- return invalidState(callback);
- case StateEnum::UNINIT:
- // We do not allow setReadCallback() to be called before we start
- // connecting.
- return invalidState(callback);
- }
-
- // We don't put a default case in the switch statement, so that the compiler
- // will warn us to update the switch statement if a new state is added.
- return invalidState(callback);
-}
-
-AsyncSocket::ReadCallback* AsyncSocket::getReadCallback() const {
- return readCallback_;
-}
-
-void AsyncSocket::write(WriteCallback* callback,
- const void* buf, size_t bytes, WriteFlags flags) {
- iovec op;
- op.iov_base = const_cast<void*>(buf);
- op.iov_len = bytes;
- writeImpl(callback, &op, 1, std::move(unique_ptr<IOBuf>()), flags);
-}
-
-void AsyncSocket::writev(WriteCallback* callback,
- const iovec* vec,
- size_t count,
- WriteFlags flags) {
- writeImpl(callback, vec, count, std::move(unique_ptr<IOBuf>()), flags);
-}
-
-void AsyncSocket::writeChain(WriteCallback* callback, unique_ptr<IOBuf>&& buf,
- WriteFlags flags) {
- size_t count = buf->countChainElements();
- if (count <= 64) {
- iovec vec[count];
- writeChainImpl(callback, vec, count, std::move(buf), flags);
- } else {
- iovec* vec = new iovec[count];
- writeChainImpl(callback, vec, count, std::move(buf), flags);
- delete[] vec;
- }
-}
-
-void AsyncSocket::writeChainImpl(WriteCallback* callback, iovec* vec,
- size_t count, unique_ptr<IOBuf>&& buf, WriteFlags flags) {
- const IOBuf* head = buf.get();
- const IOBuf* next = head;
- unsigned i = 0;
- do {
- vec[i].iov_base = const_cast<uint8_t *>(next->data());
- vec[i].iov_len = next->length();
- // IOBuf can get confused by empty iovec buffers, so increment the
- // output pointer only if the iovec buffer is non-empty. We could
- // end the loop with i < count, but that's ok.
- if (vec[i].iov_len != 0) {
- i++;
- }
- next = next->next();
- } while (next != head);
- writeImpl(callback, vec, i, std::move(buf), flags);
-}
-
-void AsyncSocket::writeImpl(WriteCallback* callback, const iovec* vec,
- size_t count, unique_ptr<IOBuf>&& buf,
- WriteFlags flags) {
- VLOG(6) << "AsyncSocket::writev() this=" << this << ", fd=" << fd_
- << ", callback=" << callback << ", count=" << count
- << ", state=" << state_;
- DestructorGuard dg(this);
- unique_ptr<IOBuf>ioBuf(std::move(buf));
- assert(eventBase_->isInEventBaseThread());
-
- if (shutdownFlags_ & (SHUT_WRITE | SHUT_WRITE_PENDING)) {
- // No new writes may be performed after the write side of the socket has
- // been shutdown.
- //
- // We could just call callback->writeError() here to fail just this write.
- // However, fail hard and use invalidState() to fail all outstanding
- // callbacks and move the socket into the error state. There's most likely
- // a bug in the caller's code, so we abort everything rather than trying to
- // proceed as best we can.
- return invalidState(callback);
- }
-
- uint32_t countWritten = 0;
- uint32_t partialWritten = 0;
- int bytesWritten = 0;
- bool mustRegister = false;
- if (state_ == StateEnum::ESTABLISHED && !connecting()) {
- if (writeReqHead_ == nullptr) {
- // If we are established and there are no other writes pending,
- // we can attempt to perform the write immediately.
- assert(writeReqTail_ == nullptr);
- assert((eventFlags_ & EventHandler::WRITE) == 0);
-
- bytesWritten = performWrite(vec, count, flags,
- &countWritten, &partialWritten);
- if (bytesWritten < 0) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("writev failed"), errno);
- return failWrite(__func__, callback, 0, ex);
- } else if (countWritten == count) {
- // We successfully wrote everything.
- // Invoke the callback and return.
- if (callback) {
- callback->writeSuccess();
- }
- return;
- } // else { continue writing the next writeReq }
- mustRegister = true;
- }
- } else if (!connecting()) {
- // Invalid state for writing
- return invalidState(callback);
- }
-
- // Create a new WriteRequest to add to the queue
- WriteRequest* req;
- try {
- req = WriteRequest::newRequest(callback, vec + countWritten,
- count - countWritten, std::move(ioBuf),
- flags);
- } catch (const std::exception& ex) {
- // we mainly expect to catch std::bad_alloc here
- AsyncSocketException tex(AsyncSocketException::INTERNAL_ERROR,
- withAddr(string("failed to append new WriteRequest: ") + ex.what()));
- return failWrite(__func__, callback, bytesWritten, tex);
- }
- req->consume(0, partialWritten, bytesWritten);
- if (writeReqTail_ == nullptr) {
- assert(writeReqHead_ == nullptr);
- writeReqHead_ = writeReqTail_ = req;
- } else {
- writeReqTail_->append(req);
- writeReqTail_ = req;
- }
-
- // Register for write events if are established and not currently
- // waiting on write events
- if (mustRegister) {
- assert(state_ == StateEnum::ESTABLISHED);
- assert((eventFlags_ & EventHandler::WRITE) == 0);
- if (!updateEventRegistration(EventHandler::WRITE, 0)) {
- assert(state_ == StateEnum::ERROR);
- return;
- }
- if (sendTimeout_ > 0) {
- // Schedule a timeout to fire if the write takes too long.
- if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to schedule send timeout"));
- return failWrite(__func__, ex);
- }
- }
- }
-}
-
-void AsyncSocket::close() {
- VLOG(5) << "AsyncSocket::close(): this=" << this << ", fd_=" << fd_
- << ", state=" << state_ << ", shutdownFlags="
- << std::hex << (int) shutdownFlags_;
-
- // close() is only different from closeNow() when there are pending writes
- // that need to drain before we can close. In all other cases, just call
- // closeNow().
- //
- // Note that writeReqHead_ can be non-nullptr even in STATE_CLOSED or
- // STATE_ERROR if close() is invoked while a previous closeNow() or failure
- // is still running. (e.g., If there are multiple pending writes, and we
- // call writeError() on the first one, it may call close(). In this case we
- // will already be in STATE_CLOSED or STATE_ERROR, but the remaining pending
- // writes will still be in the queue.)
- //
- // We only need to drain pending writes if we are still in STATE_CONNECTING
- // or STATE_ESTABLISHED
- if ((writeReqHead_ == nullptr) ||
- !(state_ == StateEnum::CONNECTING ||
- state_ == StateEnum::ESTABLISHED)) {
- closeNow();
- return;
- }
-
- // Declare a DestructorGuard to ensure that the AsyncSocket cannot be
- // destroyed until close() returns.
- DestructorGuard dg(this);
- assert(eventBase_->isInEventBaseThread());
-
- // Since there are write requests pending, we have to set the
- // SHUT_WRITE_PENDING flag, and wait to perform the real close until the
- // connect finishes and we finish writing these requests.
- //
- // Set SHUT_READ to indicate that reads are shut down, and set the
- // SHUT_WRITE_PENDING flag to mark that we want to shutdown once the
- // pending writes complete.
- shutdownFlags_ |= (SHUT_READ | SHUT_WRITE_PENDING);
-
- // If a read callback is set, invoke readEOF() immediately to inform it that
- // the socket has been closed and no more data can be read.
- if (readCallback_) {
- // Disable reads if they are enabled
- if (!updateEventRegistration(0, EventHandler::READ)) {
- // We're now in the error state; callbacks have been cleaned up
- assert(state_ == StateEnum::ERROR);
- assert(readCallback_ == nullptr);
- } else {
- ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readEOF();
- }
- }
-}
-
-void AsyncSocket::closeNow() {
- VLOG(5) << "AsyncSocket::closeNow(): this=" << this << ", fd_=" << fd_
- << ", state=" << state_ << ", shutdownFlags="
- << std::hex << (int) shutdownFlags_;
- DestructorGuard dg(this);
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
-
- switch (state_) {
- case StateEnum::ESTABLISHED:
- case StateEnum::CONNECTING:
- {
- shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
- state_ = StateEnum::CLOSED;
-
- // If the write timeout was set, cancel it.
- writeTimeout_.cancelTimeout();
-
- // If we are registered for I/O events, unregister.
- if (eventFlags_ != EventHandler::NONE) {
- eventFlags_ = EventHandler::NONE;
- if (!updateEventRegistration()) {
- // We will have been moved into the error state.
- assert(state_ == StateEnum::ERROR);
- return;
- }
- }
-
- if (fd_ >= 0) {
- ioHandler_.changeHandlerFD(-1);
- doClose();
- }
-
- if (connectCallback_) {
- ConnectCallback* callback = connectCallback_;
- connectCallback_ = nullptr;
- callback->connectErr(socketClosedLocallyEx);
- }
-
- failAllWrites(socketClosedLocallyEx);
-
- if (readCallback_) {
- ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readEOF();
- }
- return;
- }
- case StateEnum::CLOSED:
- // Do nothing. It's possible that we are being called recursively
- // from inside a callback that we invoked inside another call to close()
- // that is still running.
- return;
- case StateEnum::ERROR:
- // Do nothing. The error handling code has performed (or is performing)
- // cleanup.
- return;
- case StateEnum::UNINIT:
- assert(eventFlags_ == EventHandler::NONE);
- assert(connectCallback_ == nullptr);
- assert(readCallback_ == nullptr);
- assert(writeReqHead_ == nullptr);
- shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
- state_ = StateEnum::CLOSED;
- return;
- }
-
- LOG(DFATAL) << "AsyncSocket::closeNow() (this=" << this << ", fd=" << fd_
- << ") called in unknown state " << state_;
-}
-
-void AsyncSocket::closeWithReset() {
- // Enable SO_LINGER, with the linger timeout set to 0.
- // This will trigger a TCP reset when we close the socket.
- if (fd_ >= 0) {
- struct linger optLinger = {1, 0};
- if (setSockOpt(SOL_SOCKET, SO_LINGER, &optLinger) != 0) {
- VLOG(2) << "AsyncSocket::closeWithReset(): error setting SO_LINGER "
- << "on " << fd_ << ": errno=" << errno;
- }
- }
-
- // Then let closeNow() take care of the rest
- closeNow();
-}
-
-void AsyncSocket::shutdownWrite() {
- VLOG(5) << "AsyncSocket::shutdownWrite(): this=" << this << ", fd=" << fd_
- << ", state=" << state_ << ", shutdownFlags="
- << std::hex << (int) shutdownFlags_;
-
- // If there are no pending writes, shutdownWrite() is identical to
- // shutdownWriteNow().
- if (writeReqHead_ == nullptr) {
- shutdownWriteNow();
- return;
- }
-
- assert(eventBase_->isInEventBaseThread());
-
- // There are pending writes. Set SHUT_WRITE_PENDING so that the actual
- // shutdown will be performed once all writes complete.
- shutdownFlags_ |= SHUT_WRITE_PENDING;
-}
-
-void AsyncSocket::shutdownWriteNow() {
- VLOG(5) << "AsyncSocket::shutdownWriteNow(): this=" << this
- << ", fd=" << fd_ << ", state=" << state_
- << ", shutdownFlags=" << std::hex << (int) shutdownFlags_;
-
- if (shutdownFlags_ & SHUT_WRITE) {
- // Writes are already shutdown; nothing else to do.
- return;
- }
-
- // If SHUT_READ is already set, just call closeNow() to completely
- // close the socket. This can happen if close() was called with writes
- // pending, and then shutdownWriteNow() is called before all pending writes
- // complete.
- if (shutdownFlags_ & SHUT_READ) {
- closeNow();
- return;
- }
-
- DestructorGuard dg(this);
- assert(eventBase_ == nullptr || eventBase_->isInEventBaseThread());
-
- switch (static_cast<StateEnum>(state_)) {
- case StateEnum::ESTABLISHED:
- {
- shutdownFlags_ |= SHUT_WRITE;
-
- // If the write timeout was set, cancel it.
- writeTimeout_.cancelTimeout();
-
- // If we are registered for write events, unregister.
- if (!updateEventRegistration(0, EventHandler::WRITE)) {
- // We will have been moved into the error state.
- assert(state_ == StateEnum::ERROR);
- return;
- }
-
- // Shutdown writes on the file descriptor
- ::shutdown(fd_, SHUT_WR);
-
- // Immediately fail all write requests
- failAllWrites(socketShutdownForWritesEx);
- return;
- }
- case StateEnum::CONNECTING:
- {
- // Set the SHUT_WRITE_PENDING flag.
- // When the connection completes, it will check this flag,
- // shutdown the write half of the socket, and then set SHUT_WRITE.
- shutdownFlags_ |= SHUT_WRITE_PENDING;
-
- // Immediately fail all write requests
- failAllWrites(socketShutdownForWritesEx);
- return;
- }
- case StateEnum::UNINIT:
- // Callers normally shouldn't call shutdownWriteNow() before the socket
- // even starts connecting. Nonetheless, go ahead and set
- // SHUT_WRITE_PENDING. Once the socket eventually connects it will
- // immediately shut down the write side of the socket.
- shutdownFlags_ |= SHUT_WRITE_PENDING;
- return;
- case StateEnum::CLOSED:
- case StateEnum::ERROR:
- // We should never get here. SHUT_WRITE should always be set
- // in STATE_CLOSED and STATE_ERROR.
- VLOG(4) << "AsyncSocket::shutdownWriteNow() (this=" << this
- << ", fd=" << fd_ << ") in unexpected state " << state_
- << " with SHUT_WRITE not set ("
- << std::hex << (int) shutdownFlags_ << ")";
- assert(false);
- return;
- }
-
- LOG(DFATAL) << "AsyncSocket::shutdownWriteNow() (this=" << this << ", fd="
- << fd_ << ") called in unknown state " << state_;
-}
-
-bool AsyncSocket::readable() const {
- if (fd_ == -1) {
- return false;
- }
- struct pollfd fds[1];
- fds[0].fd = fd_;
- fds[0].events = POLLIN;
- fds[0].revents = 0;
- int rc = poll(fds, 1, 0);
- return rc == 1;
-}
-
-bool AsyncSocket::isPending() const {
- return ioHandler_.isPending();
-}
-
-bool AsyncSocket::hangup() const {
- if (fd_ == -1) {
- // sanity check, no one should ask for hangup if we are not connected.
- assert(false);
- return false;
- }
-#ifdef POLLRDHUP // Linux-only
- struct pollfd fds[1];
- fds[0].fd = fd_;
- fds[0].events = POLLRDHUP|POLLHUP;
- fds[0].revents = 0;
- poll(fds, 1, 0);
- return (fds[0].revents & (POLLRDHUP|POLLHUP)) != 0;
-#else
- return false;
-#endif
-}
-
-bool AsyncSocket::good() const {
- return ((state_ == StateEnum::CONNECTING ||
- state_ == StateEnum::ESTABLISHED) &&
- (shutdownFlags_ == 0) && (eventBase_ != nullptr));
-}
-
-bool AsyncSocket::error() const {
- return (state_ == StateEnum::ERROR);
-}
-
-void AsyncSocket::attachEventBase(EventBase* eventBase) {
- VLOG(5) << "AsyncSocket::attachEventBase(this=" << this << ", fd=" << fd_
- << ", old evb=" << eventBase_ << ", new evb=" << eventBase
- << ", state=" << state_ << ", events="
- << std::hex << eventFlags_ << ")";
- assert(eventBase_ == nullptr);
- assert(eventBase->isInEventBaseThread());
-
- eventBase_ = eventBase;
- ioHandler_.attachEventBase(eventBase);
- writeTimeout_.attachEventBase(eventBase);
-}
-
-void AsyncSocket::detachEventBase() {
- VLOG(5) << "AsyncSocket::detachEventBase(this=" << this << ", fd=" << fd_
- << ", old evb=" << eventBase_ << ", state=" << state_
- << ", events=" << std::hex << eventFlags_ << ")";
- assert(eventBase_ != nullptr);
- assert(eventBase_->isInEventBaseThread());
-
- eventBase_ = nullptr;
- ioHandler_.detachEventBase();
- writeTimeout_.detachEventBase();
-}
-
-bool AsyncSocket::isDetachable() const {
- DCHECK(eventBase_ != nullptr);
- DCHECK(eventBase_->isInEventBaseThread());
-
- return !ioHandler_.isHandlerRegistered() && !writeTimeout_.isScheduled();
-}
-
-void AsyncSocket::getLocalAddress(folly::SocketAddress* address) const {
- address->setFromLocalAddress(fd_);
-}
-
-void AsyncSocket::getPeerAddress(folly::SocketAddress* address) const {
- if (!addr_.isInitialized()) {
- addr_.setFromPeerAddress(fd_);
- }
- *address = addr_;
-}
-
-int AsyncSocket::setNoDelay(bool noDelay) {
- if (fd_ < 0) {
- VLOG(4) << "AsyncSocket::setNoDelay() called on non-open socket "
- << this << "(state=" << state_ << ")";
- return EINVAL;
-
- }
-
- int value = noDelay ? 1 : 0;
- if (setsockopt(fd_, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value)) != 0) {
- int errnoCopy = errno;
- VLOG(2) << "failed to update TCP_NODELAY option on AsyncSocket "
- << this << " (fd=" << fd_ << ", state=" << state_ << "): "
- << strerror(errnoCopy);
- return errnoCopy;
- }
-
- return 0;
-}
-
-int AsyncSocket::setCongestionFlavor(const std::string &cname) {
-
- #ifndef TCP_CONGESTION
- #define TCP_CONGESTION 13
- #endif
-
- if (fd_ < 0) {
- VLOG(4) << "AsyncSocket::setCongestionFlavor() called on non-open "
- << "socket " << this << "(state=" << state_ << ")";
- return EINVAL;
-
- }
-
- if (setsockopt(fd_, IPPROTO_TCP, TCP_CONGESTION, cname.c_str(),
- cname.length() + 1) != 0) {
- int errnoCopy = errno;
- VLOG(2) << "failed to update TCP_CONGESTION option on AsyncSocket "
- << this << "(fd=" << fd_ << ", state=" << state_ << "): "
- << strerror(errnoCopy);
- return errnoCopy;
- }
-
- return 0;
-}
-
-int AsyncSocket::setQuickAck(bool quickack) {
- if (fd_ < 0) {
- VLOG(4) << "AsyncSocket::setQuickAck() called on non-open socket "
- << this << "(state=" << state_ << ")";
- return EINVAL;
-
- }
-
-#ifdef TCP_QUICKACK // Linux-only
- int value = quickack ? 1 : 0;
- if (setsockopt(fd_, IPPROTO_TCP, TCP_QUICKACK, &value, sizeof(value)) != 0) {
- int errnoCopy = errno;
- VLOG(2) << "failed to update TCP_QUICKACK option on AsyncSocket"
- << this << "(fd=" << fd_ << ", state=" << state_ << "): "
- << strerror(errnoCopy);
- return errnoCopy;
- }
-
- return 0;
-#else
- return ENOSYS;
-#endif
-}
-
-int AsyncSocket::setSendBufSize(size_t bufsize) {
- if (fd_ < 0) {
- VLOG(4) << "AsyncSocket::setSendBufSize() called on non-open socket "
- << this << "(state=" << state_ << ")";
- return EINVAL;
- }
-
- if (setsockopt(fd_, SOL_SOCKET, SO_SNDBUF, &bufsize, sizeof(bufsize)) !=0) {
- int errnoCopy = errno;
- VLOG(2) << "failed to update SO_SNDBUF option on AsyncSocket"
- << this << "(fd=" << fd_ << ", state=" << state_ << "): "
- << strerror(errnoCopy);
- return errnoCopy;
- }
-
- return 0;
-}
-
-int AsyncSocket::setRecvBufSize(size_t bufsize) {
- if (fd_ < 0) {
- VLOG(4) << "AsyncSocket::setRecvBufSize() called on non-open socket "
- << this << "(state=" << state_ << ")";
- return EINVAL;
- }
-
- if (setsockopt(fd_, SOL_SOCKET, SO_RCVBUF, &bufsize, sizeof(bufsize)) !=0) {
- int errnoCopy = errno;
- VLOG(2) << "failed to update SO_RCVBUF option on AsyncSocket"
- << this << "(fd=" << fd_ << ", state=" << state_ << "): "
- << strerror(errnoCopy);
- return errnoCopy;
- }
-
- return 0;
-}
-
-int AsyncSocket::setTCPProfile(int profd) {
- if (fd_ < 0) {
- VLOG(4) << "AsyncSocket::setTCPProfile() called on non-open socket "
- << this << "(state=" << state_ << ")";
- return EINVAL;
- }
-
- if (setsockopt(fd_, SOL_SOCKET, SO_SET_NAMESPACE, &profd, sizeof(int)) !=0) {
- int errnoCopy = errno;
- VLOG(2) << "failed to set socket namespace option on AsyncSocket"
- << this << "(fd=" << fd_ << ", state=" << state_ << "): "
- << strerror(errnoCopy);
- return errnoCopy;
- }
-
- return 0;
-}
-
-void AsyncSocket::ioReady(uint16_t events) noexcept {
- VLOG(7) << "AsyncSocket::ioRead() this=" << this << ", fd" << fd_
- << ", events=" << std::hex << events << ", state=" << state_;
- DestructorGuard dg(this);
- assert(events & EventHandler::READ_WRITE);
- assert(eventBase_->isInEventBaseThread());
-
- uint16_t relevantEvents = events & EventHandler::READ_WRITE;
- if (relevantEvents == EventHandler::READ) {
- handleRead();
- } else if (relevantEvents == EventHandler::WRITE) {
- handleWrite();
- } else if (relevantEvents == EventHandler::READ_WRITE) {
- EventBase* originalEventBase = eventBase_;
- // If both read and write events are ready, process writes first.
- handleWrite();
-
- // Return now if handleWrite() detached us from our EventBase
- if (eventBase_ != originalEventBase) {
- return;
- }
-
- // Only call handleRead() if a read callback is still installed.
- // (It's possible that the read callback was uninstalled during
- // handleWrite().)
- if (readCallback_) {
- handleRead();
- }
- } else {
- VLOG(4) << "AsyncSocket::ioRead() called with unexpected events "
- << std::hex << events << "(this=" << this << ")";
- abort();
- }
-}
-
-ssize_t AsyncSocket::performRead(void* buf, size_t buflen) {
- ssize_t bytes = recv(fd_, buf, buflen, MSG_DONTWAIT);
- if (bytes < 0) {
- if (errno == EAGAIN || errno == EWOULDBLOCK) {
- // No more data to read right now.
- return READ_BLOCKING;
- } else {
- return READ_ERROR;
- }
- } else {
- appBytesReceived_ += bytes;
- return bytes;
- }
-}
-
-void AsyncSocket::handleRead() noexcept {
- VLOG(5) << "AsyncSocket::handleRead() this=" << this << ", fd=" << fd_
- << ", state=" << state_;
- assert(state_ == StateEnum::ESTABLISHED);
- assert((shutdownFlags_ & SHUT_READ) == 0);
- assert(readCallback_ != nullptr);
- assert(eventFlags_ & EventHandler::READ);
-
- // Loop until:
- // - a read attempt would block
- // - readCallback_ is uninstalled
- // - the number of loop iterations exceeds the optional maximum
- // - this AsyncSocket is moved to another EventBase
- //
- // When we invoke readDataAvailable() it may uninstall the readCallback_,
- // which is why need to check for it here.
- //
- // The last bullet point is slightly subtle. readDataAvailable() may also
- // detach this socket from this EventBase. However, before
- // readDataAvailable() returns another thread may pick it up, attach it to
- // a different EventBase, and install another readCallback_. We need to
- // exit immediately after readDataAvailable() returns if the eventBase_ has
- // changed. (The caller must perform some sort of locking to transfer the
- // AsyncSocket between threads properly. This will be sufficient to ensure
- // that this thread sees the updated eventBase_ variable after
- // readDataAvailable() returns.)
- uint16_t numReads = 0;
- EventBase* originalEventBase = eventBase_;
- while (readCallback_ && eventBase_ == originalEventBase) {
- // Get the buffer to read into.
- void* buf = nullptr;
- size_t buflen = 0;
- try {
- readCallback_->getReadBuffer(&buf, &buflen);
- } catch (const AsyncSocketException& ex) {
- return failRead(__func__, ex);
- } catch (const std::exception& ex) {
- AsyncSocketException tex(AsyncSocketException::BAD_ARGS,
- string("ReadCallback::getReadBuffer() "
- "threw exception: ") +
- ex.what());
- return failRead(__func__, tex);
- } catch (...) {
- AsyncSocketException ex(AsyncSocketException::BAD_ARGS,
- "ReadCallback::getReadBuffer() threw "
- "non-exception type");
- return failRead(__func__, ex);
- }
- if (buf == nullptr || buflen == 0) {
- AsyncSocketException ex(AsyncSocketException::BAD_ARGS,
- "ReadCallback::getReadBuffer() returned "
- "empty buffer");
- return failRead(__func__, ex);
- }
-
- // Perform the read
- ssize_t bytesRead = performRead(buf, buflen);
- if (bytesRead > 0) {
- readCallback_->readDataAvailable(bytesRead);
- // Fall through and continue around the loop if the read
- // completely filled the available buffer.
- // Note that readCallback_ may have been uninstalled or changed inside
- // readDataAvailable().
- if (bytesRead < buflen) {
- return;
- }
- } else if (bytesRead == READ_BLOCKING) {
- // No more data to read right now.
- return;
- } else if (bytesRead == READ_ERROR) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("recv() failed"), errno);
- return failRead(__func__, ex);
- } else {
- assert(bytesRead == READ_EOF);
- // EOF
- shutdownFlags_ |= SHUT_READ;
- if (!updateEventRegistration(0, EventHandler::READ)) {
- // we've already been moved into STATE_ERROR
- assert(state_ == StateEnum::ERROR);
- assert(readCallback_ == nullptr);
- return;
- }
-
- ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readEOF();
- return;
- }
- if (maxReadsPerEvent_ && (++numReads >= maxReadsPerEvent_)) {
- return;
- }
- }
-}
-
-/**
- * This function attempts to write as much data as possible, until no more data
- * can be written.
- *
- * - If it sends all available data, it unregisters for write events, and stops
- * the writeTimeout_.
- *
- * - If not all of the data can be sent immediately, it reschedules
- * writeTimeout_ (if a non-zero timeout is set), and ensures the handler is
- * registered for write events.
- */
-void AsyncSocket::handleWrite() noexcept {
- VLOG(5) << "AsyncSocket::handleWrite() this=" << this << ", fd=" << fd_
- << ", state=" << state_;
- if (state_ == StateEnum::CONNECTING) {
- handleConnect();
- return;
- }
-
- // Normal write
- assert(state_ == StateEnum::ESTABLISHED);
- assert((shutdownFlags_ & SHUT_WRITE) == 0);
- assert(writeReqHead_ != nullptr);
-
- // Loop until we run out of write requests,
- // or until this socket is moved to another EventBase.
- // (See the comment in handleRead() explaining how this can happen.)
- EventBase* originalEventBase = eventBase_;
- while (writeReqHead_ != nullptr && eventBase_ == originalEventBase) {
- uint32_t countWritten;
- uint32_t partialWritten;
- WriteFlags writeFlags = writeReqHead_->flags();
- if (writeReqHead_->getNext() != nullptr) {
- writeFlags = writeFlags | WriteFlags::CORK;
- }
- int bytesWritten = performWrite(writeReqHead_->getOps(),
- writeReqHead_->getOpCount(),
- writeFlags, &countWritten, &partialWritten);
- if (bytesWritten < 0) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("writev() failed"), errno);
- return failWrite(__func__, ex);
- } else if (countWritten == writeReqHead_->getOpCount()) {
- // We finished this request
- WriteRequest* req = writeReqHead_;
- writeReqHead_ = req->getNext();
-
- if (writeReqHead_ == nullptr) {
- writeReqTail_ = nullptr;
- // This is the last write request.
- // Unregister for write events and cancel the send timer
- // before we invoke the callback. We have to update the state properly
- // before calling the callback, since it may want to detach us from
- // the EventBase.
- if (eventFlags_ & EventHandler::WRITE) {
- if (!updateEventRegistration(0, EventHandler::WRITE)) {
- assert(state_ == StateEnum::ERROR);
- return;
- }
- // Stop the send timeout
- writeTimeout_.cancelTimeout();
- }
- assert(!writeTimeout_.isScheduled());
-
- // If SHUT_WRITE_PENDING is set, we should shutdown the socket after
- // we finish sending the last write request.
- //
- // We have to do this before invoking writeSuccess(), since
- // writeSuccess() may detach us from our EventBase.
- if (shutdownFlags_ & SHUT_WRITE_PENDING) {
- assert(connectCallback_ == nullptr);
- shutdownFlags_ |= SHUT_WRITE;
-
- if (shutdownFlags_ & SHUT_READ) {
- // Reads have already been shutdown. Fully close the socket and
- // move to STATE_CLOSED.
- //
- // Note: This code currently moves us to STATE_CLOSED even if
- // close() hasn't ever been called. This can occur if we have
- // received EOF from the peer and shutdownWrite() has been called
- // locally. Should we bother staying in STATE_ESTABLISHED in this
- // case, until close() is actually called? I can't think of a
- // reason why we would need to do so. No other operations besides
- // calling close() or destroying the socket can be performed at
- // this point.
- assert(readCallback_ == nullptr);
- state_ = StateEnum::CLOSED;
- if (fd_ >= 0) {
- ioHandler_.changeHandlerFD(-1);
- doClose();
- }
- } else {
- // Reads are still enabled, so we are only doing a half-shutdown
- ::shutdown(fd_, SHUT_WR);
- }
- }
- }
-
- // Invoke the callback
- WriteCallback* callback = req->getCallback();
- req->destroy();
- if (callback) {
- callback->writeSuccess();
- }
- // We'll continue around the loop, trying to write another request
- } else {
- // Partial write.
- writeReqHead_->consume(countWritten, partialWritten, bytesWritten);
- // Stop after a partial write; it's highly likely that a subsequent write
- // attempt will just return EAGAIN.
- //
- // Ensure that we are registered for write events.
- if ((eventFlags_ & EventHandler::WRITE) == 0) {
- if (!updateEventRegistration(EventHandler::WRITE, 0)) {
- assert(state_ == StateEnum::ERROR);
- return;
- }
- }
-
- // Reschedule the send timeout, since we have made some write progress.
- if (sendTimeout_ > 0) {
- if (!writeTimeout_.scheduleTimeout(sendTimeout_)) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to reschedule write timeout"));
- return failWrite(__func__, ex);
- }
- }
- return;
- }
- }
-}
-
-void AsyncSocket::checkForImmediateRead() noexcept {
- // We currently don't attempt to perform optimistic reads in AsyncSocket.
- // (However, note that some subclasses do override this method.)
- //
- // Simply calling handleRead() here would be bad, as this would call
- // readCallback_->getReadBuffer(), forcing the callback to allocate a read
- // buffer even though no data may be available. This would waste lots of
- // memory, since the buffer will sit around unused until the socket actually
- // becomes readable.
- //
- // Checking if the socket is readable now also seems like it would probably
- // be a pessimism. In most cases it probably wouldn't be readable, and we
- // would just waste an extra system call. Even if it is readable, waiting to
- // find out from libevent on the next event loop doesn't seem that bad.
-}
-
-void AsyncSocket::handleInitialReadWrite() noexcept {
- // Our callers should already be holding a DestructorGuard, but grab
- // one here just to make sure, in case one of our calling code paths ever
- // changes.
- DestructorGuard dg(this);
-
- // If we have a readCallback_, make sure we enable read events. We
- // may already be registered for reads if connectSuccess() set
- // the read calback.
- if (readCallback_ && !(eventFlags_ & EventHandler::READ)) {
- assert(state_ == StateEnum::ESTABLISHED);
- assert((shutdownFlags_ & SHUT_READ) == 0);
- if (!updateEventRegistration(EventHandler::READ, 0)) {
- assert(state_ == StateEnum::ERROR);
- return;
- }
- checkForImmediateRead();
- } else if (readCallback_ == nullptr) {
- // Unregister for read events.
- updateEventRegistration(0, EventHandler::READ);
- }
-
- // If we have write requests pending, try to send them immediately.
- // Since we just finished accepting, there is a very good chance that we can
- // write without blocking.
- //
- // However, we only process them if EventHandler::WRITE is not already set,
- // which means that we're already blocked on a write attempt. (This can
- // happen if connectSuccess() called write() before returning.)
- if (writeReqHead_ && !(eventFlags_ & EventHandler::WRITE)) {
- // Call handleWrite() to perform write processing.
- handleWrite();
- } else if (writeReqHead_ == nullptr) {
- // Unregister for write event.
- updateEventRegistration(0, EventHandler::WRITE);
- }
-}
-
-void AsyncSocket::handleConnect() noexcept {
- VLOG(5) << "AsyncSocket::handleConnect() this=" << this << ", fd=" << fd_
- << ", state=" << state_;
- assert(state_ == StateEnum::CONNECTING);
- // SHUT_WRITE can never be set while we are still connecting;
- // SHUT_WRITE_PENDING may be set, be we only set SHUT_WRITE once the connect
- // finishes
- assert((shutdownFlags_ & SHUT_WRITE) == 0);
-
- // In case we had a connect timeout, cancel the timeout
- writeTimeout_.cancelTimeout();
- // We don't use a persistent registration when waiting on a connect event,
- // so we have been automatically unregistered now. Update eventFlags_ to
- // reflect reality.
- assert(eventFlags_ == EventHandler::WRITE);
- eventFlags_ = EventHandler::NONE;
-
- // Call getsockopt() to check if the connect succeeded
- int error;
- socklen_t len = sizeof(error);
- int rv = getsockopt(fd_, SOL_SOCKET, SO_ERROR, &error, &len);
- if (rv != 0) {
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("error calling getsockopt() after connect"),
- errno);
- VLOG(4) << "AsyncSocket::handleConnect(this=" << this << ", fd="
- << fd_ << " host=" << addr_.describe()
- << ") exception:" << ex.what();
- return failConnect(__func__, ex);
- }
-
- if (error != 0) {
- AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
- "connect failed", error);
- VLOG(1) << "AsyncSocket::handleConnect(this=" << this << ", fd="
- << fd_ << " host=" << addr_.describe()
- << ") exception: " << ex.what();
- return failConnect(__func__, ex);
- }
-
- // Move into STATE_ESTABLISHED
- state_ = StateEnum::ESTABLISHED;
-
- // If SHUT_WRITE_PENDING is set and we don't have any write requests to
- // perform, immediately shutdown the write half of the socket.
- if ((shutdownFlags_ & SHUT_WRITE_PENDING) && writeReqHead_ == nullptr) {
- // SHUT_READ shouldn't be set. If close() is called on the socket while we
- // are still connecting we just abort the connect rather than waiting for
- // it to complete.
- assert((shutdownFlags_ & SHUT_READ) == 0);
- ::shutdown(fd_, SHUT_WR);
- shutdownFlags_ |= SHUT_WRITE;
- }
-
- VLOG(7) << "AsyncSocket " << this << ": fd " << fd_
- << "successfully connected; state=" << state_;
-
- // Remember the EventBase we are attached to, before we start invoking any
- // callbacks (since the callbacks may call detachEventBase()).
- EventBase* originalEventBase = eventBase_;
-
- // Call the connect callback.
- if (connectCallback_) {
- ConnectCallback* callback = connectCallback_;
- connectCallback_ = nullptr;
- callback->connectSuccess();
- }
-
- // Note that the connect callback may have changed our state.
- // (set or unset the read callback, called write(), closed the socket, etc.)
- // The following code needs to handle these situations correctly.
- //
- // If the socket has been closed, readCallback_ and writeReqHead_ will
- // always be nullptr, so that will prevent us from trying to read or write.
- //
- // The main thing to check for is if eventBase_ is still originalEventBase.
- // If not, we have been detached from this event base, so we shouldn't
- // perform any more operations.
- if (eventBase_ != originalEventBase) {
- return;
- }
-
- handleInitialReadWrite();
-}
-
-void AsyncSocket::timeoutExpired() noexcept {
- VLOG(7) << "AsyncSocket " << this << ", fd " << fd_ << ": timeout expired: "
- << "state=" << state_ << ", events=" << std::hex << eventFlags_;
- DestructorGuard dg(this);
- assert(eventBase_->isInEventBaseThread());
-
- if (state_ == StateEnum::CONNECTING) {
- // connect() timed out
- // Unregister for I/O events.
- AsyncSocketException ex(AsyncSocketException::TIMED_OUT,
- "connect timed out");
- failConnect(__func__, ex);
- } else {
- // a normal write operation timed out
- assert(state_ == StateEnum::ESTABLISHED);
- AsyncSocketException ex(AsyncSocketException::TIMED_OUT, "write timed out");
- failWrite(__func__, ex);
- }
-}
-
-ssize_t AsyncSocket::performWrite(const iovec* vec,
- uint32_t count,
- WriteFlags flags,
- uint32_t* countWritten,
- uint32_t* partialWritten) {
- // We use sendmsg() instead of writev() so that we can pass in MSG_NOSIGNAL
- // We correctly handle EPIPE errors, so we never want to receive SIGPIPE
- // (since it may terminate the program if the main program doesn't explicitly
- // ignore it).
- struct msghdr msg;
- msg.msg_name = nullptr;
- msg.msg_namelen = 0;
- msg.msg_iov = const_cast<iovec *>(vec);
-#ifdef IOV_MAX // not defined on Android
- msg.msg_iovlen = std::min(count, (uint32_t)IOV_MAX);
-#else
- msg.msg_iovlen = std::min(count, (uint32_t)UIO_MAXIOV);
-#endif
- msg.msg_control = nullptr;
- msg.msg_controllen = 0;
- msg.msg_flags = 0;
-
- int msg_flags = MSG_DONTWAIT;
-
-#ifdef MSG_NOSIGNAL // Linux-only
- msg_flags |= MSG_NOSIGNAL;
- if (isSet(flags, WriteFlags::CORK)) {
- // MSG_MORE tells the kernel we have more data to send, so wait for us to
- // give it the rest of the data rather than immediately sending a partial
- // frame, even when TCP_NODELAY is enabled.
- msg_flags |= MSG_MORE;
- }
-#endif
- if (isSet(flags, WriteFlags::EOR)) {
- // marks that this is the last byte of a record (response)
- msg_flags |= MSG_EOR;
- }
- ssize_t totalWritten = ::sendmsg(fd_, &msg, msg_flags);
- if (totalWritten < 0) {
- if (errno == EAGAIN) {
- // TCP buffer is full; we can't write any more data right now.
- *countWritten = 0;
- *partialWritten = 0;
- return 0;
- }
- // error
- *countWritten = 0;
- *partialWritten = 0;
- return -1;
- }
-
- appBytesWritten_ += totalWritten;
-
- uint32_t bytesWritten;
- uint32_t n;
- for (bytesWritten = totalWritten, n = 0; n < count; ++n) {
- const iovec* v = vec + n;
- if (v->iov_len > bytesWritten) {
- // Partial write finished in the middle of this iovec
- *countWritten = n;
- *partialWritten = bytesWritten;
- return totalWritten;
- }
-
- bytesWritten -= v->iov_len;
- }
-
- assert(bytesWritten == 0);
- *countWritten = n;
- *partialWritten = 0;
- return totalWritten;
-}
-
-/**
- * Re-register the EventHandler after eventFlags_ has changed.
- *
- * If an error occurs, fail() is called to move the socket into the error state
- * and call all currently installed callbacks. After an error, the
- * AsyncSocket is completely unregistered.
- *
- * @return Returns true on succcess, or false on error.
- */
-bool AsyncSocket::updateEventRegistration() {
- VLOG(5) << "AsyncSocket::updateEventRegistration(this=" << this
- << ", fd=" << fd_ << ", evb=" << eventBase_ << ", state=" << state_
- << ", events=" << std::hex << eventFlags_;
- assert(eventBase_->isInEventBaseThread());
- if (eventFlags_ == EventHandler::NONE) {
- ioHandler_.unregisterHandler();
- return true;
- }
-
- // Always register for persistent events, so we don't have to re-register
- // after being called back.
- if (!ioHandler_.registerHandler(eventFlags_ | EventHandler::PERSIST)) {
- eventFlags_ = EventHandler::NONE; // we're not registered after error
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("failed to update AsyncSocket event registration"));
- fail("updateEventRegistration", ex);
- return false;
- }
-
- return true;
-}
-
-bool AsyncSocket::updateEventRegistration(uint16_t enable,
- uint16_t disable) {
- uint16_t oldFlags = eventFlags_;
- eventFlags_ |= enable;
- eventFlags_ &= ~disable;
- if (eventFlags_ == oldFlags) {
- return true;
- } else {
- return updateEventRegistration();
- }
-}
-
-void AsyncSocket::startFail() {
- // startFail() should only be called once
- assert(state_ != StateEnum::ERROR);
- assert(getDestructorGuardCount() > 0);
- state_ = StateEnum::ERROR;
- // Ensure that SHUT_READ and SHUT_WRITE are set,
- // so all future attempts to read or write will be rejected
- shutdownFlags_ |= (SHUT_READ | SHUT_WRITE);
-
- if (eventFlags_ != EventHandler::NONE) {
- eventFlags_ = EventHandler::NONE;
- ioHandler_.unregisterHandler();
- }
- writeTimeout_.cancelTimeout();
-
- if (fd_ >= 0) {
- ioHandler_.changeHandlerFD(-1);
- doClose();
- }
-}
-
-void AsyncSocket::finishFail() {
- assert(state_ == StateEnum::ERROR);
- assert(getDestructorGuardCount() > 0);
-
- AsyncSocketException ex(AsyncSocketException::INTERNAL_ERROR,
- withAddr("socket closing after error"));
- if (connectCallback_) {
- ConnectCallback* callback = connectCallback_;
- connectCallback_ = nullptr;
- callback->connectErr(ex);
- }
-
- failAllWrites(ex);
-
- if (readCallback_) {
- ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readErr(ex);
- }
-}
-
-void AsyncSocket::fail(const char* fn, const AsyncSocketException& ex) {
- VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
- << state_ << " host=" << addr_.describe()
- << "): failed in " << fn << "(): "
- << ex.what();
- startFail();
- finishFail();
-}
-
-void AsyncSocket::failConnect(const char* fn, const AsyncSocketException& ex) {
- VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
- << state_ << " host=" << addr_.describe()
- << "): failed while connecting in " << fn << "(): "
- << ex.what();
- startFail();
-
- if (connectCallback_ != nullptr) {
- ConnectCallback* callback = connectCallback_;
- connectCallback_ = nullptr;
- callback->connectErr(ex);
- }
-
- finishFail();
-}
-
-void AsyncSocket::failRead(const char* fn, const AsyncSocketException& ex) {
- VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
- << state_ << " host=" << addr_.describe()
- << "): failed while reading in " << fn << "(): "
- << ex.what();
- startFail();
-
- if (readCallback_ != nullptr) {
- ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readErr(ex);
- }
-
- finishFail();
-}
-
-void AsyncSocket::failWrite(const char* fn, const AsyncSocketException& ex) {
- VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
- << state_ << " host=" << addr_.describe()
- << "): failed while writing in " << fn << "(): "
- << ex.what();
- startFail();
-
- // Only invoke the first write callback, since the error occurred while
- // writing this request. Let any other pending write callbacks be invoked in
- // finishFail().
- if (writeReqHead_ != nullptr) {
- WriteRequest* req = writeReqHead_;
- writeReqHead_ = req->getNext();
- WriteCallback* callback = req->getCallback();
- uint32_t bytesWritten = req->getBytesWritten();
- req->destroy();
- if (callback) {
- callback->writeErr(bytesWritten, ex);
- }
- }
-
- finishFail();
-}
-
-void AsyncSocket::failWrite(const char* fn, WriteCallback* callback,
- size_t bytesWritten,
- const AsyncSocketException& ex) {
- // This version of failWrite() is used when the failure occurs before
- // we've added the callback to writeReqHead_.
- VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_ << ", state="
- << state_ << " host=" << addr_.describe()
- <<"): failed while writing in " << fn << "(): "
- << ex.what();
- startFail();
-
- if (callback != nullptr) {
- callback->writeErr(bytesWritten, ex);
- }
-
- finishFail();
-}
-
-void AsyncSocket::failAllWrites(const AsyncSocketException& ex) {
- // Invoke writeError() on all write callbacks.
- // This is used when writes are forcibly shutdown with write requests
- // pending, or when an error occurs with writes pending.
- while (writeReqHead_ != nullptr) {
- WriteRequest* req = writeReqHead_;
- writeReqHead_ = req->getNext();
- WriteCallback* callback = req->getCallback();
- if (callback) {
- callback->writeErr(req->getBytesWritten(), ex);
- }
- req->destroy();
- }
-}
-
-void AsyncSocket::invalidState(ConnectCallback* callback) {
- VLOG(5) << "AsyncSocket(this=" << this << ", fd=" << fd_
- << "): connect() called in invalid state " << state_;
-
- /*
- * The invalidState() methods don't use the normal failure mechanisms,
- * since we don't know what state we are in. We don't want to call
- * startFail()/finishFail() recursively if we are already in the middle of
- * cleaning up.
- */
-
- AsyncSocketException ex(AsyncSocketException::ALREADY_OPEN,
- "connect() called with socket in invalid state");
- if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
- if (callback) {
- callback->connectErr(ex);
- }
- } else {
- // We can't use failConnect() here since connectCallback_
- // may already be set to another callback. Invoke this ConnectCallback
- // here; any other connectCallback_ will be invoked in finishFail()
- startFail();
- if (callback) {
- callback->connectErr(ex);
- }
- finishFail();
- }
-}
-
-void AsyncSocket::invalidState(ReadCallback* callback) {
- VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
- << "): setReadCallback(" << callback
- << ") called in invalid state " << state_;
-
- AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
- "setReadCallback() called with socket in "
- "invalid state");
- if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
- if (callback) {
- callback->readErr(ex);
- }
- } else {
- startFail();
- if (callback) {
- callback->readErr(ex);
- }
- finishFail();
- }
-}
-
-void AsyncSocket::invalidState(WriteCallback* callback) {
- VLOG(4) << "AsyncSocket(this=" << this << ", fd=" << fd_
- << "): write() called in invalid state " << state_;
-
- AsyncSocketException ex(AsyncSocketException::NOT_OPEN,
- withAddr("write() called with socket in invalid state"));
- if (state_ == StateEnum::CLOSED || state_ == StateEnum::ERROR) {
- if (callback) {
- callback->writeErr(0, ex);
- }
- } else {
- startFail();
- if (callback) {
- callback->writeErr(0, ex);
- }
- finishFail();
- }
-}
-
-void AsyncSocket::doClose() {
- if (fd_ == -1) return;
- if (shutdownSocketSet_) {
- shutdownSocketSet_->close(fd_);
- } else {
- ::close(fd_);
- }
- fd_ = -1;
-}
-
-std::ostream& operator << (std::ostream& os,
- const AsyncSocket::StateEnum& state) {
- os << static_cast<int>(state);
- return os;
-}
-
-std::string AsyncSocket::withAddr(const std::string& s) {
- // Don't use addr_ directly because it may not be initialized
- // e.g. if constructed from fd
- folly::SocketAddress peer, local;
- try {
- getPeerAddress(&peer);
- getLocalAddress(&local);
- } catch (const std::exception&) {
- // ignore
- } catch (...) {
- // ignore
- }
- return s + " (peer=" + peer.describe() + ", local=" + local.describe() + ")";
-}
-
-} // folly
+++ /dev/null
-/*
- * Copyright 2014 Facebook, Inc.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-#pragma once
-
-#include <sys/types.h>
-#include <sys/socket.h>
-#include <glog/logging.h>
-#include <folly/SocketAddress.h>
-#include <folly/io/ShutdownSocketSet.h>
-#include <folly/io/IOBuf.h>
-#include <folly/io/async/AsyncTimeout.h>
-#include <folly/io/async/AsyncSocketException.h>
-#include <folly/io/async/AsyncTransport.h>
-#include <folly/io/async/EventHandler.h>
-#include <folly/io/async/DelayedDestruction.h>
-
-#include <memory>
-#include <map>
-
-namespace folly {
-
-/**
- * A class for performing asynchronous I/O on a socket.
- *
- * AsyncSocket allows users to asynchronously wait for data on a socket, and
- * to asynchronously send data.
- *
- * The APIs for reading and writing are intentionally asymmetric. Waiting for
- * data to read is a persistent API: a callback is installed, and is notified
- * whenever new data is available. It continues to be notified of new events
- * until it is uninstalled.
- *
- * AsyncSocket does not provide read timeout functionality, because it
- * typically cannot determine when the timeout should be active. Generally, a
- * timeout should only be enabled when processing is blocked waiting on data
- * from the remote endpoint. For server sockets, the timeout should not be
- * active if the server is currently processing one or more outstanding
- * requests for this socket. For client sockets, the timeout should not be
- * active if there are no requests pending on the socket. Additionally, if a
- * client has multiple pending requests, it will ususally want a separate
- * timeout for each request, rather than a single read timeout.
- *
- * The write API is fairly intuitive: a user can request to send a block of
- * data, and a callback will be informed once the entire block has been
- * transferred to the kernel, or on error. AsyncSocket does provide a send
- * timeout, since most callers want to give up if the remote end stops
- * responding and no further progress can be made sending the data.
- */
-
-class AsyncSocket : virtual public AsyncTransport {
- public:
- typedef std::unique_ptr<AsyncSocket, Destructor> UniquePtr;
-
- class ConnectCallback {
- public:
- virtual ~ConnectCallback() {}
-
- /**
- * connectSuccess() will be invoked when the connection has been
- * successfully established.
- */
- virtual void connectSuccess() noexcept = 0;
-
- /**
- * connectErr() will be invoked if the connection attempt fails.
- *
- * @param ex An exception describing the error that occurred.
- */
- virtual void connectErr(const AsyncSocketException& ex)
- noexcept = 0;
- };
-
- class ReadCallback {
- public:
- virtual ~ReadCallback() {}
-
- /**
- * When data becomes available, getReadBuffer() will be invoked to get the
- * buffer into which data should be read.
- *
- * This method allows the ReadCallback to delay buffer allocation until
- * data becomes available. This allows applications to manage large
- * numbers of idle connections, without having to maintain a separate read
- * buffer for each idle connection.
- *
- * It is possible that in some cases, getReadBuffer() may be called
- * multiple times before readDataAvailable() is invoked. In this case, the
- * data will be written to the buffer returned from the most recent call to
- * readDataAvailable(). If the previous calls to readDataAvailable()
- * returned different buffers, the ReadCallback is responsible for ensuring
- * that they are not leaked.
- *
- * If getReadBuffer() throws an exception, returns a nullptr buffer, or
- * returns a 0 length, the ReadCallback will be uninstalled and its
- * readError() method will be invoked.
- *
- * getReadBuffer() is not allowed to change the transport state before it
- * returns. (For example, it should never uninstall the read callback, or
- * set a different read callback.)
- *
- * @param bufReturn getReadBuffer() should update *bufReturn to contain the
- * address of the read buffer. This parameter will never
- * be nullptr.
- * @param lenReturn getReadBuffer() should update *lenReturn to contain the
- * maximum number of bytes that may be written to the read
- * buffer. This parameter will never be nullptr.
- */
- virtual void getReadBuffer(void** bufReturn, size_t* lenReturn) = 0;
-
- /**
- * readDataAvailable() will be invoked when data has been successfully read
- * into the buffer returned by the last call to getReadBuffer().
- *
- * The read callback remains installed after readDataAvailable() returns.
- * It must be explicitly uninstalled to stop receiving read events.
- * getReadBuffer() will be called at least once before each call to
- * readDataAvailable(). getReadBuffer() will also be called before any
- * call to readEOF().
- *
- * @param len The number of bytes placed in the buffer.
- */
- virtual void readDataAvailable(size_t len) noexcept = 0;
-
- /**
- * readEOF() will be invoked when the transport is closed.
- *
- * The read callback will be automatically uninstalled immediately before
- * readEOF() is invoked.
- */
- virtual void readEOF() noexcept = 0;
-
- /**
- * readError() will be invoked if an error occurs reading from the
- * transport.
- *
- * The read callback will be automatically uninstalled immediately before
- * readError() is invoked.
- *
- * @param ex An exception describing the error that occurred.
- */
- virtual void readErr(const AsyncSocketException& ex)
- noexcept = 0;
- };
-
- class WriteCallback {
- public:
- virtual ~WriteCallback() {}
-
- /**
- * writeSuccess() will be invoked when all of the data has been
- * successfully written.
- *
- * Note that this mainly signals that the buffer containing the data to
- * write is no longer needed and may be freed or re-used. It does not
- * guarantee that the data has been fully transmitted to the remote
- * endpoint. For example, on socket-based transports, writeSuccess() only
- * indicates that the data has been given to the kernel for eventual
- * transmission.
- */
- virtual void writeSuccess() noexcept = 0;
-
- /**
- * writeError() will be invoked if an error occurs writing the data.
- *
- * @param bytesWritten The number of bytes that were successfull
- * @param ex An exception describing the error that occurred.
- */
- virtual void writeErr(size_t bytesWritten,
- const AsyncSocketException& ex)
- noexcept = 0;
- };
-
- /**
- * Create a new unconnected AsyncSocket.
- *
- * connect() must later be called on this socket to establish a connection.
- */
- explicit AsyncSocket(EventBase* evb);
-
- void setShutdownSocketSet(ShutdownSocketSet* ss);
-
- /**
- * Create a new AsyncSocket and begin the connection process.
- *
- * @param evb EventBase that will manage this socket.
- * @param address The address to connect to.
- * @param connectTimeout Optional timeout in milliseconds for the connection
- * attempt.
- */
- AsyncSocket(EventBase* evb,
- const folly::SocketAddress& address,
- uint32_t connectTimeout = 0);
-
- /**
- * Create a new AsyncSocket and begin the connection process.
- *
- * @param evb EventBase that will manage this socket.
- * @param ip IP address to connect to (dotted-quad).
- * @param port Destination port in host byte order.
- * @param connectTimeout Optional timeout in milliseconds for the connection
- * attempt.
- */
- AsyncSocket(EventBase* evb,
- const std::string& ip,
- uint16_t port,
- uint32_t connectTimeout = 0);
-
- /**
- * Create a AsyncSocket from an already connected socket file descriptor.
- *
- * Note that while AsyncSocket enables TCP_NODELAY for sockets it creates
- * when connecting, it does not change the socket options when given an
- * existing file descriptor. If callers want TCP_NODELAY enabled when using
- * this version of the constructor, they need to explicitly call
- * setNoDelay(true) after the constructor returns.
- *
- * @param evb EventBase that will manage this socket.
- * @param fd File descriptor to take over (should be a connected socket).
- */
- AsyncSocket(EventBase* evb, int fd);
-
- /**
- * Helper function to create a shared_ptr<AsyncSocket>.
- *
- * This passes in the correct destructor object, since AsyncSocket's
- * destructor is protected and cannot be invoked directly.
- */
- static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb) {
- return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb),
- Destructor());
- }
-
- /**
- * Helper function to create a shared_ptr<AsyncSocket>.
- */
- static std::shared_ptr<AsyncSocket> newSocket(
- EventBase* evb,
- const folly::SocketAddress& address,
- uint32_t connectTimeout = 0) {
- return std::shared_ptr<AsyncSocket>(
- new AsyncSocket(evb, address, connectTimeout),
- Destructor());
- }
-
- /**
- * Helper function to create a shared_ptr<AsyncSocket>.
- */
- static std::shared_ptr<AsyncSocket> newSocket(
- EventBase* evb,
- const std::string& ip,
- uint16_t port,
- uint32_t connectTimeout = 0) {
- return std::shared_ptr<AsyncSocket>(
- new AsyncSocket(evb, ip, port, connectTimeout),
- Destructor());
- }
-
- /**
- * Helper function to create a shared_ptr<AsyncSocket>.
- */
- static std::shared_ptr<AsyncSocket> newSocket(EventBase* evb, int fd) {
- return std::shared_ptr<AsyncSocket>(new AsyncSocket(evb, fd),
- Destructor());
- }
-
- /**
- * Destroy the socket.
- *
- * AsyncSocket::destroy() must be called to destroy the socket.
- * The normal destructor is private, and should not be invoked directly.
- * This prevents callers from deleting a AsyncSocket while it is invoking a
- * callback.
- */
- virtual void destroy();
-
- /**
- * Get the EventBase used by this socket.
- */
- EventBase* getEventBase() const override {
- return eventBase_;
- }
-
- /**
- * Get the file descriptor used by the AsyncSocket.
- */
- virtual int getFd() const {
- return fd_;
- }
-
- /**
- * Extract the file descriptor from the AsyncSocket.
- *
- * This will immediately cause any installed callbacks to be invoked with an
- * error. The AsyncSocket may no longer be used after the file descriptor
- * has been extracted.
- *
- * Returns the file descriptor. The caller assumes ownership of the
- * descriptor, and it will not be closed when the AsyncSocket is destroyed.
- */
- virtual int detachFd();
-
- /**
- * Uniquely identifies a handle to a socket option value. Each
- * combination of level and option name corresponds to one socket
- * option value.
- */
- class OptionKey {
- public:
- bool operator<(const OptionKey& other) const {
- if (level == other.level) {
- return optname < other.optname;
- }
- return level < other.level;
- }
- int apply(int fd, int val) const {
- return setsockopt(fd, level, optname, &val, sizeof(val));
- }
- int level;
- int optname;
- };
-
- // Maps from a socket option key to its value
- typedef std::map<OptionKey, int> OptionMap;
-
- static const OptionMap emptyOptionMap;
- static const folly::SocketAddress anyAddress;
-
- /**
- * Initiate a connection.
- *
- * @param callback The callback to inform when the connection attempt
- * completes.
- * @param address The address to connect to.
- * @param timeout A timeout value, in milliseconds. If the connection
- * does not succeed within this period,
- * callback->connectError() will be invoked.
- */
- virtual void connect(ConnectCallback* callback,
- const folly::SocketAddress& address,
- int timeout = 0,
- const OptionMap &options = emptyOptionMap,
- const folly::SocketAddress& bindAddr = anyAddress
- ) noexcept;
- void connect(ConnectCallback* callback, const std::string& ip, uint16_t port,
- int timeout = 00,
- const OptionMap &options = emptyOptionMap) noexcept;
-
- /**
- * Set the send timeout.
- *
- * If write requests do not make any progress for more than the specified
- * number of milliseconds, fail all pending writes and close the socket.
- *
- * If write requests are currently pending when setSendTimeout() is called,
- * the timeout interval is immediately restarted using the new value.
- *
- * (See the comments for AsyncSocket for an explanation of why AsyncSocket
- * provides setSendTimeout() but not setRecvTimeout().)
- *
- * @param milliseconds The timeout duration, in milliseconds. If 0, no
- * timeout will be used.
- */
- void setSendTimeout(uint32_t milliseconds) override;
-
- /**
- * Get the send timeout.
- *
- * @return Returns the current send timeout, in milliseconds. A return value
- * of 0 indicates that no timeout is set.
- */
- uint32_t getSendTimeout() const override {
- return sendTimeout_;
- }
-
- /**
- * Set the maximum number of reads to execute from the underlying
- * socket each time the EventBase detects that new ingress data is
- * available. The default is unlimited, but callers can use this method
- * to limit the amount of data read from the socket per event loop
- * iteration.
- *
- * @param maxReads Maximum number of reads per data-available event;
- * a value of zero means unlimited.
- */
- void setMaxReadsPerEvent(uint16_t maxReads) {
- maxReadsPerEvent_ = maxReads;
- }
-
- /**
- * Get the maximum number of reads this object will execute from
- * the underlying socket each time the EventBase detects that new
- * ingress data is available.
- *
- * @returns Maximum number of reads per data-available event; a value
- * of zero means unlimited.
- */
- uint16_t getMaxReadsPerEvent() const {
- return maxReadsPerEvent_;
- }
-
- // Read and write methods
- void setReadCB(ReadCallback* callback);
- ReadCallback* getReadCallback() const;
-
- void write(WriteCallback* callback, const void* buf, size_t bytes,
- WriteFlags flags = WriteFlags::NONE);
- void writev(WriteCallback* callback, const iovec* vec, size_t count,
- WriteFlags flags = WriteFlags::NONE);
- void writeChain(WriteCallback* callback,
- std::unique_ptr<folly::IOBuf>&& buf,
- WriteFlags flags = WriteFlags::NONE);
-
- // Methods inherited from AsyncTransport
- void close() override;
- void closeNow() override;
- void closeWithReset() override;
- void shutdownWrite() override;
- void shutdownWriteNow() override;
-
- bool readable() const override;
- bool isPending() const override;
- virtual bool hangup() const;
- bool good() const override;
- bool error() const override;
- void attachEventBase(EventBase* eventBase) override;
- void detachEventBase() override;
- bool isDetachable() const override;
-
- void getLocalAddress(
- folly::SocketAddress* address) const override;
- void getPeerAddress(
- folly::SocketAddress* address) const override;
-
- bool isEorTrackingEnabled() const override { return false; }
-
- void setEorTracking(bool track) override {}
-
- bool connecting() const override {
- return (state_ == StateEnum::CONNECTING);
- }
-
- size_t getAppBytesWritten() const override {
- return appBytesWritten_;
- }
-
- size_t getRawBytesWritten() const override {
- return getAppBytesWritten();
- }
-
- size_t getAppBytesReceived() const override {
- return appBytesReceived_;
- }
-
- size_t getRawBytesReceived() const override {
- return getAppBytesReceived();
- }
-
- // Methods controlling socket options
-
- /**
- * Force writes to be transmitted immediately.
- *
- * This controls the TCP_NODELAY socket option. When enabled, TCP segments
- * are sent as soon as possible, even if it is not a full frame of data.
- * When disabled, the data may be buffered briefly to try and wait for a full
- * frame of data.
- *
- * By default, TCP_NODELAY is enabled for AsyncSocket objects.
- *
- * This method will fail if the socket is not currently open.
- *
- * @return Returns 0 if the TCP_NODELAY flag was successfully updated,
- * or a non-zero errno value on error.
- */
- int setNoDelay(bool noDelay);
-
- /*
- * Set the Flavor of Congestion Control to be used for this Socket
- * Please check '/lib/modules/<kernel>/kernel/net/ipv4' for tcp_*.ko
- * first to make sure the module is available for plugging in
- * Alternatively you can choose from net.ipv4.tcp_allowed_congestion_control
- */
- int setCongestionFlavor(const std::string &cname);
-
- /*
- * Forces ACKs to be sent immediately
- *
- * @return Returns 0 if the TCP_QUICKACK flag was successfully updated,
- * or a non-zero errno value on error.
- */
- int setQuickAck(bool quickack);
-
- /**
- * Set the send bufsize
- */
- int setSendBufSize(size_t bufsize);
-
- /**
- * Set the recv bufsize
- */
- int setRecvBufSize(size_t bufsize);
-
- /**
- * Sets a specific tcp personality
- * Available only on kernels 3.2 and greater
- */
- #define SO_SET_NAMESPACE 41
- int setTCPProfile(int profd);
-
-
- /**
- * Generic API for reading a socket option.
- *
- * @param level same as the "level" parameter in getsockopt().
- * @param optname same as the "optname" parameter in getsockopt().
- * @param optval pointer to the variable in which the option value should
- * be returned.
- * @return same as the return value of getsockopt().
- */
- template <typename T>
- int getSockOpt(int level, int optname, T *optval) {
- return getsockopt(fd_, level, optname, optval, sizeof(T));
- }
-
- /**
- * Generic API for setting a socket option.
- *
- * @param level same as the "level" parameter in getsockopt().
- * @param optname same as the "optname" parameter in getsockopt().
- * @param optval the option value to set.
- * @return same as the return value of setsockopt().
- */
- template <typename T>
- int setSockOpt(int level, int optname, const T *optval) {
- return setsockopt(fd_, level, optname, optval, sizeof(T));
- }
-
- protected:
- enum ReadResultEnum {
- READ_EOF = 0,
- READ_ERROR = -1,
- READ_BLOCKING = -2,
- };
-
- /**
- * Protected destructor.
- *
- * Users of AsyncSocket must never delete it directly. Instead, invoke
- * destroy() instead. (See the documentation in DelayedDestruction.h for
- * more details.)
- */
- ~AsyncSocket();
-
- enum class StateEnum : uint8_t {
- UNINIT,
- CONNECTING,
- ESTABLISHED,
- CLOSED,
- ERROR
- };
-
- friend std::ostream& operator << (std::ostream& os, const StateEnum& state);
-
- enum ShutdownFlags {
- /// shutdownWrite() called, but we are still waiting on writes to drain
- SHUT_WRITE_PENDING = 0x01,
- /// writes have been completely shut down
- SHUT_WRITE = 0x02,
- /**
- * Reads have been shutdown.
- *
- * At the moment we don't distinguish between remote read shutdown
- * (received EOF from the remote end) and local read shutdown. We can
- * only receive EOF when a read callback is set, and we immediately inform
- * it of the EOF. Therefore there doesn't seem to be any reason to have a
- * separate state of "received EOF but the local side may still want to
- * read".
- *
- * We also don't currently provide any API for only shutting down the read
- * side of a socket. (This is a no-op as far as TCP is concerned, anyway.)
- */
- SHUT_READ = 0x04,
- };
-
- class WriteRequest;
-
- class WriteTimeout : public AsyncTimeout {
- public:
- WriteTimeout(AsyncSocket* socket, EventBase* eventBase)
- : AsyncTimeout(eventBase)
- , socket_(socket) {}
-
- virtual void timeoutExpired() noexcept {
- socket_->timeoutExpired();
- }
-
- private:
- AsyncSocket* socket_;
- };
-
- class IoHandler : public EventHandler {
- public:
- IoHandler(AsyncSocket* socket, EventBase* eventBase)
- : EventHandler(eventBase, -1)
- , socket_(socket) {}
- IoHandler(AsyncSocket* socket, EventBase* eventBase, int fd)
- : EventHandler(eventBase, fd)
- , socket_(socket) {}
-
- virtual void handlerReady(uint16_t events) noexcept {
- socket_->ioReady(events);
- }
-
- private:
- AsyncSocket* socket_;
- };
-
- void init();
-
- // event notification methods
- void ioReady(uint16_t events) noexcept;
- virtual void checkForImmediateRead() noexcept;
- virtual void handleInitialReadWrite() noexcept;
- virtual void handleRead() noexcept;
- virtual void handleWrite() noexcept;
- virtual void handleConnect() noexcept;
- void timeoutExpired() noexcept;
-
- /**
- * Attempt to read from the socket.
- *
- * @param buf The buffer to read data into.
- * @param buflen The length of the buffer.
- *
- * @return Returns the number of bytes read, or READ_EOF on EOF, or
- * READ_ERROR on error, or READ_BLOCKING if the operation will
- * block.
- */
- virtual ssize_t performRead(void* buf, size_t buflen);
-
- /**
- * Populate an iovec array from an IOBuf and attempt to write it.
- *
- * @param callback Write completion/error callback.
- * @param vec Target iovec array; caller retains ownership.
- * @param count Number of IOBufs to write, beginning at start of buf.
- * @param buf Chain of iovecs.
- * @param flags set of flags for the underlying write calls, like cork
- */
- void writeChainImpl(WriteCallback* callback, iovec* vec,
- size_t count, std::unique_ptr<folly::IOBuf>&& buf,
- WriteFlags flags);
-
- /**
- * Write as much data as possible to the socket without blocking,
- * and queue up any leftover data to send when the socket can
- * handle writes again.
- *
- * @param callback The callback to invoke when the write is completed.
- * @param vec Array of buffers to write; this method will make a
- * copy of the vector (but not the buffers themselves)
- * if the write has to be completed asynchronously.
- * @param count Number of elements in vec.
- * @param buf The IOBuf that manages the buffers referenced by
- * vec, or a pointer to nullptr if the buffers are not
- * associated with an IOBuf. Note that ownership of
- * the IOBuf is transferred here; upon completion of
- * the write, the AsyncSocket deletes the IOBuf.
- * @param flags Set of write flags.
- */
- void writeImpl(WriteCallback* callback, const iovec* vec, size_t count,
- std::unique_ptr<folly::IOBuf>&& buf,
- WriteFlags flags = WriteFlags::NONE);
-
- /**
- * Attempt to write to the socket.
- *
- * @param vec The iovec array pointing to the buffers to write.
- * @param count The length of the iovec array.
- * @param flags Set of write flags.
- * @param countWritten On return, the value pointed to by this parameter
- * will contain the number of iovec entries that were
- * fully written.
- * @param partialWritten On return, the value pointed to by this parameter
- * will contain the number of bytes written in the
- * partially written iovec entry.
- *
- * @return Returns the total number of bytes written, or -1 on error. If no
- * data can be written immediately, 0 is returned.
- */
- virtual ssize_t performWrite(const iovec* vec, uint32_t count,
- WriteFlags flags, uint32_t* countWritten,
- uint32_t* partialWritten);
-
- bool updateEventRegistration();
-
- /**
- * Update event registration.
- *
- * @param enable Flags of events to enable. Set it to 0 if no events
- * need to be enabled in this call.
- * @param disable Flags of events
- * to disable. Set it to 0 if no events need to be disabled in this
- * call.
- *
- * @return true iff the update is successful.
- */
- bool updateEventRegistration(uint16_t enable, uint16_t disable);
-
- // Actually close the file descriptor and set it to -1 so we don't
- // accidentally close it again.
- void doClose();
-
- // error handling methods
- void startFail();
- void finishFail();
- void fail(const char* fn, const AsyncSocketException& ex);
- void failConnect(const char* fn, const AsyncSocketException& ex);
- void failRead(const char* fn, const AsyncSocketException& ex);
- void failWrite(const char* fn, WriteCallback* callback, size_t bytesWritten,
- const AsyncSocketException& ex);
- void failWrite(const char* fn, const AsyncSocketException& ex);
- void failAllWrites(const AsyncSocketException& ex);
- void invalidState(ConnectCallback* callback);
- void invalidState(ReadCallback* callback);
- void invalidState(WriteCallback* callback);
-
- std::string withAddr(const std::string& s);
-
- StateEnum state_; ///< StateEnum describing current state
- uint8_t shutdownFlags_; ///< Shutdown state (ShutdownFlags)
- uint16_t eventFlags_; ///< EventBase::HandlerFlags settings
- int fd_; ///< The socket file descriptor
- mutable
- folly::SocketAddress addr_; ///< The address we tried to connect to
- uint32_t sendTimeout_; ///< The send timeout, in milliseconds
- uint16_t maxReadsPerEvent_; ///< Max reads per event loop iteration
- EventBase* eventBase_; ///< The EventBase
- WriteTimeout writeTimeout_; ///< A timeout for connect and write
- IoHandler ioHandler_; ///< A EventHandler to monitor the fd
-
- ConnectCallback* connectCallback_; ///< ConnectCallback
- ReadCallback* readCallback_; ///< ReadCallback
- WriteRequest* writeReqHead_; ///< Chain of WriteRequests
- WriteRequest* writeReqTail_; ///< End of WriteRequest chain
- ShutdownSocketSet* shutdownSocketSet_;
- size_t appBytesReceived_; ///< Num of bytes received from socket
- size_t appBytesWritten_; ///< Num of bytes written to socket
-};
-
-
-} // folly
projects / folly.git / commitdiff