<commit_before>// Copyright (c) 2013, Cloudera, inc.

#include "rpc/connection.h"

#include <boost/foreach.hpp>
#include <boost/intrusive/list.hpp>
#include <boost/lexical_cast.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <stdint.h>

#include <iostream>
#include <string>
#include <vector>

#include "gutil/map-util.h"
#include "rpc/auth_store.h"
#include "rpc/constants.h"
#include "rpc/messenger.h"
#include "rpc/reactor.h"
#include "rpc/rpc_controller.h"
#include "rpc/rpc_header.pb.h"
#include "rpc/sasl_client.h"
#include "rpc/sasl_server.h"
#include "rpc/transfer.h"
#include "util/debug-util.h"
#include "util/net/sockaddr.h"
#include "util/status.h"
#include "util/trace.h"

using std::tr1::shared_ptr;
using std::vector;

namespace kudu {
namespace rpc {

///
/// Connection
///
Connection::Connection(ReactorThread *reactor_thread, const Sockaddr &remote,
                       int socket, Direction direction)
  : reactor_thread_(reactor_thread),
    socket_(socket),
    remote_(remote),
    direction_(direction),
    last_activity_time_(reactor_thread_->cur_time()),
    is_epoll_registered_(false),
    next_call_id_(1),
    sasl_client_(kSaslAppName, socket),
    sasl_server_(kSaslAppName, socket),
    negotiation_complete_(false) {
}

Status Connection::SetNonBlocking(bool enabled) {
  return socket_.SetNonBlocking(enabled);
}

void Connection::EpollRegister(ev::loop_ref& loop) {
  DVLOG(4) << "Registering connection for epoll: " << ToString();
  write_io_.set(loop);
  write_io_.set(socket_.GetFd(), ev::WRITE);
  write_io_.set<Connection, &Connection::WriteHandler>(this);
  if (direction_ == CLIENT && negotiation_complete_) {
    write_io_.start();
  }
  read_io_.set(loop);
  read_io_.set(socket_.GetFd(), ev::READ);
  read_io_.set<Connection, &Connection::ReadHandler>(this);
  read_io_.start();
  is_epoll_registered_ = true;
}

Connection::~Connection() {
  // Must clear the outbound_transfers_ list before deleting.
  CHECK(outbound_transfers_.begin() == outbound_transfers_.end());

  // It's crucial that the connection is Shutdown first -- otherwise
  // our destructor will end up calling read_io_.stop() and write_io_.stop()
  // from a possibly non-reactor thread context. This can then make all
  // hell break loose with libev.
  CHECK(!is_epoll_registered_);
}

bool Connection::Idle() const {
  // check if we're in the middle of receiving something
  InboundTransfer *transfer = inbound_.get();<commit_msg>rpc: Add DCHECK to enforce only reactor thread starts the libev epoll<commit_after>// Copyright (c) 2013, Cloudera, inc.

#include "rpc/connection.h"

#include <boost/foreach.hpp>
#include <boost/intrusive/list.hpp>
#include <boost/lexical_cast.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
#include <stdint.h>

#include <iostream>
#include <string>
#include <vector>

#include "gutil/map-util.h"
#include "rpc/auth_store.h"
#include "rpc/constants.h"
#include "rpc/messenger.h"
#include "rpc/reactor.h"
#include "rpc/rpc_controller.h"
#include "rpc/rpc_header.pb.h"
#include "rpc/sasl_client.h"
#include "rpc/sasl_server.h"
#include "rpc/transfer.h"
#include "util/debug-util.h"
#include "util/net/sockaddr.h"
#include "util/status.h"
#include "util/trace.h"

using std::tr1::shared_ptr;
using std::vector;

namespace kudu {
namespace rpc {

///
/// Connection
///
Connection::Connection(ReactorThread *reactor_thread, const Sockaddr &remote,
                       int socket, Direction direction)
  : reactor_thread_(reactor_thread),
    socket_(socket),
    remote_(remote),
    direction_(direction),
    last_activity_time_(reactor_thread_->cur_time()),
    is_epoll_registered_(false),
    next_call_id_(1),
    sasl_client_(kSaslAppName, socket),
    sasl_server_(kSaslAppName, socket),
    negotiation_complete_(false) {
}

Status Connection::SetNonBlocking(bool enabled) {
  return socket_.SetNonBlocking(enabled);
}

void Connection::EpollRegister(ev::loop_ref& loop) {
  DCHECK(reactor_thread_->IsCurrentThread());
  DVLOG(4) << "Registering connection for epoll: " << ToString();
  write_io_.set(loop);
  write_io_.set(socket_.GetFd(), ev::WRITE);
  write_io_.set<Connection, &Connection::WriteHandler>(this);
  if (direction_ == CLIENT && negotiation_complete_) {
    write_io_.start();
  }
  read_io_.set(loop);
  read_io_.set(socket_.GetFd(), ev::READ);
  read_io_.set<Connection, &Connection::ReadHandler>(this);
  read_io_.start();
  is_epoll_registered_ = true;
}

Connection::~Connection() {
  // Must clear the outbound_transfers_ list before deleting.
  CHECK(outbound_transfers_.begin() == outbound_transfers_.end());

  // It's crucial that the connection is Shutdown first -- otherwise
  // our destructor will end up calling read_io_.stop() and write_io_.stop()
  // from a possibly non-reactor thread context. This can then make all
  // hell break loose with libev.
  CHECK(!is_epoll_registered_);
}

bool Connection::Idle() const {
  // check if we're in the middle of receiving something
  InboundTransfer *transfer = inbound_.get();
{'fraction_non_alphanumeric': 0.14025356576862125, 'fraction_numerical': 0.0027733755942947703, 'mean_word_length': 4.912177985948477, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 28, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3061247', 'n_tokens_mistral': 1741, 'n_tokens_neox': 1709, 'n_words': 363}