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         612 :     Reader(TcpSessionPtr session, ReadHandler read_fn, Buffer buffer)
      62        2448 :         : Task(session->reader_task_id(), session->GetSessionInstance()),
      63         612 :           session_(session), read_fn_(read_fn), buffer_(buffer) {
      64         612 :     }
      65         612 :     virtual bool Run() {
      66         612 :         if (session_->IsEstablished()) {
      67         612 :             read_fn_(buffer_);
      68         612 :             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         611 :                 session_->AsyncReadStart();
      75             :             }
      76             :         }
      77         612 :         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        3935 : TcpSession::TcpSession(
      88             :     TcpServer *server, Socket *socket, bool async_read_ready,
      89        3935 :     size_t buffer_send_size)
      90        3935 :     : server_(server),
      91        3935 :       socket_(socket),
      92        3935 :       read_on_connect_(async_read_ready),
      93        3935 :       established_(false),
      94        3935 :       closed_(false),
      95        3935 :       direction_(ACTIVE),
      96        3935 :       writer_(new TcpMessageWriter(this, buffer_send_size)),
      97       11804 :       name_("-") {
      98        3934 :     refcount_ = 0;
      99        3935 :     if (reader_task_id_ == -1) {
     100          57 :         TaskScheduler *scheduler = TaskScheduler::GetInstance();
     101          57 :         reader_task_id_ = scheduler->GetTaskId("io::ReaderTask");
     102             :     }
     103        3935 :     if (server_) {
     104        3932 :         io_strand_.reset(new Strand(server->event_manager()->io_service()->get_executor()));
     105             :     }
     106        3935 :     defer_reader_ = false;
     107        3935 :     write_blocked_ = false;
     108        3935 :     tcp_close_in_progress_ = false;
     109        3935 : }
     110             : 
     111        3931 : TcpSession::~TcpSession() {
     112        3931 :     assert(!established_);
     113        3931 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
     114        9043 :          iter != buffer_queue_.end(); ++iter) {
     115        5112 :         DeleteBuffer(*iter);
     116             :     }
     117        3931 :     buffer_queue_.clear();
     118        3931 : }
     119             : 
     120       18179 : mutable_buffer TcpSession::AllocateBuffer(size_t buffer_size) {
     121       18179 :     uint8_t *data = new uint8_t[buffer_size];
     122       18179 :     mutable_buffer buffer = mutable_buffer(data, buffer_size);
     123       18179 :     buffer_queue_.push_back(buffer);
     124       18179 :     return buffer;
     125             : }
     126             : 
     127       18169 : void TcpSession::DeleteBuffer(mutable_buffer buffer) {
     128       18169 :     uint8_t *data = buffer_cast<uint8_t *>(buffer);
     129       18169 :     delete[] data;
     130       18179 : }
     131             : 
     132       15357 : static int BufferCmp(const mutable_buffer &lhs, const const_buffer &rhs) {
     133       15357 :     const uint8_t *lp = buffer_cast<uint8_t *>(lhs);
     134       15357 :     const uint8_t *rp = buffer_cast<const uint8_t *>(rhs);
     135       15357 :     if (lp < rp) {
     136        2250 :         return -1;
     137             :     }
     138       13107 :     if (lp > rp) {
     139          46 :         return 1;
     140             :     }
     141       13061 :     return 0;
     142             : }
     143             : 
     144       11289 : void TcpSession::ReleaseBuffer(Buffer buffer) {
     145       11289 :     std::scoped_lock lock(mutex_);
     146       11293 :     ReleaseBufferLocked(buffer);
     147       11288 : }
     148             : 
     149       13067 : void TcpSession::ReleaseBufferLocked(Buffer buffer) {
     150       13067 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
     151       15360 :          iter != buffer_queue_.end(); ++iter) {
     152       15360 :         if (BufferCmp(*iter, buffer) == 0) {
     153       13061 :             DeleteBuffer(*iter);
     154       13067 :             buffer_queue_.erase(iter);
     155       13062 :             return;
     156             :         }
     157             :     }
     158           0 :     assert(false);
     159             : }
     160             : 
     161       19465 : void TcpSession::AsyncReadStartInternal(TcpSessionPtr session) {
     162             :     // Update socket read block time.
     163       19465 :     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       19465 :     std::scoped_lock lock(mutex_);
     172       19465 :     AsyncReadSome();
     173       19465 : }
     174             : 
     175       19465 : void TcpSession::AsyncReadStart() {
     176       19465 :     if (io_strand_) {
     177       19463 :         boost::asio::detail::recycling_allocator<void> allocator;
     178       38927 :         io_strand_->post(bind(&TcpSession::AsyncReadStartInternal, this,
     179       38918 :                          TcpSessionPtr(this)), allocator);
     180             :     }
     181       19468 : }
     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        9253 : void TcpSession::AsyncReadSome() {
     194        9253 :     if (IsEstablishedLocked()) {
     195       18506 :         socket()->async_read_some(null_buffers(),
     196       18506 :             bind(&TcpSession::AsyncReadHandler, TcpSessionPtr(this)));
     197             :     }
     198        9253 : }
     199             : 
     200        7611 : void TcpSession::AsyncWrite(const uint8_t *data, std::size_t size) {
     201        7611 :     async_write(*socket(), buffer(data, size),
     202       15222 :         bind(&TcpSession::AsyncWriteHandler, TcpSessionPtr(this),
     203             :              error, bytes_transferred));
     204        7611 : }
     205             : 
     206      172088 : TcpSession::Endpoint TcpSession::local_endpoint() const {
     207      172088 :     std::scoped_lock lock(mutex_);
     208      172088 :     if (!established_)
     209           0 :         return Endpoint();
     210             : 
     211      172088 :     error_code error;
     212      172088 :     Endpoint local = socket()->local_endpoint(error);
     213      172088 :     if (error) {
     214           0 :         return Endpoint();
     215             :     }
     216      172088 :     return local;
     217      172088 : }
     218             : 
     219       12153 : void TcpSession::set_observer(EventObserver observer) {
     220       12153 :     std::scoped_lock lock(obs_mutex_);
     221       12167 :     observer_ = observer;
     222       12121 : }
     223             : 
     224        3637 : void TcpSession::SetName() {
     225        3637 :     ostringstream out;
     226        3637 :     error_code error;
     227        3637 :     Endpoint local;
     228             : 
     229        3637 :     local = socket()->local_endpoint(error);
     230        3637 :     out << local.address().to_string() << ":" << local.port() << "::";
     231        3637 :     out << remote_.address().to_string() << ":" << remote_.port();
     232             : 
     233        3637 :     name_ = out.str();
     234             : 
     235        3637 :     out.str("");
     236        3637 :     std::string hostname = "";
     237        3637 :     if (local.address().is_v4()) {
     238        3634 :         hostname = ResolveCanonicalName(local.address().to_string());
     239             :     } else {
     240           3 :         hostname = ResolveCanonicalNameIPv6(local.address().to_string());
     241             :     }
     242        3637 :     out << hostname << ":" << remote_.address().to_string();
     243        3637 :     uve_key_str_ = out.str();
     244        3637 : }
     245             : 
     246        3637 : void TcpSession::SessionEstablished(Endpoint remote,
     247             :                                     Direction direction) {
     248        3637 :     established_ = true;
     249        3637 :     remote_ = remote;
     250        3637 :     remote_addr_str_ = remote.address().to_string();
     251        3637 :     direction_ = direction;
     252        3637 :     SetName();
     253        3637 : }
     254             : 
     255        1745 : void TcpSession::Accepted() {
     256        1745 :     TCP_SESSION_LOG_DEBUG(this, TCP_DIR_OUT,
     257             :                           "Passive session Accept complete");
     258             :     {
     259        1745 :         std::scoped_lock obs_lock(obs_mutex_);
     260        1745 :         if (observer_) {
     261        1553 :             observer_(this, ACCEPT);
     262             :         }
     263        1745 :     }
     264             : 
     265        1745 :     if (read_on_connect_) {
     266         214 :         AsyncReadStart();
     267             :     }
     268        1745 : }
     269             : 
     270        1842 : bool TcpSession::Connected(Endpoint remote) {
     271        1842 :     assert(refcount_);
     272             : 
     273             :     {
     274        1842 :         std::scoped_lock lock(mutex_);
     275        1842 :         if (closed_) {
     276           0 :             return false;
     277             :         }
     278        1842 :         SessionEstablished(remote, TcpSession::ACTIVE);
     279        1842 :     }
     280        1842 :     SetSocketOptions();
     281             : 
     282        1842 :     TCP_SESSION_LOG_DEBUG(this, TCP_DIR_IN,
     283             :                           "Active session connection complete");
     284             : 
     285             :     {
     286        1842 :         std::scoped_lock obs_lock(obs_mutex_);
     287        1842 :         if (observer_) {
     288        1842 :             observer_(this, CONNECT_COMPLETE);
     289             :         }
     290        1842 :     }
     291             : 
     292        1842 :     if (read_on_connect_) {
     293        1842 :         AsyncReadStart();
     294             :     }
     295        1842 :     return true;
     296             : }
     297             : 
     298         226 : void TcpSession::ConnectFailed() {
     299         226 :     std::scoped_lock obs_lock(obs_mutex_);
     300         226 :     if (observer_) {
     301         226 :         observer_(this, CONNECT_FAILED);
     302             :     }
     303         226 : }
     304             : 
     305             : // Requires: lock must not be held
     306        3922 : void TcpSession::CloseInternal(const error_code &ec,
     307             :                                bool call_observer, bool notify_server) {
     308        3922 :     std::unique_lock<std::mutex> lock(mutex_);
     309             : 
     310        3922 :     if (socket() != NULL && !closed_) {
     311        3906 :         error_code error;
     312        3906 :         socket()->shutdown(tcp::socket::shutdown_both, error);
     313        3932 :         if (error) {
     314         503 :             TCP_SESSION_LOG_ERROR(this, TCP_DIR_OUT,
     315             :                 "Shutdown failed due to error: " << error.message());
     316             :         }
     317        3930 :         socket()->close(error);
     318             :     }
     319        3934 :     closed_ = true;
     320        3934 :     tcp_close_in_progress_ = false;
     321             : 
     322        3935 :     if (!established_) {
     323         298 :         return;
     324             :     }
     325        3637 :     established_ = false;
     326             : 
     327             :     // copy the ec to close reason
     328        3637 :     close_reason_ = ec;
     329             : 
     330             :     // Take a reference through intrusive pointer to protect session from
     331             :     // possibly getting deleted from another thread.
     332        3637 :     TcpSessionPtr session = TcpSessionPtr(this);
     333        3637 :     lock.unlock();
     334             : 
     335        3637 :     if (call_observer) {
     336        1989 :         std::scoped_lock obs_lock(obs_mutex_);
     337        1989 :         if (observer_) {
     338        1628 :             observer_(this, CLOSE);
     339             :         }
     340        1989 :     }
     341             : 
     342        3637 :     if (notify_server) {
     343        3587 :         server_->OnSessionClose(this);
     344             :     }
     345        3935 : }
     346             : 
     347       10098 : void TcpSession::TriggerAsyncReadHandler() {
     348       10098 :     if (io_strand_) {
     349       10098 :         boost::asio::detail::recycling_allocator<void> allocator;
     350       20196 :         io_strand_->post(bind(&TcpSession::AsyncReadHandler,
     351       20196 :                               TcpSessionPtr(this)), allocator);
     352             :     }
     353       10098 : }
     354             : 
     355        7523 : void TcpSession::Close() {
     356        7523 :     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        7533 :     if (closed_) {
     362        5279 :         return;
     363             :     }
     364             : 
     365        2254 :     if (server_ && writer_->IsWritePending()) {
     366         373 :         tcp_close_in_progress_ = true;
     367         373 :         return;
     368             :     }
     369        1870 :     lock.unlock();
     370             : 
     371        1883 :     error_code ec;
     372        1883 :     CloseInternal(ec, false);
     373        7545 : }
     374             : 
     375             : // virtual method overriden in derrived classes.
     376           2 : void TcpSession::WriteReady(const error_code &error) {
     377           2 : }
     378             : 
     379       12102 : void TcpSession::AsyncWriteHandler(TcpSessionPtr session,
     380             :                                    const error_code &error,
     381             :                                    std::size_t wrote) {
     382       12102 :     std::unique_lock<std::mutex> lock(session->mutex_);
     383       12102 :     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       12102 :     if (session->IsClosedLocked()) return;
     395             : 
     396             :     // Update socket write bytes statistics.
     397       12005 :     session->stats_.write_bytes += wrote;
     398       12005 :     session->server_->stats_.write_bytes += wrote;
     399             : 
     400       12005 :     bool send_ready = false;
     401       12005 :     bool more_write = session->writer_->UpdateBufferQueue(wrote, &send_ready);
     402             : 
     403             :     // Subsequent write
     404       12005 :     if (more_write) {
     405        2955 :         session->writer_->TriggerAsyncWrite();
     406        9050 :     } else if (session->tcp_close_in_progress_) {
     407         216 :         lock.unlock();
     408         216 :         session->CloseInternal(error, true);
     409         216 :         return;
     410             :     }
     411             : 
     412       11789 :     lock.unlock();
     413       11789 :     if (send_ready)
     414           4 :         session->WriteReady(error);
     415       11789 :     return;
     416       12102 : }
     417             : 
     418        9185 : void TcpSession::AsyncWriteInternal(TcpSessionPtr session) {
     419             : 
     420        9185 :     std::scoped_lock lock(session->mutex_);
     421             : 
     422             :     //
     423             :     // Ignore if connection is already closed.
     424             :     //
     425        9185 :     if (session->IsClosedLocked()) return;
     426        9147 :     session->writer_->TriggerAsyncWrite();
     427        9185 : }
     428             : 
     429       10410 : bool TcpSession::Send(const uint8_t *data, size_t size, size_t *sent) {
     430       10410 :     bool ret = true;
     431       10410 :     std::unique_lock<std::mutex> lock(mutex_);
     432             : 
     433             :     // Reset sent, if provided.
     434       10412 :     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       10412 :     if (!IsEstablishedLocked()) return false;
     441             : 
     442        9911 :     if (socket()->non_blocking()) {
     443        9910 :         error_code error;
     444        9910 :         int len = writer_->AsyncSend(data, size, &error);
     445        9908 :         lock.unlock();
     446        9911 :         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        9911 :         if ((size_t) len != size)
     455          14 :             ret = false;
     456        9911 :         if (sent) *sent = (len > 0) ? len : 0;
     457             :     }
     458        9911 :     return ret;
     459       10412 : }
     460             : 
     461         612 : Task* TcpSession::CreateReaderTask(mutable_buffer buffer,
     462             :                                   size_t bytes_transferred) {
     463         612 :     Buffer rdbuf(buffer_cast<const uint8_t *>(buffer), bytes_transferred);
     464         612 :     Reader *task = new Reader(TcpSessionPtr(this),
     465         612 :                               bind(&TcpSession::OnRead, this, _1), rdbuf);
     466         612 :     return (task);
     467             : }
     468             : 
     469        9649 : size_t TcpSession::ReadSome(mutable_buffer buffer, error_code *error) {
     470        9649 :     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         825 : size_t TcpSession::GetReadBufferSize() const {
     477         825 :     error_code error;
     478         825 :     size_t size = socket_->available(error);
     479         825 :     if (size < kDefaultBufferSize)
     480         524 :         size = kDefaultBufferSize;
     481         825 :     return size;
     482             : }
     483             : 
     484       19298 : void TcpSession::AsyncReadHandler(TcpSessionPtr session) {
     485       19298 :     std::unique_lock<std::mutex> lock(session->mutex_);
     486       19298 :     if (session->closed_) {
     487        1119 :         return;
     488             :     }
     489             : 
     490             :     mutable_buffer buffer =
     491       18179 :         session->AllocateBuffer(session->GetReadBufferSize());
     492             : 
     493       18179 :     error_code error;
     494       18179 :     size_t bytes_transferred = session->ReadSome(buffer, &error);
     495       18179 :     if (session->IsSocketErrorHard(error)) {
     496        1774 :         session->ReleaseBufferLocked(buffer);
     497             :         // eof is returned when the peer closed the socket, no need to log error
     498        1774 :         if (error != boost::asio::error::eof) {
     499         641 :             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         641 :             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        1774 :         lock.unlock();
     515        1774 :         session->CloseInternal(error, true);
     516        1774 :         return;
     517             :     }
     518             : 
     519             :     // Update read statistics.
     520       16405 :     session->stats_.read_calls++;
     521       16405 :     session->stats_.read_bytes += bytes_transferred;
     522       16405 :     session->server_->stats_.read_calls++;
     523       16405 :     session->server_->stats_.read_bytes += bytes_transferred;
     524             : 
     525       16405 :     Task *task = session->CreateReaderTask(buffer, bytes_transferred);
     526             :     // Starting a new task for the session
     527       16405 :     TaskScheduler *scheduler = TaskScheduler::GetInstance();
     528       16405 :     scheduler->Enqueue(task);
     529       19298 : }
     530             : 
     531        4636 : int TcpSession::GetSessionInstance() const {
     532        4636 :     return Task::kTaskInstanceAny;
     533             : }
     534             : 
     535             : 
     536        4870 : int32_t TcpSession::local_port() const {
     537        4870 :     if (socket() == NULL) {
     538           0 :         return -1;
     539             :     }
     540        4870 :     error_code error;
     541        4870 :     Endpoint local = socket()->local_endpoint(error);
     542        4870 :     if (IsSocketErrorHard(error)) {
     543           0 :         return -1;
     544             :     }
     545        4871 :     return local.port();
     546             : }
     547             : 
     548        4871 : int32_t TcpSession::remote_port() const {
     549        4871 :     if (socket() == NULL) {
     550           0 :         return -1;
     551             :     }
     552        4869 :     error_code error;
     553        4869 :     Endpoint remote = socket()->remote_endpoint(error);
     554        4870 :     if (IsSocketErrorHard(error)) {
     555         303 :         return -1;
     556             :     }
     557        4566 :     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           0 : int TcpSession::SetDscpSocketOption(uint8_t value) {
     571           0 :     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           2 : TcpMessageReader::TcpMessageReader(TcpSession *session,
     579           2 :                                    ReceiveCallback callback)
     580           2 :     : session_(session), callback_(callback), offset_(0), remain_(-1) {
     581           2 : }
     582             : 
     583           2 : TcpMessageReader::~TcpMessageReader() {
     584           2 : }
     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           9 : void TcpMessageReader::OnRead(Buffer buffer) {
     673           9 :     const int kHeaderLenSize = GetHeaderLenSize();
     674           9 :     size_t size = TcpSession::BufferSize(buffer);
     675           9 :     TCP_SESSION_LOG_UT_DEBUG(session_, TCP_DIR_IN, "Read " << size << " bytes");
     676             : 
     677           9 :     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           6 :     int avail = size - offset_;
     711          10 :     while (avail > 0) {
     712           7 :         int msglength = MsgLength(buffer, offset_);
     713           7 :         if (msglength < 0) {
     714           1 :             break;
     715             :         }
     716           6 :         if (msglength > avail) {
     717           1 :             remain_ = msglength - avail;
     718           1 :             break;
     719             :         }
     720             :         // Receive the message
     721             :         bool success =
     722           5 :             callback_(TcpSession::BufferData(buffer) + offset_, msglength);
     723           5 :         offset_ += msglength;
     724           5 :         avail -= msglength;
     725           5 :         if (!success)
     726           1 :             return;
     727             :     }
     728             : 
     729           5 :     if (avail > 0) {
     730           2 :         queue_.push_back(buffer);
     731             :     } else {
     732           3 :         session_->ReleaseBuffer(buffer);
     733           3 :         offset_ = 0;
     734           3 :         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       40018 : bool TcpSession::IsSocketErrorHard(const error_code &ec) {
     744       40018 :     if (!ec)
     745       32254 :         return false;
     746        7760 :     if (ec == try_again)
     747        5683 :         return false;
     748        2077 :     if (ec == would_block)
     749           0 :         return false;
     750        2077 :     if (ec == in_progress)
     751           0 :         return false;
     752        2077 :     if (ec == interrupted)
     753           0 :         return false;
     754        2077 :     if (ec == network_down)
     755           0 :         return false;
     756        2077 :     if (ec == network_reset)
     757           0 :         return false;
     758        2077 :     if (ec == network_unreachable)
     759           0 :         return false;
     760        2077 :     if (ec == no_buffer_space)
     761           0 :         return false;
     762             : 
     763        2077 :     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        1703 : error_code TcpSession::SetSocketKeepaliveOptions(int keepalive_time,
     804             :         int keepalive_intvl, int keepalive_probes, int tcp_user_timeout_val) {
     805        1703 :     error_code ec;
     806        1703 :     socket_base::keep_alive keep_alive_option(true);
     807        1703 :     socket()->set_option(keep_alive_option, ec);
     808        1703 :     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        1681 :     keepalive_idle_time keepalive_idle_time_option(keepalive_time);
     816        1681 :     socket()->set_option(keepalive_idle_time_option, ec);
     817        1681 :     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        1681 :     keepalive_interval keepalive_interval_option(keepalive_intvl);
     837        1681 :     socket()->set_option(keepalive_interval_option, ec);
     838        1681 :     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        1681 :     keepalive_count keepalive_count_option(keepalive_probes);
     847        1681 :     socket()->set_option(keepalive_count_option, ec);
     848        1681 :     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        1681 :     tcp_user_timeout tcp_user_timeout_option(tcp_user_timeout_val);
     857        1681 :     socket()->set_option(tcp_user_timeout_option, ec);
     858        1681 :     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        1681 :     return ec;
     867             : }
     868             : 
     869        5761 : error_code TcpSession::SetSocketOptions() {
     870        5761 :     error_code ec;
     871             : 
     872             :     //
     873             :     // Make socket write non-blocking
     874             :     //
     875        5761 :     socket()->non_blocking(true, ec);
     876        5761 :     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        5761 :     char *buffer_size_str = getenv("TCP_SESSION_SOCKET_BUFFER_SIZE");
     883        5761 :     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           9 : void TcpSession::GetRxSocketStats(SocketIOStats *socket_stats) const {
     914           9 :     stats_.GetRxStats(socket_stats);
     915           9 : }
     916             : 
     917           7 : void TcpSession::GetTxSocketStats(SocketIOStats *socket_stats) const {
     918           7 :     stats_.GetTxStats(socket_stats);
     919           7 : }