<commit_before>// Copyright (c)2008-2011, Preferred Infrastructure Inc.
// 
// All rights reserved.
// 
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
// 
//     * Redistributions of source code must retain the above copyright
//       notice, this list of conditions and the following disclaimer.
// 
//     * Redistributions in binary form must reproduce the above
//       copyright notice, this list of conditions and the following
//       disclaimer in the documentation and/or other materials provided
//       with the distribution.
// 
//     * Neither the name of Preferred Infrastructure nor the names of other
//       contributors may be used to endorse or promote products derived
//       from this software without specific prior written permission.
// 
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

#include "base.h"

#include "../../lang/shared_ptr.h"
#include "../../concurrent/thread.h"
#include "../../data/serialization.h"

using namespace pfi::lang;
using namespace pfi::concurrent;
using namespace pfi::data::serialization;

using namespace std;

namespace pfi{
namespace network{
namespace rpc{

static const char *PING_MSG = "<<<PING>>>";
static const char *PONG_MSG = "<<<PONG>>>";
static const char *EXIT_MSG = "<<<EXIT>>>";

// TODO: catch exception which std::mutex::mutex() throws
rpc_server::rpc_server(int version, double timeout_sec) : version(version),
                                                          timeout_sec(timeout_sec),
                                                          state_mutex(),
                                                          stop_condition(),
                                                          state(server_state::STOPPED),
                                                          num_running_threads(0),
                                                          threads() {
  port_num = 0;
}

rpc_server::~rpc_server()
{
  for (size_t i = 0; i < threads.size(); ++i) {
    if (threads[i]) {
      threads[i]->join();
    }
  }
  if (socket) {
    socket->close();
    socket.reset();
  }
}

void rpc_server::add(const string &name, const pfi::lang::shared_ptr<invoker_base>& invoker)
{
  funcs[name]=invoker;
}

bool rpc_server::serv(uint16_t port, int nthreads)
{
  if (!start(port, nthreads)) {
    return false;
  }
  wait_until_stopped();
  return true;
}

bool rpc_server::start(uint16_t port, int nthreads)
{
  socket = pfi::lang::shared_ptr<server_socket>(new server_socket());
  if (!socket->create(port))
    return false;

  port_num = socket->port();

  {
    std::lock_guard<std::mutex> lock(state_mutex);
    // NOTE: state should be set as `running` before starting threads
    //       because each thread runs while the state is `running`.
    state = server_state::RUNNING;
    threads = vector<pfi::lang::shared_ptr<thread>>(nthreads);
    for (int i=0; i<nthreads; i++){
      threads[i]=pfi::lang::shared_ptr<thread>(new thread(bind(&rpc_server::process, this, socket)));
      if (!threads[i]->start()) return false;
      ++num_running_threads;
    }
  }
  return true;
}

void rpc_server::stop()
{
  std::lock_guard<std::mutex> lock(state_mutex);
  if (!is_running_unsafe()) {
    return;
  }
  state = server_state::STOPPING;
  if (socket) {
    socket->close();
    for (size_t i = 0; i < threads.size(); ++i) {
      rpc_client client("localhost", port());
      auto request_termination = client.call<bool()>(EXIT_MSG);
      // NOTE: Now that socket is closed,
      //       when there is no server thread which waits at `server::socket::accept()`
      //       this RPC call will throw `cannot_get_connection` exception.
      try {
        request_termination();
      } catch (const cannot_get_connection&) {}
    }
  }
}

void rpc_server::wait_until_stopped()
{
  std::unique_lock<std::mutex> lock(state_mutex);
  if (is_stopped_unsafe()) {
    return;
  }
  stop_condition.wait(lock, [this]() { return state == server_state::STOPPED; });
}

bool rpc_server::is_running() const
{
  std::lock_guard<std::mutex> lock(state_mutex);
  return is_running_unsafe();
}

bool rpc_server::is_stopping() const
{
  std::lock_guard<std::mutex> lock(state_mutex);
  return is_stopping_unsafe();
}

bool rpc_server::is_stopped() const
{
  std::lock_guard<std::mutex> lock(state_mutex);
  return is_stopped_unsafe();
}

uint16_t rpc_server::port() const
{
  return port_num;
}

void rpc_server::process(const pfi::lang::shared_ptr<server_socket>& ssock)
{
  while (is_running()){
    pfi::lang::shared_ptr<stream_socket> sock(ssock->accept());
    if (!sock) continue;
    sock->set_nodelay(true);
    if (timeout_sec > 0.0) {
      if (!sock->set_timeout(timeout_sec)) {
        continue;
      }
    }

    socketstream ss(sock);
    binary_iarchive ia(ss);
    binary_oarchive oa(ss);
    for (;;){
      string name;
      ia>>name;

      // NOTE: When a connection ends (e.g. by destructor of the connected client being called),
      //       the server thread gets empty string from `binary_archive`
      //       and exits this for loop here.
      if (!ss)
        break;

      if (name==PING_MSG){
        string t(PONG_MSG);
        oa<<t;
        oa.flush();
        continue;
      } else if (name==EXIT_MSG) {
        string result_code("OK");
        oa<<result_code;
        oa.flush();
        string ret("1");
        oa<<ret;
        oa.flush();
        break;
      }

      if (funcs.count(name)==0){
        string msg(string("NotFound ")+name);
        oa<<msg;
        oa.flush();
        break;
      }

      int req_ver;
      if (ia >> req_ver && req_ver!=version){
        string msg(string("VersionMismatch ")+lexical_cast<string>(version)+" "+lexical_cast<string>(req_ver));
        oa<<msg;
        oa.flush();
        break;
      }

      try{
        funcs[name]->invoke(ia,oa);
      }
      catch(const rpc_error &){
      }
    }
  }
  notify_if_this_is_the_last_running_threads();
}

bool rpc_server::is_running_unsafe() const
{
  return state == server_state::RUNNING;
}

bool rpc_server::is_stopping_unsafe() const
{
  return state == server_state::STOPPING;
}

bool rpc_server::is_stopped_unsafe() const
{
  return state == server_state::STOPPED;
}

void rpc_server::notify_if_this_is_the_last_running_threads()
{
  {
    std::lock_guard<std::mutex> lock(state_mutex);
    if (--num_running_threads > 0) {
      return;
    }
    state = server_state::STOPPED;
  }
  stop_condition.notify_one();
}

rpc_client::rpc_client(const string &host, uint16_t port, int version)
  :host(host), port(port), version(version)
{
}

rpc_client::~rpc_client()
{
}

pfi::lang::shared_ptr<socketstream> rpc_client::get_connection()
{
  for (int i=0;i<2;i++){
    if (!ss || !(*ss)){
      ss=pfi::lang::shared_ptr<socketstream>(new socketstream(host, port));
      if (!(*ss)){
        ss.reset();
        continue;
      }
      if (!ss->socket()->set_nodelay(true)){
        ss.reset();
        continue;
      }
    }

    binary_iarchive ia(*ss);
    binary_oarchive oa(*ss);
    string ping(PING_MSG);
    oa<<ping;
    oa.flush();
    if (!ss){
      ss.reset();
      continue;
    }
    string pong;
    ia>>pong;
    if (!ss){
      ss.reset();
      continue;
    }
    if (pong!=PONG_MSG){
      ss.reset();
      continue;
    }

    break;
  }

  return ss;
}

void rpc_client::return_connection(const pfi::lang::shared_ptr<socketstream>& css)
{
  ss=css;
}

int rpc_client::get_version()
{
  return version;
}

} // rpc
} // network
} // pfi
<commit_msg>Call rpc_server::stop() when rpc_server::start() fails in rpc_server::serv()<commit_after>// Copyright (c)2008-2011, Preferred Infrastructure Inc.
// 
// All rights reserved.
// 
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are met:
// 
//     * Redistributions of source code must retain the above copyright
//       notice, this list of conditions and the following disclaimer.
// 
//     * Redistributions in binary form must reproduce the above
//       copyright notice, this list of conditions and the following
//       disclaimer in the documentation and/or other materials provided
//       with the distribution.
// 
//     * Neither the name of Preferred Infrastructure nor the names of other
//       contributors may be used to endorse or promote products derived
//       from this software without specific prior written permission.
// 
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

