Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  * Copyright Contributors to the OpenSDN project
       4             :  */
       5             : 
       6             : #include "boost/asio/detail/recycling_allocator.hpp"
       7             : 
       8             : #include "io/tcp_message_write.h"
       9             : 
      10             : #include "base/util.h"
      11             : #include "base/logging.h"
      12             : #include "io/tcp_session.h"
      13             : #include "io/io_log.h"
      14             : 
      15             : using boost::asio::buffer;
      16             : using boost::asio::buffer_cast;
      17             : using boost::asio::mutable_buffer;
      18             : using boost::system::error_code;
      19             : using std::min;
      20             : 
      21             : const int TcpMessageWriter::kDefaultWriteBufferSize;
      22             : const int TcpMessageWriter::kMaxPendingBufferSize;
      23             : const int TcpMessageWriter::kMinPendingBufferSize;
      24             : 
      25       29584 : TcpMessageWriter::TcpMessageWriter(TcpSession *session,
      26       29584 :                                    size_t buffer_send_size) :
      27       29589 :     offset_(0), last_write_(0), buffer_send_size_(buffer_send_size),
      28       29584 :     session_(session) {
      29       29589 : }
      30             : 
      31       29605 : TcpMessageWriter::~TcpMessageWriter() {
      32       29605 :     for (BufferQueue::iterator iter = buffer_queue_.begin();
      33       30345 :          iter != buffer_queue_.end(); ++iter) {
      34         744 :         DeleteBuffer(*iter);
      35             :     }
      36       29598 :     buffer_queue_.clear();
      37       29597 : }
      38             : 
      39     1825487 : int TcpMessageWriter::AsyncSend(const uint8_t *data, size_t len, error_code *ec) {
      40             : 
      41     1825487 :     int write = len;
      42             : 
      43     1825487 :     if (buffer_queue_.empty()) {
      44     1540876 :         BufferAppend(data, len);
      45     1540590 :         if (session_->io_strand_) {
      46     1540590 :             boost::asio::detail::recycling_allocator<void> allocator;
      47     3081545 :             session_->io_strand_->post(bind(&TcpSession::AsyncWriteInternal,
      48     3081145 :                                        session_, TcpSessionPtr(session_)), allocator);
      49             :         }
      50             :     } else {
      51      284654 :         BufferAppend(data, len);
      52             :     }
      53             : 
      54     1825578 :     if ((GetBufferQueueSize() - offset_) > TcpMessageWriter::kMaxPendingBufferSize) {
      55          17 :         if (!session_->write_blocked_) {
      56             :             /* throttle the sender */
      57           5 :             session_->stats_.write_blocked++;
      58           5 :             session_->server_->stats_.write_blocked++;
      59           5 :             session_->stats_.write_block_start_time = UTCTimestampUsec();
      60           5 :             session_->write_blocked_ = true;
      61             :         }
      62          15 :         write = 0;
      63             :     }
      64             : 
      65     1824990 :     return write;
      66             : }
      67             : 
      68     1828375 : void TcpMessageWriter::TriggerAsyncWrite() {
      69             : 
      70             :     /* assert if there is an async write in progress */
      71     1828375 :     assert(last_write_ == 0);
      72     1828375 :     assert(!buffer_queue_.empty());
      73             : 
      74     1828375 :     boost::asio::mutable_buffer head = buffer_queue_.front();
      75     1828375 :     size_t remaining = buffer_size(head) - offset_;
      76     1828375 :     last_write_  = min(buffer_send_size_, remaining);
      77             : 
      78             :     // Update socket write call statistics.
      79     1828375 :     session_->stats_.write_calls++;
      80     1828375 :     session_->server_->stats_.write_calls++;
      81             : 
      82     1828375 :     const uint8_t *data = buffer_cast<const uint8_t *>(head) + offset_;
      83     1828375 :     session_->AsyncWrite(data, last_write_);
      84     1828375 : }
      85             : 
      86     1827675 : bool TcpMessageWriter::UpdateBufferQueue(size_t wrote, bool *send_ready) {
      87             : 
      88     1827675 :     assert(last_write_ == wrote);
      89     1827675 :     assert(!buffer_queue_.empty());
      90             : 
      91     1827675 :     bool more_write = true;
      92     1827675 :     last_write_ = 0;
      93     1827675 :     *send_ready = false;
      94             : 
      95     1827675 :     boost::asio::mutable_buffer head = buffer_queue_.front();
      96     1827675 :     if ((offset_ + wrote) == buffer_size(head)) {
      97     1824958 :         offset_ = 0;
      98     1824958 :         DeleteBuffer(head);
      99     1824958 :         buffer_queue_.pop_front();
     100             :     } else {
     101        2717 :         offset_ += wrote;
     102             :     }
     103             : 
     104     1827675 :     if (session_->write_blocked_ && ((GetBufferQueueSize() - offset_)  <
     105     1827675 :                                      TcpMessageWriter::kMinPendingBufferSize)) {
     106           5 :         uint64_t blocked_usecs =  UTCTimestampUsec() -
     107           5 :                 session_->stats_.write_block_start_time;
     108           5 :         session_->stats_.write_blocked_duration_usecs += blocked_usecs;
     109           5 :         session_->server_->stats_.write_blocked_duration_usecs += blocked_usecs;
     110           5 :         session_->write_blocked_ = false;
     111           5 :         *send_ready = true;
     112             :     }
     113             : 
     114     1827675 :     if (buffer_queue_.empty()) {
     115     1540278 :         buffer_queue_.clear();
     116     1540278 :         more_write = false;
     117             :     }
     118             : 
     119     1827675 :     return more_write;
     120             : }
     121             : 
     122     1825489 : void TcpMessageWriter::BufferAppend(const uint8_t *src, int bytes) {
     123     1825489 :     uint8_t *data = new uint8_t[bytes];
     124     1825616 :     memcpy(data, src, bytes);
     125     1825616 :     mutable_buffer buffer = mutable_buffer(data, bytes);
     126     1825400 :     buffer_queue_.push_back(buffer);
     127     1825092 : }
     128             : 
     129     1825702 : void TcpMessageWriter::DeleteBuffer(mutable_buffer buffer) {
     130     1825702 :     const uint8_t *data = buffer_cast<const uint8_t *>(buffer);
     131     1825702 :     delete[] data;
     132     1825702 :     return;
     133             : }
     134             :