<commit_before>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
///     http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Dr. <NAME>
/// @author <NAME>
////////////////////////////////////////////////////////////////////////////////

//#define DEBUG_STATISTICS

#include "SocketTask.h"

#include "Basics/MutexLocker.h"
#include "Basics/StringBuffer.h"
#include "Basics/socket-utils.h"
#include "Endpoint/ConnectionInfo.h"
#include "Logger/Logger.h"
#include "Scheduler/EventLoop.h"
#include "Scheduler/JobGuard.h"
#include "Scheduler/Scheduler.h"

using namespace arangodb::basics;
using namespace arangodb::rest;

// -----------------------------------------------------------------------------
// --SECTION--                                      constructors and destructors
// -----------------------------------------------------------------------------

SocketTask::SocketTask(arangodb::EventLoop loop,
                       std::unique_ptr<arangodb::Socket> socket,
                       arangodb::ConnectionInfo&& connectionInfo,
                       double keepAliveTimeout, bool skipInit = false)
    : Task(loop, "SocketTask"),
      _connectionInfo(connectionInfo),
      _readBuffer(TRI_UNKNOWN_MEM_ZONE, READ_BLOCK_SIZE + 1, false),
      _peer(std::move(socket)),
      _keepAliveTimeout(static_cast<long>(keepAliveTimeout * 1000)),
      _keepAliveTimer(_peer->_ioService, _keepAliveTimeout),
      _useKeepAliveTimer(keepAliveTimeout > 0.0),
      _keepAliveTimerActive(false),
      _closeRequested(false),
      _abandoned(false) {
  ConnectionStatisticsAgent::acquire();
  connectionStatisticsAgentSetStart();

  if (!skipInit) {
    _peer->setNonBlocking(true);
    if (!_peer->handshake()) {
      _closedSend = true;
      _closedReceive = true;
    }
  }
}

// -----------------------------------------------------------------------------
// --SECTION--                                                    public methods
// -----------------------------------------------------------------------------

SocketTask::~SocketTask() {
  boost::system::error_code err;
  if (_keepAliveTimerActive) {
    _keepAliveTimer.cancel(err);
  }

  if (err) {
    LOG_TOPIC(ERR, Logger::COMMUNICATION) << "unable to cancel _keepAliveTimer";
  } 
}

void SocketTask::start() {
  if (_closedSend || _closedReceive) {
    LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "cannot start, channel closed";
    return;
  }

  if (_closeRequested) {
    LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
        << "cannot start, close alread in progress";
    return;
  }

  LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
      << "starting communication between server <-> client on socket";
  LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
      << _connectionInfo.serverAddress << ":" << _connectionInfo.serverPort
      << " <-> " << _connectionInfo.clientAddress << ":"
      << _connectionInfo.clientPort;

  auto self = shared_from_this();
  _loop._ioService->post([self, this]() { asyncReadSome(); });
}

// -----------------------------------------------------------------------------
// --SECTION--                                                 protected methods
// -----------------------------------------------------------------------------

void SocketTask::addWriteBuffer(std::unique_ptr<StringBuffer> buffer,
                                RequestStatisticsAgent* statistics) {
  TRI_request_statistics_t* stat =
      statistics == nullptr ? nullptr : statistics->steal();

  addWriteBuffer(buffer.release(), stat);
}