#include "base.h"

#include "../../lang/shared_ptr.h"
#include "../../concurrent/thread.h"
#include "../../data/serialization.h"

using namespace pfi::lang;
using namespace pfi::concurrent;
using namespace pfi::data::serialization;

using namespace std;

namespace pfi{
namespace network{
namespace rpc{

static const char *PING_MSG = "<<<PING>>>";
static const char *PONG_MSG = "<<<PONG>>>";
static const char *EXIT_MSG = "<<<EXIT>>>";

// TODO: catch exception which std::mutex::mutex() throws
rpc_server::rpc_server(int version, double timeout_sec) : version(version),
                                                          timeout_sec(timeout_sec),
                                                          state_mutex(),
                                                          stop_condition(),
                                                          state(server_state::STOPPED),
                                                          num_running_threads(0),
                                                          threads() {
  port_num = 0;
}

rpc_server::~rpc_server()
{
  for (size_t i = 0; i < threads.size(); ++i) {
    if (threads[i]) {
      threads[i]->join();
    }
  }
  if (socket) {
    socket->close();
    socket.reset();
  }
}

void rpc_server::add(const string &name, const pfi::lang::shared_ptr<invoker_base>& invoker)
{
  funcs[name]=invoker;
}

bool rpc_server::serv(uint16_t port, int nthreads)
{
  if (!start(port, nthreads)) {
    stop();
    return false;
  }
  wait_until_stopped();
  return true;
}

bool rpc_server::start(uint16_t port, int nthreads)
{
  socket = pfi::lang::shared_ptr<server_socket>(new server_socket());
  if (!socket->create(port))
    return false;

  port_num = socket->port();

  {
    std::lock_guard<std::mutex> lock(state_mutex);
    // NOTE: state should be set as `running` before starting threads
    //       because each thread runs while the state is `running`.
    state = server_state::RUNNING;
    threads = vector<pfi::lang::shared_ptr<thread>>(nthreads);
    for (int i=0; i<nthreads; i++){
      threads[i]=pfi::lang::shared_ptr<thread>(new thread(bind(&rpc_server::process, this, socket)));
      if (!threads[i]->start()) return false;
      ++num_running_threads;
    }
  }
  return true;
}

void rpc_server::stop()
{
  std::lock_guard<std::mutex> lock(state_mutex);
  if (!is_running_unsafe()) {
    return;
  }
  state = server_state::STOPPING;
  if (socket) {
    socket->close();
    for (size_t i = 0; i < threads.size(); ++i) {
      rpc_client client("localhost", port());
      auto request_termination = client.call<bool()>(EXIT_MSG);
      // NOTE: Now that socket is closed,
      //       when there is no server thread which waits at `server::socket::accept()`
      //       this RPC call will throw `cannot_get_connection` exception.
      try {
        request_termination();
      } catch (const cannot_get_connection&) {}
    }
  }
}

void rpc_server::wait_until_stopped()
{
  std::unique_lock<std::mutex> lock(state_mutex);
  if (is_stopped_unsafe()) {
    return;
  }
  stop_condition.wait(lock, [this]() { return state == server_state::STOPPED; });
}

bool rpc_server::is_running() const
{
  std::lock_guard<std::mutex> lock(state_mutex);
  return is_running_unsafe();
}

bool rpc_server::is_stopping() const
{
  std::lock_guard<std::mutex> lock(state_mutex);
  return is_stopping_unsafe();
}

bool rpc_server::is_stopped() const
{
  std::lock_guard<std::mutex> lock(state_mutex);
  return is_stopped_unsafe();
}

uint16_t rpc_server::port() const
{
  return port_num;
}

void rpc_server::process(const pfi::lang::shared_ptr<server_socket>& ssock)
{
  while (is_running()){
    pfi::lang::shared_ptr<stream_socket> sock(ssock->accept());
    if (!sock) continue;
    sock->set_nodelay(true);
    if (timeout_sec > 0.0) {
      if (!sock->set_timeout(timeout_sec)) {
        continue;
      }
    }

    socketstream ss(sock);
    binary_iarchive ia(ss);
    binary_oarchive oa(ss);
    for (;;){
      string name;
      ia>>name;

      // NOTE: When a connection ends (e.g. by destructor of the connected client being called),
      //       the server thread gets empty string from `binary_archive`
      //       and exits this for loop here.
      if (!ss)
        break;

      if (name==PING_MSG){
        string t(PONG_MSG);
        oa<<t;
        oa.flush();
        continue;
      } else if (name==EXIT_MSG) {
        string result_code("OK");
        oa<<result_code;
        oa.flush();
        string ret("1");
        oa<<ret;
        oa.flush();
        break;
      }

      if (funcs.count(name)==0){
        string msg(string("NotFound ")+name);
        oa<<msg;
        oa.flush();
        break;
      }

      int req_ver;
      if (ia >> req_ver && req_ver!=version){
        string msg(string("VersionMismatch ")+lexical_cast<string>(version)+" "+lexical_cast<string>(req_ver));
        oa<<msg;
        oa.flush();
        break;
      }

      try{
        funcs[name]->invoke(ia,oa);
      }
      catch(const rpc_error &){
      }
    }
  }
  notify_if_this_is_the_last_running_threads();
}

bool rpc_server::is_running_unsafe() const
{
  return state == server_state::RUNNING;
}

bool rpc_server::is_stopping_unsafe() const
{
  return state == server_state::STOPPING;
}

bool rpc_server::is_stopped_unsafe() const
{
  return state == server_state::STOPPED;
}

void rpc_server::notify_if_this_is_the_last_running_threads()
{
  {
    std::lock_guard<std::mutex> lock(state_mutex);
    if (--num_running_threads > 0) {
      return;
    }
    state = server_state::STOPPED;
  }
  stop_condition.notify_one();
}

rpc_client::rpc_client(const string &host, uint16_t port, int version)
  :host(host), port(port), version(version)
{
}

rpc_client::~rpc_client()
{
}

pfi::lang::shared_ptr<socketstream> rpc_client::get_connection()
{
  for (int i=0;i<2;i++){
    if (!ss || !(*ss)){
      ss=pfi::lang::shared_ptr<socketstream>(new socketstream(host, port));
      if (!(*ss)){
        ss.reset();
        continue;
      }
      if (!ss->socket()->set_nodelay(true)){
        ss.reset();
        continue;
      }
    }

    binary_iarchive ia(*ss);
    binary_oarchive oa(*ss);
    string ping(PING_MSG);
    oa<<ping;
    oa.flush();
    if (!ss){
      ss.reset();
      continue;
    }
    string pong;
    ia>>pong;
    if (!ss){
      ss.reset();
      continue;
    }
    if (pong!=PONG_MSG){
      ss.reset();
      continue;
    }

    break;
  }

  return ss;
}

void rpc_client::return_connection(const pfi::lang::shared_ptr<socketstream>& css)
{
  ss=css;
}

int rpc_client::get_version()
{
  return version;
}

} // rpc
} // network
} // pfi

{'fraction_non_alphanumeric': 0.13282042106518757, 'fraction_numerical': 0.003339893839088686, 'mean_word_length': 2.771479982006298, 'pattern_counts': {'":': 0, '<': 93, '<?xml version=': 0, '>': 101, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 24, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4859265', 'n_tokens_mistral': 5353, 'n_tokens_neox': 5148, 'n_words': 1335}