123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263 |
- // Copyright (c) 2012 The Chromium Authors. All rights reserved.
- // Use of this source code is governed by a BSD-style license that can be
- // found in the LICENSE file.
- #include "tools/android/forwarder2/forwarder.h"
- #include <utility>
- #include "base/check.h"
- #include "base/posix/eintr_wrapper.h"
- #include "tools/android/forwarder2/socket.h"
- namespace forwarder2 {
- namespace {
- const int kBufferSize = 32 * 1024;
- } // namespace
- // Helper class to buffer reads and writes from one socket to another.
- // Each implements a small buffer connected two one input socket, and
- // one output socket.
- //
- // socket_from_ ---> [BufferedCopier] ---> socket_to_
- //
- // These objects are used in a pair to handle duplex traffic, as in:
- //
- // -------> [BufferedCopier_1] --->
- // | |
- // socket_1 * * socket_2
- // | |
- // <------ [BufferedCopier_2] <----
- //
- // When a BufferedCopier is in the READING state (see below), it only listens
- // to events on its input socket, and won't detect when its output socket
- // disconnects. To work around this, its peer will call its Close() method
- // when that happens.
- class Forwarder::BufferedCopier {
- public:
- // Possible states:
- // READING - Empty buffer and Waiting for input.
- // WRITING - Data in buffer, and waiting for output.
- // CLOSING - Like WRITING, but do not try to read after that.
- // CLOSED - Completely closed.
- //
- // State transitions are:
- //
- // T01: READING ---[receive data]---> WRITING
- // T02: READING ---[error on input socket]---> CLOSED
- // T03: READING ---[Close() call]---> CLOSED
- //
- // T04: WRITING ---[write partial data]---> WRITING
- // T05: WRITING ---[write all data]----> READING
- // T06: WRITING ---[error on output socket]----> CLOSED
- // T07: WRITING ---[Close() call]---> CLOSING
- //
- // T08: CLOSING ---[write partial data]---> CLOSING
- // T09: CLOSING ---[write all data]----> CLOSED
- // T10: CLOSING ---[Close() call]---> CLOSING
- // T11: CLOSING ---[error on output socket] ---> CLOSED
- //
- enum State {
- STATE_READING = 0,
- STATE_WRITING = 1,
- STATE_CLOSING = 2,
- STATE_CLOSED = 3,
- };
- // Does NOT own the pointers.
- BufferedCopier(Socket* socket_from, Socket* socket_to)
- : socket_from_(socket_from),
- socket_to_(socket_to),
- bytes_read_(0),
- write_offset_(0),
- peer_(NULL),
- state_(STATE_READING) {}
- BufferedCopier(const BufferedCopier&) = delete;
- BufferedCopier& operator=(const BufferedCopier&) = delete;
- // Sets the 'peer_' field pointing to the other BufferedCopier in a pair.
- void SetPeer(BufferedCopier* peer) {
- DCHECK(!peer_);
- peer_ = peer;
- }
- bool is_closed() const { return state_ == STATE_CLOSED; }
- // Gently asks to close a buffer. Called either by the peer or the forwarder.
- void Close() {
- switch (state_) {
- case STATE_READING:
- state_ = STATE_CLOSED; // T03
- break;
- case STATE_WRITING:
- state_ = STATE_CLOSING; // T07
- break;
- case STATE_CLOSING:
- break; // T10
- case STATE_CLOSED:
- break;
- }
- }
- // Call this before select(). This updates |read_fds|,
- // |write_fds| and |max_fd| appropriately *if* the buffer isn't closed.
- void PrepareSelect(fd_set* read_fds, fd_set* write_fds, int* max_fd) {
- int fd;
- switch (state_) {
- case STATE_READING:
- DCHECK(bytes_read_ == 0);
- DCHECK(write_offset_ == 0);
- fd = socket_from_->fd();
- if (fd < 0) {
- ForceClose(); // T02
- return;
- }
- FD_SET(fd, read_fds);
- break;
- case STATE_WRITING:
- case STATE_CLOSING:
- DCHECK(bytes_read_ > 0);
- DCHECK(write_offset_ < bytes_read_);
- fd = socket_to_->fd();
- if (fd < 0) {
- ForceClose(); // T06
- return;
- }
- FD_SET(fd, write_fds);
- break;
- case STATE_CLOSED:
- return;
- }
- *max_fd = std::max(*max_fd, fd);
- }
- // Call this after a select() call to operate over the buffer.
- void ProcessSelect(const fd_set& read_fds, const fd_set& write_fds) {
- int fd;
- int ret;
- // With FORTIFY_SOURCE, FD_ISSET is implemented as a function that takes a
- // non-const fd_set*. Make a copy of the passed arguments so we can safely
- // take a reference.
- fd_set read_fds_copy = read_fds;
- fd_set write_fds_copy = write_fds;
- switch (state_) {
- case STATE_READING:
- fd = socket_from_->fd();
- if (fd < 0) {
- state_ = STATE_CLOSED; // T02
- return;
- }
- if (!FD_ISSET(fd, &read_fds_copy))
- return;
- ret = socket_from_->NonBlockingRead(buffer_, kBufferSize);
- if (ret <= 0) {
- ForceClose(); // T02
- return;
- }
- bytes_read_ = ret;
- write_offset_ = 0;
- state_ = STATE_WRITING; // T01
- break;
- case STATE_WRITING:
- case STATE_CLOSING:
- fd = socket_to_->fd();
- if (fd < 0) {
- ForceClose(); // T06 + T11
- return;
- }
- if (!FD_ISSET(fd, &write_fds_copy))
- return;
- ret = socket_to_->NonBlockingWrite(buffer_ + write_offset_,
- bytes_read_ - write_offset_);
- if (ret <= 0) {
- ForceClose(); // T06 + T11
- return;
- }
- write_offset_ += ret;
- if (write_offset_ < bytes_read_)
- return; // T08 + T04
- write_offset_ = 0;
- bytes_read_ = 0;
- if (state_ == STATE_CLOSING) {
- ForceClose(); // T09
- return;
- }
- state_ = STATE_READING; // T05
- break;
- case STATE_CLOSED:
- break;
- }
- }
- private:
- // Internal method used to close the buffer and notify the peer, if any.
- void ForceClose() {
- if (peer_) {
- peer_->Close();
- peer_ = NULL;
- }
- state_ = STATE_CLOSED;
- }
- // Not owned.
- Socket* socket_from_;
- Socket* socket_to_;
- int bytes_read_;
- int write_offset_;
- BufferedCopier* peer_;
- State state_;
- char buffer_[kBufferSize];
- };
- Forwarder::Forwarder(std::unique_ptr<Socket> socket1,
- std::unique_ptr<Socket> socket2)
- : socket1_(std::move(socket1)),
- socket2_(std::move(socket2)),
- buffer1_(new BufferedCopier(socket1_.get(), socket2_.get())),
- buffer2_(new BufferedCopier(socket2_.get(), socket1_.get())) {
- buffer1_->SetPeer(buffer2_.get());
- buffer2_->SetPeer(buffer1_.get());
- }
- Forwarder::~Forwarder() {
- DCHECK(thread_checker_.CalledOnValidThread());
- }
- void Forwarder::RegisterFDs(fd_set* read_fds, fd_set* write_fds, int* max_fd) {
- DCHECK(thread_checker_.CalledOnValidThread());
- buffer1_->PrepareSelect(read_fds, write_fds, max_fd);
- buffer2_->PrepareSelect(read_fds, write_fds, max_fd);
- }
- void Forwarder::ProcessEvents(const fd_set& read_fds, const fd_set& write_fds) {
- DCHECK(thread_checker_.CalledOnValidThread());
- buffer1_->ProcessSelect(read_fds, write_fds);
- buffer2_->ProcessSelect(read_fds, write_fds);
- }
- bool Forwarder::IsClosed() const {
- DCHECK(thread_checker_.CalledOnValidThread());
- return buffer1_->is_closed() && buffer2_->is_closed();
- }
- void Forwarder::Shutdown() {
- DCHECK(thread_checker_.CalledOnValidThread());
- buffer1_->Close();
- buffer2_->Close();
- }
- } // namespace forwarder2
|