void SocketTask::addWriteBuffer(StringBuffer* buffer,
                                TRI_request_statistics_t* stat) {
  if (_closedSend) {
    LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "SocketTask::addWriteBuffer - "
                                               "dropping output, send stream "
                                               "already closed";

    delete buffer;
    if (stat) {
      LOG_TOPIC(TRACE, Logger::COMMUNICATION)
          << "SocketTask::addWriteBuffer - Statistics release: "
          << stat->to_string();
      TRI_ReleaseRequestStatistics(stat);
    } else {
      LOG_TOPIC(TRACE, Logger::COMMUNICATION)
          << "SocketTask::addWriteBuffer - "
             "Statistics release: nullptr - "
             "nothing to release";
    }

    return;
  }

  MUTEX_LOCKER(locker, _writeLock);

  if (_writeBuffer != nullptr) {
    _writeBuffers.push_back(buffer);
    _writeBuffersStats.push_back(stat);
    return;
  }

  _writeBuffer = buffer;
  _writeBufferStatistics = stat;  // threadsafe? why not pass to
                                  // completedWriteBuffer does this work with
                                  // async?

  if(_writeBufferStatistics){
    _writeBufferStatistics->_writeStart = TRI_StatisticsTime();
  }

  if (_writeBuffer != nullptr) {
    size_t total = _writeBuffer->length();
    size_t written = 0;

    boost::system::error_code err;
    do {
      err.assign(boost::system::errc::success,
                 boost::system::generic_category());
      written = _peer->write(_writeBuffer, err);
      if(_writeBufferStatistics){
        _writeBufferStatistics->_sentBytes += written;
      }
      if (written == total) {
        locker.unlock();
        completedWriteBuffer();
        return;
      }
    } while (err == boost::asio::error::would_block);

    if (err != boost::system::errc::success) {
      LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
          << "SocketTask::addWriteBuffer (write_some) - write on stream "
          << " failed with: " << err.message();
      closeStream();
      return;
    }
      
    boost::system::error_code ec;
    ec.assign(boost::system::errc::success,
              boost::system::generic_category());

    auto self = shared_from_this();
    auto handler = [self, this](const boost::system::error_code& ec,
                                std::size_t transferred) {
      MUTEX_LOCKER(locker, _writeLock);

      if (_writeBufferStatistics){
        _writeBufferStatistics->_sentBytes += transferred;
      }

      if (ec) {
        LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
            << "SocketTask::addWriterBuffer(async_write) - write on stream "
            << " failed with: " << ec.message();
        closeStream();
      } else {
        locker.unlock();
        completedWriteBuffer();
      }
    };
<commit_msg>switch immediately to async writes when socket would block during write<commit_after>////////////////////////////////////////////////////////////////////////////////
/// DISCLAIMER
///
/// Copyright 2014-2016 ArangoDB GmbH, Cologne, Germany
/// Copyright 2004-2014 triAGENS GmbH, Cologne, Germany
///
/// Licensed under the Apache License, Version 2.0 (the "License");
/// you may not use this file except in compliance with the License.
/// You may obtain a copy of the License at
///
///     http://www.apache.org/licenses/LICENSE-2.0
///
/// Unless required by applicable law or agreed to in writing, software
/// distributed under the License is distributed on an "AS IS" BASIS,
/// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
/// See the License for the specific language governing permissions and
/// limitations under the License.
///
/// Copyright holder is ArangoDB GmbH, Cologne, Germany
///
/// @author Dr. <NAME>
/// @author <NAME>
////////////////////////////////////////////////////////////////////////////////

//#define DEBUG_STATISTICS

#include "SocketTask.h"

#include "Basics/MutexLocker.h"
#include "Basics/StringBuffer.h"
#include "Basics/socket-utils.h"
#include "Endpoint/ConnectionInfo.h"
#include "Logger/Logger.h"
#include "Scheduler/EventLoop.h"
#include "Scheduler/JobGuard.h"
#include "Scheduler/Scheduler.h"

using namespace arangodb::basics;
using namespace arangodb::rest;

// -----------------------------------------------------------------------------
// --SECTION--                                      constructors and destructors
// -----------------------------------------------------------------------------

SocketTask::SocketTask(arangodb::EventLoop loop,
                       std::unique_ptr<arangodb::Socket> socket,
                       arangodb::ConnectionInfo&& connectionInfo,
                       double keepAliveTimeout, bool skipInit = false)
    : Task(loop, "SocketTask"),
      _connectionInfo(connectionInfo),
      _readBuffer(TRI_UNKNOWN_MEM_ZONE, READ_BLOCK_SIZE + 1, false),
      _peer(std::move(socket)),
      _keepAliveTimeout(static_cast<long>(keepAliveTimeout * 1000)),
      _keepAliveTimer(_peer->_ioService, _keepAliveTimeout),
      _useKeepAliveTimer(keepAliveTimeout > 0.0),
      _keepAliveTimerActive(false),
      _closeRequested(false),
      _abandoned(false) {
  ConnectionStatisticsAgent::acquire();
  connectionStatisticsAgentSetStart();

  if (!skipInit) {
    _peer->setNonBlocking(true);
    if (!_peer->handshake()) {
      _closedSend = true;
      _closedReceive = true;
    }
  }
}

// -----------------------------------------------------------------------------
// --SECTION--                                                    public methods
// -----------------------------------------------------------------------------

SocketTask::~SocketTask() {
  boost::system::error_code err;
  if (_keepAliveTimerActive) {
    _keepAliveTimer.cancel(err);
  }

  if (err) {
    LOG_TOPIC(ERR, Logger::COMMUNICATION) << "unable to cancel _keepAliveTimer";
  }
}

void SocketTask::start() {
  if (_closedSend || _closedReceive) {
    LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "cannot start, channel closed";
    return;
  }

  if (_closeRequested) {
    LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
        << "cannot start, close alread in progress";
    return;
  }

  LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
      << "starting communication between server <-> client on socket";
  LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
      << _connectionInfo.serverAddress << ":" << _connectionInfo.serverPort
      << " <-> " << _connectionInfo.clientAddress << ":"
      << _connectionInfo.clientPort;

  auto self = shared_from_this();
  _loop._ioService->post([self, this]() { asyncReadSome(); });
}

// -----------------------------------------------------------------------------
// --SECTION--                                                 protected methods
// -----------------------------------------------------------------------------

void SocketTask::addWriteBuffer(std::unique_ptr<StringBuffer> buffer,
                                RequestStatisticsAgent* statistics) {
  TRI_request_statistics_t* stat =
      statistics == nullptr ? nullptr : statistics->steal();

  addWriteBuffer(buffer.release(), stat);
}

void SocketTask::addWriteBuffer(StringBuffer* buffer,
                                TRI_request_statistics_t* stat) {
  if (_closedSend) {
    LOG_TOPIC(DEBUG, Logger::COMMUNICATION) << "SocketTask::addWriteBuffer - "
                                               "dropping output, send stream "
                                               "already closed";

    delete buffer;
    if (stat) {
      LOG_TOPIC(TRACE, Logger::COMMUNICATION)
          << "SocketTask::addWriteBuffer - Statistics release: "
          << stat->to_string();
      TRI_ReleaseRequestStatistics(stat);
    } else {
      LOG_TOPIC(TRACE, Logger::COMMUNICATION)
          << "SocketTask::addWriteBuffer - "
             "Statistics release: nullptr - "
             "nothing to release";
    }

    return;
  }

  MUTEX_LOCKER(locker, _writeLock);

  if (_writeBuffer != nullptr) {
    _writeBuffers.push_back(buffer);
    _writeBuffersStats.push_back(stat);
    return;
  }

  _writeBuffer = buffer;
  _writeBufferStatistics = stat;  // threadsafe? why not pass to
                                  // completedWriteBuffer does this work with
                                  // async?

  if(_writeBufferStatistics){
    _writeBufferStatistics->_writeStart = TRI_StatisticsTime();
  }

  if (_writeBuffer != nullptr) {
    size_t total = _writeBuffer->length();
    size_t written = 0;

    boost::system::error_code err;
    written = _peer->write(_writeBuffer, err);

    if(_writeBufferStatistics){
      _writeBufferStatistics->_sentBytes += written;
    }

    if (written == total) {
      locker.unlock();
      completedWriteBuffer();
      return;
    }

    if (err != boost::system::errc::success && err!= boost::asio::error::would_block) {
      LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
          << "SocketTask::addWriteBuffer (write_some) - write on stream "
          << " failed with: " << err.message();
      closeStream();
      return;
    }

    boost::system::error_code ec;
    ec.assign(boost::system::errc::success,
              boost::system::generic_category());

    auto self = shared_from_this();
    auto handler = [self, this](const boost::system::error_code& ec,
                                std::size_t transferred) {
      MUTEX_LOCKER(locker, _writeLock);

      if (_writeBufferStatistics){
        _writeBufferStatistics->_sentBytes += transferred;
      }

      if (ec) {
        LOG_TOPIC(DEBUG, Logger::COMMUNICATION)
            << "SocketTask::addWriterBuffer(async_write) - write on stream "
            << " failed with: " << ec.message();
        closeStream();
      } else {
        locker.unlock();
        completedWriteBuffer();
      }
    };

{'fraction_non_alphanumeric': 0.20026910275476242, 'fraction_numerical': 0.003965724807024998, 'mean_word_length': 1.9105523495465788, 'pattern_counts': {'":': 4, '<': 101, '<?xml version=': 0, '>': 41, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3929545', 'n_tokens_mistral': 3759, 'n_tokens_neox': 3617, 'n_words': 798}