Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #include "io/tcp_session.h"
       6             : 
       7             : #include <algorithm>
       8             : #include <string>
       9             : 
      10             : #include <boost/asio.hpp>
      11             : #include <boost/asio/detail/socket_option.hpp>
      12             : #include <boost/bind/bind.hpp>
      13             : #include <boost/scoped_array.hpp>
      14             : #include <boost/asio/detail/recycling_allocator.hpp>
      15             : 
      16             : #include "base/logging.h"
      17             : #include "base/address_util.h"
      18             : #include "io/event_manager.h"
      19             : #include "io/io_log.h"
      20             : #include "io/io_utils.h"
      21             : #include "io/tcp_message_write.h"
      22             : #include "io/tcp_server.h"
      23             : #include "base/address_util.h"
      24             : 
      25             : using boost::asio::async_write;
      26             : using boost::asio::buffer;
      27             : using boost::asio::buffer_cast;
      28             : using boost::asio::detail::socket_option::integer;
      29             : using boost::asio::const_buffer;
      30             : using boost::asio::mutable_buffer;
      31             : using boost::asio::mutable_buffers_1;
      32             : using boost::asio::null_buffers;
      33             : using boost::asio::socket_base;
      34             : using boost::bind;
      35             : using boost::function;
      36             : using boost::scoped_array;
      37             : using boost::system::error_code;
      38             : using std::min;
      39             : using std::ostringstream;
      40             : using std::string;
      41             : using namespace boost::placeholders;
      42             : 
      43             : using boost::asio::error::try_again;
      44             : using boost::asio::error::would_block;
      45             : using boost::asio::error::in_progress;
      46             : using boost::asio::error::interrupted;
      47             : using boost::asio::error::network_down;
      48             : using boost::asio::error::network_reset;
      49             : using boost::asio::error::network_unreachable;
      50             : using boost::asio::error::no_buffer_space;
      51             : using boost::asio::placeholders::error;
      52             : using boost::asio::placeholders::bytes_transferred;
      53             : using boost::asio::ip::tcp;
      54             : 
      55             : int TcpSession::reader_task_id_ = -1;
      56             : 
      57             : class TcpSession::Reader : public Task {
      58             : public:
      59             :     typedef function<void(Buffer)> ReadHandler;
      60             : 
      61        4960 :     Reader(TcpSessionPtr session, ReadHandler read_fn, Buffer buffer)
      62       19840 :         : Task(session->reader_task_id(), session->GetSessionInstance()),
      63        4960 :           session_(session), read_fn_(read_fn), buffer_(buffer) {
      64        4960 :     }
      65        4956 :     virtual bool Run() {
      66        4956 :         if (session_->IsEstablished()) {
      67        4824 :             read_fn_(buffer_);
      68        4827 :             if (session_->IsReaderDeferred()) {
      69             :                 // Update socket read block count.
      70           1 :                 session_->stats_.read_block_start_time = UTCTimestampUsec();
      71           1 :                 session_->stats_.read_blocked++;
      72           1 :                 session_->server_->stats_.read_blocked++;
      73             :             } else {
      74        4825 :                 session_->AsyncReadStart();
      75             :             }
      76             :         }
      77        4957 :         return true;
      78             :     }
      79           0 :     string Description() const { return "TcpSession::Reader"; }
      80             : 
      81             : private:
      82             :     TcpSessionPtr session_;
      83             :     ReadHandler read_fn_;
      84             :     Buffer buffer_;
      85             : };
      86             : 
      87        2356 : TcpSession::TcpSession(
      88             :     TcpServer *server, Socket *socket, bool async_read_ready,
      89        2356 :     size_t buffer_send_size)
      90        2353 :     : server_(server),
      91        2358 :       socket_(socket),
      92        2349 :       read_on_connect_(async_read_ready),
      93        2349 :       established_(false),
      94        2349 :       closed_(false),
      95        2347 :       direction_(ACTIVE),
      96        2336 :       writer_(new TcpMessageWriter(this, buffer_send_size)),
      97        7032 :       name_("-") {
      98        2319 :     refcount_ = 0;
      99        2362 :     if (reader_task_id_ == -1) {
     100          60 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     101          60 :         reader_task_id_ = scheduler->GetTaskId("io::ReaderTask");
     102             :     }
     103        2362 :     if (server_) {
     104        2345 :         io_strand_.reset(new Strand(server->event_manager()->io_service()->get_executor()));
     105             :     }
     106        2361 :     defer_reader_ = false;
     107        2360 :     write_blocked_ = false;
     108        2361 :     tcp_close_in_progress_ = false;
     109        2361 : }
     110             : 
     111        2360 : TcpSession::~TcpSession() {
     112        2360 :     assert(!established_);
     113        2360 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
     114        7060 :          iter != buffer_queue_.end(); ++iter) {
     115        4700 :         DeleteBuffer(*iter);
     116             :     }
     117        2360 :     buffer_queue_.clear();
     118        2361 : }
     119             : 
     120        9975 : mutable_buffer TcpSession::AllocateBuffer(size_t buffer_size) {
     121        9975 :     uint8_t *data = new uint8_t[buffer_size];
     122        9975 :     mutable_buffer buffer = mutable_buffer(data, buffer_size);
     123        9975 :     buffer_queue_.push_back(buffer);
     124        9975 :     return buffer;
     125             : }
     126             : 
     127        9968 : void TcpSession::DeleteBuffer(mutable_buffer buffer) {
     128        9968 :     uint8_t *data = buffer_cast<uint8_t *>(buffer);
     129        9967 :     delete[] data;
     130        9975 : }
     131             : 
     132        5806 : static int BufferCmp(const mutable_buffer &lhs, const const_buffer &rhs) {
     133        5806 :     const uint8_t *lp = buffer_cast<uint8_t *>(lhs);
     134        5806 :     const uint8_t *rp = buffer_cast<const uint8_t *>(rhs);
     135        5806 :     if (lp < rp) {
     136         504 :         return -1;
     137             :     }
     138        5302 :     if (lp > rp) {
     139          33 :         return 1;
     140             :     }
     141        5269 :     return 0;
     142             : }
     143             : 
     144        4467 : void TcpSession::ReleaseBuffer(Buffer buffer) {
     145        4467 :     std::scoped_lock lock(mutex_);
     146        4468 :     ReleaseBufferLocked(buffer);
     147        4469 : }
     148             : 
     149        5272 : void TcpSession::ReleaseBufferLocked(Buffer buffer) {
     150        5272 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
     151        5809 :          iter != buffer_queue_.end(); ++iter) {
     152        5808 :         if (BufferCmp(*iter, buffer) == 0) {
     153        5269 :             DeleteBuffer(*iter);
     154        5275 :             buffer_queue_.erase(iter);
     155        5275 :             return;
     156             :         }
     157             :     }
     158           0 :     assert(false);
     159             : }
     160             : 
     161       11093 : void TcpSession::AsyncReadStartInternal(TcpSessionPtr session) {
     162             :     // Update socket read block time.
     163       11093 :     if (stats_.read_block_start_time) {
     164           1 :         uint64_t blocked_usecs = UTCTimestampUsec() -
     165           1 :             stats_.read_block_start_time;
     166           1 :         stats_.read_block_start_time = 0;
     167           1 :         stats_.read_blocked_duration_usecs += blocked_usecs;
     168           1 :         server_->stats_.read_blocked_duration_usecs += blocked_usecs;
     169             :     }
     170             : 
     171       11093 :     std::scoped_lock lock(mutex_);
     172       11093 :     AsyncReadSome();
     173       11093 : }
     174             : 
     175       11095 : void TcpSession::AsyncReadStart() {
     176       11095 :     if (io_strand_) {
     177       11095 :         boost::asio::detail::recycling_allocator<void> allocator;
     178       22191 :         io_strand_->post(bind(&TcpSession::AsyncReadStartInternal, this,
     179       22175 :                          TcpSessionPtr(this)), allocator);
     180             :     }
     181       11093 : }
     182             : 
     183           2 : void TcpSession::SetDeferReader(bool defer_reader) {
     184           2 :     if (defer_reader_ != defer_reader) {
     185           2 :         defer_reader_ = defer_reader;
     186             :         // Call AsyncReadStart if reader was previously deferred
     187           2 :         if (!defer_reader_) {
     188           1 :             AsyncReadStart();
     189             :         }
     190             :     }
     191           2 : }
     192             : 
     193        7424 : void TcpSession::AsyncReadSome() {
     194        7424 :     if (IsEstablishedLocked()) {
     195       14720 :         socket()->async_read_some(null_buffers(),
     196       14720 :             bind(&TcpSession::AsyncReadHandler, TcpSessionPtr(this)));
     197             :     }
     198        7424 : }
     199             : 
     200        7123 : void TcpSession::AsyncWrite(const uint8_t *data, std::size_t size) {
     201        7123 :     async_write(*socket(), buffer(data, size),
     202       14246 :         bind(&TcpSession::AsyncWriteHandler, TcpSessionPtr(this),
     203             :              error, bytes_transferred));
     204        7123 : }
     205             : 
     206      160727 : TcpSession::Endpoint TcpSession::local_endpoint() const {
     207      160727 :     std::scoped_lock lock(mutex_);
     208      160727 :     if (!established_)
     209           0 :         return Endpoint();
     210             : 
     211      160727 :     error_code error;
     212      160727 :     Endpoint local = socket()->local_endpoint(error);
     213      160727 :     if (error) {
     214           0 :         return Endpoint();
     215             :     }
     216      160727 :     return local;
     217      160727 : }
     218             : 
     219        5003 : void TcpSession::set_observer(EventObserver observer) {
     220        5003 :     std::scoped_lock lock(obs_mutex_);
     221        5013 :     observer_ = observer;
     222        4995 : }
     223             : 
     224        2199 : void TcpSession::SetName() {
     225        2199 :     ostringstream out;
     226        2199 :     error_code error;
     227        2199 :     Endpoint local;
     228             : 
     229        2199 :     local = socket()->local_endpoint(error);
     230        2199 :     out << local.address().to_string() << ":" << local.port() << "::";
     231        2199 :     out << remote_.address().to_string() << ":" << remote_.port();
     232             : 
     233        2199 :     name_ = out.str();
     234             : 
     235        2199 :     out.str("");
     236        2199 :     std::string hostname = "";
     237        2199 :     if (local.address().is_v4()) {
     238        2196 :         hostname = ResolveCanonicalName(local.address().to_string());
     239             :     } else {
     240           3 :         hostname = ResolveCanonicalNameIPv6(local.address().to_string());
     241             :     }
     242        2199 :     out << hostname << ":" << remote_.address().to_string();
     243        2199 :     uve_key_str_ = out.str();
     244        2199 : }
     245             : 
     246        2199 : void TcpSession::SessionEstablished(Endpoint remote,
     247             :                                     Direction direction) {
     248        2199 :     established_ = true;
     249        2199 :     remote_ = remote;
     250        2199 :     remote_addr_str_ = remote.address().to_string();
     251        2199 :     direction_ = direction;
     252        2199 :     SetName();
     253        2199 : }
     254             : 
     255        1035 : void TcpSession::Accepted() {
     256        1035 :     TCP_SESSION_LOG_DEBUG(this, TCP_DIR_OUT,
     257             :                           "Passive session Accept complete");
     258             :     {
     259        1035 :         std::scoped_lock obs_lock(obs_mutex_);
     260        1035 :         if (observer_) {
     261          35 :             observer_(this, ACCEPT);
     262             :         }
     263        1035 :     }
     264             : 
     265        1035 :     if (read_on_connect_) {
     266         214 :         AsyncReadStart();
     267             :     }
     268        1035 : }
     269             : 
     270        1122 : bool TcpSession::Connected(Endpoint remote) {
     271        1122 :     assert(refcount_);
     272             : 
     273             :     {
     274        1122 :         std::scoped_lock lock(mutex_);
     275        1122 :         if (closed_) {
     276           0 :             return false;
     277             :         }
     278        1122 :         SessionEstablished(remote, TcpSession::ACTIVE);
     279        1122 :     }
     280        1122 :     SetSocketOptions();
     281             : 
     282        1122 :     TCP_SESSION_LOG_DEBUG(this, TCP_DIR_IN,
     283             :                           "Active session connection complete");
     284             : 
     285             :     {
     286        1122 :         std::scoped_lock obs_lock(obs_mutex_);
     287        1122 :         if (observer_) {
     288        1122 :             observer_(this, CONNECT_COMPLETE);
     289             :         }
     290        1122 :     }
     291             : 
     292        1122 :     if (read_on_connect_) {
     293        1122 :         AsyncReadStart();
     294             :     }
     295        1122 :     return true;
     296             : }
     297             : 
     298          49 : void TcpSession::ConnectFailed() {
     299          49 :     std::scoped_lock obs_lock(obs_mutex_);
     300          49 :     if (observer_) {
     301          48 :         observer_(this, CONNECT_FAILED);
     302             :     }
     303          49 : }
     304             : 
     305             : // Requires: lock must not be held
     306        2371 : void TcpSession::CloseInternal(const error_code &ec,
     307             :                                bool call_observer, bool notify_server) {
     308        2371 :     std::unique_lock<std::mutex> lock(mutex_);
     309             : 
     310        2371 :     if (socket() != NULL && !closed_) {
     311        2356 :         error_code error;
     312        2356 :         socket()->shutdown(tcp::socket::shutdown_both, error);
     313        2357 :         if (error) {
     314         448 :             TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     315             :                 "Shutdown failed due to error: " << error.message());
     316             :         }
     317        2356 :         socket()->close(error);
     318             :     }
     319        2373 :     closed_ = true;
     320        2373 :     tcp_close_in_progress_ = false;
     321             : 
     322        2372 :     if (!established_) {
     323         174 :         return;
     324             :     }
     325        2198 :     established_ = false;
     326             : 
     327             :     // copy the ec to close reason
     328        2198 :     close_reason_ = ec;
     329             : 
     330             :     // Take a reference through intrusive pointer to protect session from
     331             :     // possibly getting deleted from another thread.
     332        2198 :     TcpSessionPtr session = TcpSessionPtr(this);
     333        2198 :     lock.unlock();
     334             : 
     335        2199 :     if (call_observer) {
     336        1348 :         std::scoped_lock obs_lock(obs_mutex_);
     337        1348 :         if (observer_) {
     338         512 :             observer_(this, CLOSE);
     339             :         }
     340        1348 :     }
     341             : 
     342        2199 :     if (notify_server) {
     343        2156 :         server_->OnSessionClose(this);
     344             :     }
     345        2373 : }
     346             : 
     347        3653 : void TcpSession::TriggerAsyncReadHandler() {
     348        3653 :     if (io_strand_) {
     349        3653 :         boost::asio::detail::recycling_allocator<void> allocator;
     350        7306 :         io_strand_->post(bind(&TcpSession::AsyncReadHandler,
     351        7306 :                               TcpSessionPtr(this)), allocator);
     352             :     }
     353        3653 : }
     354             : 
     355        5259 : void TcpSession::Close() {
     356        5259 :     std::unique_lock<std::mutex> lock(mutex_);
     357             : 
     358             :     // Close can be called by application during cleanup. At this time
     359             :     // session may be already closed due to error and there may be write
     360             :     // data in the buffer, ignore if socket is closed.
     361        5264 :     if (closed_) {
     362        3253 :         return;
     363             :     }
     364             : 
     365        2011 :     if (server_ && writer_->IsWritePending()) {
     366        1032 :         tcp_close_in_progress_ = true;
     367        1035 :         return;
     368             :     }
     369         974 :     lock.unlock();
     370             : 
     371         978 :     error_code ec;
     372         978 :     CloseInternal(ec, false);
     373        5267 : }
     374             : 
     375             : // virtual method overriden in derrived classes.
     376           2 : void TcpSession::WriteReady(const error_code &error) {
     377           2 : }
     378             : 
     379        9784 : void TcpSession::AsyncWriteHandler(TcpSessionPtr session,
     380             :                                    const error_code &error,
     381             :                                    std::size_t wrote) {
     382        9784 :     std::unique_lock<std::mutex> lock(session->mutex_);
     383        9784 :     if (session->IsSocketErrorHard(error)) {
     384           0 :         lock.unlock();
     385           0 :         TCP_SESSION_LOG_ERROR(session, TCP_DIR_OUT,
     386             :                               "Write failed due to error: " << error.message());
     387           0 :         session->CloseInternal(error, true);
     388           0 :         return;
     389             :     }
     390             : 
     391             :     //
     392             :     // Ignore if connection is already closed.
     393             :     //
     394        9784 :     if (session->IsClosedLocked()) return;
     395             : 
     396             :     // Update socket write bytes statistics.
     397        9783 :     session->stats_.write_bytes += wrote;
     398        9783 :     session->server_->stats_.write_bytes += wrote;
     399             : 
     400        9783 :     bool send_ready = false;
     401        9783 :     bool more_write = session->writer_->UpdateBufferQueue(wrote, &send_ready);
     402             : 
     403             :     // Subsequent write
     404        9783 :     if (more_write) {
     405        2657 :         session->writer_->TriggerAsyncWrite();
     406        7126 :     } else if (session->tcp_close_in_progress_) {
     407         544 :         lock.unlock();
     408         544 :         session->CloseInternal(error, true);
     409         544 :         return;
     410             :     }
     411             : 
     412        9239 :     lock.unlock();
     413        9239 :     if (send_ready)
     414           4 :         session->WriteReady(error);
     415        9239 :     return;
     416        9784 : }
     417             : 
     418        7128 : void TcpSession::AsyncWriteInternal(TcpSessionPtr session) {
     419             : 
     420        7128 :     std::scoped_lock lock(session->mutex_);
     421             : 
     422             :     //
     423             :     // Ignore if connection is already closed.
     424             :     //
     425        7128 :     if (session->IsClosedLocked()) return;
     426        7127 :     session->writer_->TriggerAsyncWrite();
     427        7128 : }
     428             : 
     429        7663 : bool TcpSession::Send(const uint8_t *data, size_t size, size_t *sent) {
     430        7663 :     bool ret = true;
     431        7663 :     std::unique_lock<std::mutex> lock(mutex_);
     432             : 
     433             :     // Reset sent, if provided.
     434        7676 :     if (sent) *sent = 0;
     435             : 
     436             :     //
     437             :     // If the session closed in the mean while, bail out
     438             :     // If session close is triggered, but close in progress, bail out
     439             :     //
     440        7676 :     if (!IsEstablishedLocked()) return false;
     441             : 
     442        7526 :     if (socket()->non_blocking()) {
     443        7520 :         error_code error;
     444        7520 :         int len = writer_->AsyncSend(data, size, &error);
     445        7530 :         lock.unlock();
     446        7550 :         if (len < 0) {
     447           0 :             TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     448             :                                   "Write failed due to error: "
     449             :                                   << error.category().name() << " "
     450             :                                   << error.message());
     451           0 :             CloseInternal(error, true);
     452           0 :             return false;
     453             :         }
     454        7550 :         if ((size_t) len != size)
     455          14 :             ret = false;
     456        7550 :         if (sent) *sent = (len > 0) ? len : 0;
     457             :     }
     458        7550 :     return ret;
     459        7687 : }
     460             : 
     461        4960 : Task* TcpSession::CreateReaderTask(mutable_buffer buffer,
     462             :                                   size_t bytes_transferred) {
     463        4960 :     Buffer rdbuf(buffer_cast<const uint8_t *>(buffer), bytes_transferred);
     464        4960 :     Reader *task = new Reader(TcpSessionPtr(this),
     465        4960 :                               bind(&TcpSession::OnRead, this, _1), rdbuf);
     466        4960 :     return (task);
     467             : }
     468             : 
     469        5995 : size_t TcpSession::ReadSome(mutable_buffer buffer, error_code *error) {
     470        5995 :     return socket()->read_some(mutable_buffers_1(buffer), *error);
     471             : }
     472             : 
     473             : // Tests with large data have shown large amounts of data being read in one
     474             : // read_some() call, if available. Hence, allocate memory for all the bytes
     475             : // available in the socket, but no less t han kDefaultBufferSize.
     476        5726 : size_t TcpSession::GetReadBufferSize() const {
     477        5726 :     error_code error;
     478        5726 :     size_t size = socket_->available(error);
     479        5726 :     if (size < kDefaultBufferSize)
     480        5435 :         size = kDefaultBufferSize;
     481        5726 :     return size;
     482             : }
     483             : 
     484       10961 : void TcpSession::AsyncReadHandler(TcpSessionPtr session) {
     485       10961 :     std::unique_lock<std::mutex> lock(session->mutex_);
     486       10961 :     if (session->closed_) {
     487         986 :         return;
     488             :     }
     489             : 
     490             :     mutable_buffer buffer =
     491        9975 :         session->AllocateBuffer(session->GetReadBufferSize());
     492             : 
     493        9975 :     error_code error;
     494        9975 :     size_t bytes_transferred = session->ReadSome(buffer, &error);
     495        9975 :     if (session->IsSocketErrorHard(error)) {
     496         806 :         session->ReleaseBufferLocked(buffer);
     497             :         // eof is returned when the peer closed the socket, no need to log error
     498         806 :         if (error != boost::asio::error::eof) {
     499           4 :             if (strcmp(error.category().name(), "asio.ssl") == 0  &&
     500           0 :                 error.value() == SSL_SHORT_READ_ERROR) {
     501           0 :             TCP_SESSION_LOG_DEBUG(session, TCP_DIR_IN,
     502             :                     "Read failed due to error "
     503             :                     << error.category().name() << " "
     504             :                     << error.value()
     505             :                     << " : " << error.message());
     506             :             } else {
     507           4 :             TCP_SESSION_LOG_ERROR(session, TCP_DIR_IN,
     508             :                     "Read failed due to error "
     509             :                     << error.category().name() << " "
     510             :                     << error.value()
     511             :                     << " : " << error.message());
     512             :             }
     513             :         }
     514         806 :         lock.unlock();
     515         806 :         session->CloseInternal(error, true);
     516         806 :         return;
     517             :     }
     518             : 
     519             :     // Update read statistics.
     520        9169 :     session->stats_.read_calls++;
     521        9169 :     session->stats_.read_bytes += bytes_transferred;
     522        9169 :     session->server_->stats_.read_calls++;
     523        9169 :     session->server_->stats_.read_bytes += bytes_transferred;
     524             : 
     525        9169 :     Task *task = session->CreateReaderTask(buffer, bytes_transferred);
     526             :     // Starting a new task for the session
     527        9169 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     528        9169 :     scheduler->Enqueue(task);
     529       10961 : }
     530             : 
     531        4599 : int TcpSession::GetSessionInstance() const {
     532        4599 :     return Task::kTaskInstanceAny;
     533             : }
     534             : 
     535             : 
     536         127 : int32_t TcpSession::local_port() const {
     537         127 :     if (socket() == NULL) {
     538           0 :         return -1;
     539             :     }
     540         127 :     error_code error;
     541         127 :     Endpoint local = socket()->local_endpoint(error);
     542         127 :     if (IsSocketErrorHard(error)) {
     543           0 :         return -1;
     544             :     }
     545         127 :     return local.port();
     546             : }
     547             : 
     548         127 : int32_t TcpSession::remote_port() const {
     549         127 :     if (socket() == NULL) {
     550           0 :         return -1;
     551             :     }
     552         127 :     error_code error;
     553         127 :     Endpoint remote = socket()->remote_endpoint(error);
     554         127 :     if (IsSocketErrorHard(error)) {
     555          88 :         return -1;
     556             :     }
     557          39 :     return remote.port();
     558             : }
     559             : 
     560           0 : int TcpSession::SetMd5SocketOption(uint32_t peer_ip,
     561             :                                    const string &md5_password) {
     562           0 :     return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip,
     563           0 :                                         md5_password);
     564             : }
     565             : 
     566           0 : int TcpSession::ClearMd5SocketOption(uint32_t peer_ip) {
     567           0 :     return server()->SetMd5SocketOption(socket_->native_handle(), peer_ip, "");
     568             : }
     569             : 
     570         700 : int TcpSession::SetDscpSocketOption(uint8_t value) {
     571         700 :     return server()->SetDscpSocketOption(socket()->native_handle(), value);
     572             : }
     573             : 
     574           0 : uint8_t TcpSession::GetDscpValue() const {
     575           0 :     return server_->GetDscpValue(socket()->native_handle());
     576             : }
     577             : 
     578        1688 : TcpMessageReader::TcpMessageReader(TcpSession *session,
     579        1688 :                                    ReceiveCallback callback)
     580        1688 :     : session_(session), callback_(callback), offset_(0), remain_(-1) {
     581        1682 : }
     582             : 
     583        1695 : TcpMessageReader::~TcpMessageReader() {
     584        1695 : }
     585             : 
     586             : // Returns a buffer allocation size that is larger than the message.
     587           2 : int TcpMessageReader::AllocBufferSize(int length) {
     588           2 :     const int kMaxMessageSize = GetMaxMessageSize();
     589           2 :     if (length == -1) {
     590           0 :         return kMaxMessageSize;
     591             :     }
     592           2 :     int bufsize = 1 << 8;
     593           3 :     for (; bufsize < kMaxMessageSize && bufsize < length; bufsize <<= 1) {
     594             :     }
     595           2 :     return bufsize;
     596             : }
     597             : 
     598           2 : uint8_t *TcpMessageReader::BufferConcat(uint8_t *data, Buffer buffer,
     599             :                                         int msglength) {
     600           2 :     uint8_t *dst = data;
     601             : 
     602           7 :     while (!queue_.empty()) {
     603           5 :         Buffer head = queue_.front();
     604           5 :         const uint8_t *cp = TcpSession::BufferData(head) + offset_;
     605           5 :         int bytes = TcpSession::BufferSize(head) - offset_;
     606           5 :         assert((dst - data) + bytes < msglength);
     607           5 :         memcpy(dst, cp, bytes);
     608           5 :         dst += bytes;
     609           5 :         queue_.pop_front();
     610           5 :         session_->ReleaseBuffer(head);
     611           5 :         offset_ = 0;
     612           5 :         remain_ = -1;
     613             :     }
     614             : 
     615           2 :     int count = msglength - (dst - data);
     616           2 :     assert((dst - data) + count <= msglength);
     617           2 :     memcpy(dst, TcpSession::BufferData(buffer), count);
     618           2 :     offset_ = count;
     619             : 
     620           2 :     return data;
     621             : }
     622             : 
     623           3 : int TcpMessageReader::QueueByteLength() const {
     624           3 :     int total = 0;
     625           3 :     for (BufferQueue::const_iterator iter = queue_.begin();
     626           9 :          iter != queue_.end(); ++iter) {
     627           6 :         if (total == 0) {
     628           3 :             total = TcpSession::BufferSize(*iter) - offset_;
     629             :         } else {
     630           3 :             total += TcpSession::BufferSize(*iter);
     631             :         }
     632             :     }
     633           3 :     return total;
     634             : }
     635             : 
     636           2 : TcpMessageReader::Buffer TcpMessageReader::PullUp(
     637             :                uint8_t *data, Buffer buffer, size_t size) const {
     638           2 :     size_t offset = 0;
     639             : 
     640           2 :     for (BufferQueue::const_iterator iter = queue_.begin();
     641           7 :          iter != queue_.end(); ++iter) {
     642             :         const uint8_t *cp;
     643             :         int avail;
     644           5 :         if (offset == 0) {
     645           2 :             cp = TcpSession::BufferData(*iter) + offset_;
     646           2 :             avail = TcpSession::BufferSize(*iter) - offset_;
     647             :         } else {
     648           3 :             cp = TcpSession::BufferData(*iter);
     649           3 :             avail = TcpSession::BufferSize(*iter);
     650             :         }
     651           5 :         int remain = size - offset;
     652           5 :         avail = min(avail, remain);
     653           5 :         assert(offset + avail <= size);
     654           5 :         memcpy(data + offset, cp, avail);
     655           5 :         offset += avail;
     656             :     }
     657             : 
     658           2 :     int avail = TcpSession::BufferSize(buffer);
     659           2 :     int remain = size - offset;
     660           2 :     avail = min(avail, remain);
     661           2 :     assert(offset + avail <= size);
     662           2 :     memcpy(data + offset, TcpSession::BufferData(buffer), avail);
     663           2 :     offset += avail;
     664             : 
     665           2 :     if (offset < size) {
     666           0 :         return Buffer();
     667             :     }
     668           2 :     return Buffer(data, size);
     669             : }
     670             : 
     671             : // Read the socket stream and send messages to the peer object.
     672        4240 : void TcpMessageReader::OnRead(Buffer buffer) {
     673        4240 :     const int kHeaderLenSize = GetHeaderLenSize();
     674        4239 :     size_t size = TcpSession::BufferSize(buffer);
     675        4238 :     TCP_SESSION_LOG_UT_DEBUG(session_, TCP_DIR_IN, "Read " << size << " bytes");
     676             : 
     677        4238 :     if (!queue_.empty()) {
     678           5 :         int msglength = MsgLength(queue_.front(), offset_);
     679           5 :         if (msglength < 0) {
     680           3 :             int queuelen = QueueByteLength();
     681           3 :             if (queuelen + static_cast<int>(size) < kHeaderLenSize) {
     682           1 :                 queue_.push_back(buffer);
     683           1 :                 return;
     684             :             }
     685           2 :             scoped_array<uint8_t> data(new uint8_t[kHeaderLenSize]);
     686           2 :             Buffer header = PullUp(data.get(), buffer, kHeaderLenSize);
     687           2 :             assert(TcpSession::BufferSize(header) == (size_t) kHeaderLenSize);
     688             : 
     689           2 :             msglength = MsgLength(header, 0);
     690           2 :             remain_ = msglength - queuelen;
     691           2 :         }
     692             : 
     693           4 :         assert(remain_ > 0);
     694           4 :         if (size < (size_t) remain_) {
     695           2 :             queue_.push_back(buffer);
     696           2 :             remain_ -= size;
     697           2 :             return;
     698             :         }
     699             : 
     700             :         // concat the buffers into a contiguous message.
     701           2 :         scoped_array<uint8_t> data(new uint8_t[AllocBufferSize(msglength)]);
     702           2 :         BufferConcat(data.get(), buffer, msglength);
     703           2 :         assert(remain_ == -1);
     704             :         // Receive the message
     705           2 :         bool success = callback_(data.get(), msglength);
     706           2 :         if (!success)
     707           0 :             return;
     708           2 :     }
     709             : 
     710        4235 :     int avail = size - offset_;
     711        8613 :     while (avail > 0) {
     712        4383 :         int msglength = MsgLength(buffer, offset_);
     713        4382 :         if (msglength < 0) {
     714           1 :             break;
     715             :         }
     716        4381 :         if (msglength > avail) {
     717           1 :             remain_ = msglength - avail;
     718           1 :             break;
     719             :         }
     720             :         // Receive the message
     721             :         bool success =
     722        4380 :             callback_(TcpSession::BufferData(buffer) + offset_, msglength);
     723        4382 :         offset_ += msglength;
     724        4382 :         avail -= msglength;
     725        4382 :         if (!success)
     726           4 :             return;
     727             :     }
     728             : 
     729        4232 :     if (avail > 0) {
     730           2 :         queue_.push_back(buffer);
     731             :     } else {
     732        4230 :         session_->ReleaseBuffer(buffer);
     733        4232 :         offset_ = 0;
     734        4232 :         assert(remain_ == -1);
     735             :     }
     736             : }
     737             : 
     738             : //
     739             : // Check if a socker error is hard and fatal. Only then should we close the
     740             : // socket. Soft errors like EINTR and EAGAIN should be ignored or properly
     741             : // handled with retries
     742             : //
     743       20013 : bool TcpSession::IsSocketErrorHard(const error_code &ec) {
     744       20013 :     if (!ec)
     745       18565 :         return false;
     746        1448 :     if (ec == try_again)
     747         554 :         return false;
     748         894 :     if (ec == would_block)
     749           0 :         return false;
     750         894 :     if (ec == in_progress)
     751           0 :         return false;
     752         894 :     if (ec == interrupted)
     753           0 :         return false;
     754         894 :     if (ec == network_down)
     755           0 :         return false;
     756         894 :     if (ec == network_reset)
     757           0 :         return false;
     758         894 :     if (ec == network_unreachable)
     759           0 :         return false;
     760         894 :     if (ec == no_buffer_space)
     761           0 :         return false;
     762             : 
     763         894 :     return true;
     764             : }
     765             : 
     766           0 : error_code TcpSession::SetTcpNoDelay() {
     767           0 :     error_code ec;
     768           0 :     boost::asio::ip::tcp::no_delay no_delay_option(true);
     769           0 :     socket()->set_option(no_delay_option, ec);
     770           0 :     if (ec) {
     771           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     772             :                 "tcp_no_delay set error: " << ec);
     773             :     }
     774           0 :     return ec;
     775             : }
     776             : 
     777           0 : error_code TcpSession::SetTcpSendBufSize(uint32_t size) {
     778           0 :     error_code ec;
     779           0 :     socket_base::send_buffer_size send_buffer_size_option(size);
     780           0 :     socket()->set_option(send_buffer_size_option, ec);
     781           0 :     if (ec) {
     782           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     783             :                               "send_buffer_size set error: " << ec);
     784           0 :         return ec;
     785             :     }
     786             : 
     787           0 :     return ec;
     788             : }
     789             : 
     790           0 : error_code TcpSession::SetTcpRecvBufSize(uint32_t size) {
     791           0 :     error_code ec;
     792           0 :     socket_base::receive_buffer_size receive_buffer_size_option(size);
     793           0 :     socket()->set_option(receive_buffer_size_option, ec);
     794           0 :     if (ec) {
     795           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_IN,
     796             :                               "receive_buffer_size set error: " << ec);
     797           0 :         return ec;
     798             :     }
     799             : 
     800           0 :     return ec;
     801             : }
     802             : 
     803         184 : error_code TcpSession::SetSocketKeepaliveOptions(int keepalive_time,
     804             :         int keepalive_intvl, int keepalive_probes, int tcp_user_timeout_val) {
     805         184 :     error_code ec;
     806         184 :     socket_base::keep_alive keep_alive_option(true);
     807         184 :     socket()->set_option(keep_alive_option, ec);
     808         184 :     if (ec) {
     809          22 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     810             :                 "keep_alive set error: " << ec);
     811          22 :         return ec;
     812             :     }
     813             : #ifdef TCP_KEEPIDLE
     814             :     typedef integer< IPPROTO_TCP, TCP_KEEPIDLE > keepalive_idle_time;
     815         162 :     keepalive_idle_time keepalive_idle_time_option(keepalive_time);
     816         162 :     socket()->set_option(keepalive_idle_time_option, ec);
     817         162 :     if (ec) {
     818           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     819             :             "keepalive_idle_time: " << keepalive_time << " set error: " << ec);
     820           0 :         return ec;
     821             :     }
     822             : #elif TCP_KEEPALIVE
     823             :     typedef integer< IPPROTO_TCP, TCP_KEEPALIVE > keepalive_idle_time;
     824             :     keepalive_idle_time keepalive_idle_time_option(keepalive_time);
     825             :     socket()->set_option(keepalive_idle_time_option, ec);
     826             :     if (ec) {
     827             :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     828             :             "keepalive_idle_time: " << keepalive_time << " set error: " << ec);
     829             :         return ec;
     830             :     }
     831             : #else
     832             : #error No TCP keepalive option defined.
     833             : #endif
     834             : #ifdef TCP_KEEPINTVL
     835             :     typedef integer< IPPROTO_TCP, TCP_KEEPINTVL > keepalive_interval;
     836         162 :     keepalive_interval keepalive_interval_option(keepalive_intvl);
     837         162 :     socket()->set_option(keepalive_interval_option, ec);
     838         162 :     if (ec) {
     839           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     840             :             "keepalive_interval: " << keepalive_intvl << " set error: " << ec);
     841           0 :         return ec;
     842             :     }
     843             : #endif
     844             : #ifdef TCP_KEEPCNT
     845             :     typedef integer< IPPROTO_TCP, TCP_KEEPCNT > keepalive_count;
     846         162 :     keepalive_count keepalive_count_option(keepalive_probes);
     847         162 :     socket()->set_option(keepalive_count_option, ec);
     848         162 :     if (ec) {
     849           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     850             :             "keepalive_probes: " << keepalive_probes << " set error: " << ec);
     851           0 :         return ec;
     852             :     }
     853             : #endif
     854             : #ifdef TCP_USER_TIMEOUT
     855             :     typedef integer< IPPROTO_TCP, TCP_USER_TIMEOUT > tcp_user_timeout;
     856         162 :     tcp_user_timeout tcp_user_timeout_option(tcp_user_timeout_val);
     857         162 :     socket()->set_option(tcp_user_timeout_option, ec);
     858         162 :     if (ec) {
     859           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     860             :             "tcp_user_timeout: " << tcp_user_timeout_val << " set error: "
     861             :             << ec);
     862           0 :         return ec;
     863             :     }
     864             : #endif
     865             : 
     866         162 :     return ec;
     867             : }
     868             : 
     869        3468 : error_code TcpSession::SetSocketOptions() {
     870        3468 :     error_code ec;
     871             : 
     872             :     //
     873             :     // Make socket write non-blocking
     874             :     //
     875        3468 :     socket()->non_blocking(true, ec);
     876        3467 :     if (ec) {
     877           0 :         TCP_SESSION_LOG_ERROR(this, TCP_DIR_NA,
     878             :                               "Cannot set socket non blocking: " << ec);
     879           0 :         return ec;
     880             :     }
     881             : 
     882        3467 :     char *buffer_size_str = getenv("TCP_SESSION_SOCKET_BUFFER_SIZE");
     883        3467 :     if (!buffer_size_str) return ec;
     884             : 
     885           0 :     uint32_t sz = static_cast<uint32_t>(strtoul(buffer_size_str, NULL, 0));
     886           0 :     if (sz) {
     887             :         //
     888             :         // Set socket send and receive buffer size
     889             :         //
     890             :         // Currently used only under test environments to trigger partial
     891             :         // sends more deterministically
     892             :         //
     893           0 :         socket_base::send_buffer_size send_buffer_size_option(sz);
     894           0 :         socket()->set_option(send_buffer_size_option, ec);
     895           0 :         if (ec) {
     896           0 :             TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     897             :                                   "send_buffer_size set error: " << ec);
     898           0 :             return ec;
     899             :         }
     900             : 
     901           0 :         socket_base::receive_buffer_size receive_buffer_size_option(sz);
     902           0 :         socket()->set_option(receive_buffer_size_option, ec);
     903           0 :         if (ec) {
     904           0 :             TCP_SESSION_LOG_ERROR(this, TCP_DIR_IN,
     905             :                                   "receive_buffer_size set error: " << ec);
     906           0 :             return ec;
     907             :         }
     908             :     }
     909             : 
     910           0 :     return ec;
     911             : }
     912             : 
     913           8 : void TcpSession::GetRxSocketStats(SocketIOStats *socket_stats) const {
     914           8 :     stats_.GetRxStats(socket_stats);
     915           8 : }
     916             : 
     917           6 : void TcpSession::GetTxSocketStats(SocketIOStats *socket_stats) const {
     918           6 :     stats_.GetTxStats(socket_stats);
     919           6 : }