123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269 |
- /*
- * Copyright 2014-present 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/AsyncPipe.h>
- #include <folly/FileUtil.h>
- #include <folly/io/async/AsyncSocketException.h>
- using folly::IOBuf;
- using folly::IOBufQueue;
- using std::string;
- using std::unique_ptr;
- namespace folly {
- AsyncPipeReader::~AsyncPipeReader() {
- close();
- }
- void AsyncPipeReader::failRead(const AsyncSocketException& ex) {
- VLOG(5) << "AsyncPipeReader(this=" << this << ", fd=" << fd_
- << "): failed while reading: " << ex.what();
- DCHECK(readCallback_ != nullptr);
- AsyncReader::ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readErr(ex);
- close();
- }
- void AsyncPipeReader::close() {
- unregisterHandler();
- if (fd_ >= 0) {
- changeHandlerFD(-1);
- if (closeCb_) {
- closeCb_(fd_);
- } else {
- ::close(fd_);
- }
- fd_ = -1;
- }
- }
- void AsyncPipeReader::handlerReady(uint16_t events) noexcept {
- DestructorGuard dg(this);
- CHECK(events & EventHandler::READ);
- VLOG(5) << "AsyncPipeReader::handlerReady() this=" << this << ", fd=" << fd_;
- assert(readCallback_ != nullptr);
- while (readCallback_) {
- // - What API does callback support?
- const auto movable = readCallback_->isBufferMovable(); // noexcept
- // Get the buffer to read into.
- void* buf = nullptr;
- size_t buflen = 0;
- std::unique_ptr<IOBuf> ioBuf;
- if (movable) {
- ioBuf = IOBuf::create(readCallback_->maxBufferSize());
- buf = ioBuf->writableBuffer();
- buflen = ioBuf->capacity();
- } else {
- try {
- readCallback_->getReadBuffer(&buf, &buflen);
- } catch (const std::exception& ex) {
- AsyncSocketException aex(
- AsyncSocketException::BAD_ARGS,
- string("ReadCallback::getReadBuffer() "
- "threw exception: ") +
- ex.what());
- failRead(aex);
- return;
- } catch (...) {
- AsyncSocketException aex(
- AsyncSocketException::BAD_ARGS,
- string("ReadCallback::getReadBuffer() "
- "threw non-exception type"));
- failRead(aex);
- return;
- }
- if (buf == nullptr || buflen == 0) {
- AsyncSocketException aex(
- AsyncSocketException::INVALID_STATE,
- string("ReadCallback::getReadBuffer() "
- "returned empty buffer"));
- failRead(aex);
- return;
- }
- }
- // Perform the read
- ssize_t bytesRead = folly::readNoInt(fd_, buf, buflen);
- if (bytesRead > 0) {
- if (movable) {
- ioBuf->append(std::size_t(bytesRead));
- readCallback_->readBufferAvailable(std::move(ioBuf));
- } else {
- readCallback_->readDataAvailable(size_t(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 (static_cast<size_t>(bytesRead) < buflen) {
- return;
- }
- } else if (bytesRead < 0 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
- // No more data to read right now.
- return;
- } else if (bytesRead < 0) {
- AsyncSocketException ex(
- AsyncSocketException::INVALID_STATE, "read failed", errno);
- failRead(ex);
- return;
- } else {
- assert(bytesRead == 0);
- // EOF
- unregisterHandler();
- AsyncReader::ReadCallback* callback = readCallback_;
- readCallback_ = nullptr;
- callback->readEOF();
- return;
- }
- // Max reads per loop?
- }
- }
- void AsyncPipeWriter::write(
- unique_ptr<folly::IOBuf> buf,
- AsyncWriter::WriteCallback* callback) {
- if (closed()) {
- if (callback) {
- AsyncSocketException ex(
- AsyncSocketException::NOT_OPEN, "attempt to write to closed pipe");
- callback->writeErr(0, ex);
- }
- return;
- }
- bool wasEmpty = (queue_.empty());
- folly::IOBufQueue iobq;
- iobq.append(std::move(buf));
- std::pair<folly::IOBufQueue, AsyncWriter::WriteCallback*> p(
- std::move(iobq), callback);
- queue_.emplace_back(std::move(p));
- if (wasEmpty) {
- handleWrite();
- } else {
- CHECK(!queue_.empty());
- CHECK(isHandlerRegistered());
- }
- }
- void AsyncPipeWriter::writeChain(
- folly::AsyncWriter::WriteCallback* callback,
- std::unique_ptr<folly::IOBuf>&& buf,
- WriteFlags) {
- write(std::move(buf), callback);
- }
- void AsyncPipeWriter::closeOnEmpty() {
- VLOG(5) << "close on empty";
- if (queue_.empty()) {
- closeNow();
- } else {
- closeOnEmpty_ = true;
- CHECK(isHandlerRegistered());
- }
- }
- void AsyncPipeWriter::closeNow() {
- VLOG(5) << "close now";
- if (!queue_.empty()) {
- failAllWrites(AsyncSocketException(
- AsyncSocketException::NOT_OPEN, "closed with pending writes"));
- }
- if (fd_ >= 0) {
- unregisterHandler();
- changeHandlerFD(-1);
- if (closeCb_) {
- closeCb_(fd_);
- } else {
- close(fd_);
- }
- fd_ = -1;
- }
- }
- void AsyncPipeWriter::failAllWrites(const AsyncSocketException& ex) {
- DestructorGuard dg(this);
- while (!queue_.empty()) {
- // the first entry of the queue could have had a partial write, but needs to
- // be tracked.
- if (queue_.front().second) {
- queue_.front().second->writeErr(0, ex);
- }
- queue_.pop_front();
- }
- }
- void AsyncPipeWriter::handlerReady(uint16_t events) noexcept {
- CHECK(events & EventHandler::WRITE);
- handleWrite();
- }
- void AsyncPipeWriter::handleWrite() {
- DestructorGuard dg(this);
- assert(!queue_.empty());
- do {
- auto& front = queue_.front();
- folly::IOBufQueue& curQueue = front.first;
- DCHECK(!curQueue.empty());
- // someday, support writev. The logic for partial writes is a bit complex
- const IOBuf* head = curQueue.front();
- CHECK(head->length());
- ssize_t rc = folly::writeNoInt(fd_, head->data(), head->length());
- if (rc < 0) {
- if (errno == EAGAIN || errno == EWOULDBLOCK) {
- // pipe is full
- VLOG(5) << "write blocked";
- registerHandler(EventHandler::WRITE);
- return;
- } else {
- failAllWrites(AsyncSocketException(
- AsyncSocketException::INTERNAL_ERROR, "write failed", errno));
- closeNow();
- return;
- }
- } else if (rc == 0) {
- registerHandler(EventHandler::WRITE);
- return;
- }
- curQueue.trimStart(size_t(rc));
- if (curQueue.empty()) {
- auto cb = front.second;
- queue_.pop_front();
- if (cb) {
- cb->writeSuccess();
- }
- } else {
- VLOG(5) << "partial write blocked";
- }
- } while (!queue_.empty());
- if (closeOnEmpty_) {
- closeNow();
- } else {
- unregisterHandler();
- }
- }
- } // namespace folly
|