<commit_before>/*
 * Copyright (c) 2011-2012 by <NAME>, Zuse Institute Berlin
 *               2010-2011 by <NAME>, Zuse Institute Berlin
 *
 *
 * Licensed under the BSD License, see LICENSE file for details.
 *
 */

#include "libxtreemfs/client_implementation.h"

#include <cstdlib>

#include <boost/bind.hpp>
#include <boost/thread/thread.hpp>
#include <list>
#include <string>

#include "libxtreemfs/async_write_handler.h"
#include "libxtreemfs/execute_sync_request.h"
#include "libxtreemfs/helper.h"
#include "libxtreemfs/options.h"
#include "libxtreemfs/pbrpc_url.h"
#include "libxtreemfs/uuid_iterator.h"
#include "libxtreemfs/volume_implementation.h"
#include "libxtreemfs/xtreemfs_exception.h"
#include "util/error_log.h"
#include "xtreemfs/DIRServiceClient.h"
#include "xtreemfs/MRCServiceClient.h"
#include "xtreemfs/OSDServiceClient.h"

using namespace std;
using namespace xtreemfs::pbrpc;
using namespace xtreemfs::util;

namespace xtreemfs {

ClientImplementation::ClientImplementation(
    const std::string& dir_service_address,
    const xtreemfs::pbrpc::UserCredentials& user_credentials,
    const xtreemfs::rpc::SSLOptions* ssl_options,
    const Options& options)
  : was_shutdown_(false),
    dir_service_auth_(),
    dir_service_user_credentials_(user_credentials),
    dir_service_ssl_options_(ssl_options),
    options_(options),
    list_open_volumes_(),
    uuid_cache_(),
    network_client_(NULL),
    network_client_thread_(NULL),
    dir_service_client_(NULL),
    client_uuid_("") {
  dir_service_addresses.AddUUID(dir_service_address);

  // Set bogus auth object.
  auth_bogus_.set_auth_type(AUTH_NONE);

  // Currently no AUTH is needed to access the DIR.
  dir_service_auth_.set_auth_type(AUTH_NONE);

  initialize_logger(options.log_level_string,
                    options.log_file_path,
                    LEVEL_WARN);
  initialize_error_log(20);

  if (Logging::log->loggingActive(LEVEL_DEBUG)) {
    Logging::log->getLog(LEVEL_DEBUG) << "Created a new libxtreemfs Client "
        "object (version " << options.version_string << ")" << endl;
  }
}

ClientImplementation::~ClientImplementation() {
  Shutdown();
  if (!list_open_volumes_.empty()) {
    string error = "Client::~Client(): Not all XtreemFS volumes were closed."
        " Did you forget to call Client::Shutdown()? Memory leaks are the"
        " consequence.";
    Logging::log->getLog(LEVEL_ERROR) << error << endl;
    ErrorLog::error_log->AppendError(error);
  }

  // Since we wait for outstanding requests, the RPC client (network_client_)
  // has to shutdown first and then we can wait for the Vivaldi thread.
  // The other way around a deadlock might occur.
  if (vivaldi_thread_.get() && vivaldi_thread_->joinable()) {
    vivaldi_thread_->join();
  }

  atexit(google::protobuf::ShutdownProtobufLibrary);

  shutdown_logger();
  shutdown_error_log();
}

void ClientImplementation::Start() {
  // start network (rpc) client
  network_client_.reset(new xtreemfs::rpc::Client(
      options_.connect_timeout_s,
      options_.request_timeout_s,
      options_.linger_timeout_s,
      dir_service_ssl_options_));

  network_client_thread_.reset(
      new boost::thread(boost::bind(&xtreemfs::rpc::Client::run,
                                    network_client_.get())));

  dir_service_client_.reset(new DIRServiceClient(network_client_.get()));

  GenerateVersion4UUID(&client_uuid_);
  assert(!client_uuid_.empty());

  // Start vivaldi thread if configured
  if (options_.vivaldi_enable) {
    if (Logging::log->loggingActive(LEVEL_INFO)) {
      Logging::log->getLog(LEVEL_INFO)
          << "Starting vivaldi..." << endl;
    }
    vivaldi_.reset(new Vivaldi(network_client_.get(),
                               dir_service_client_.get(),
                               &dir_service_addresses,
                               this->GetUUIDResolver(),
                               options_));
    vivaldi_thread_.reset(new boost::thread(boost::bind(&xtreemfs::Vivaldi::Run,
                                                        vivaldi_.get())));
  }

  async_write_callback_thread_.reset(new boost::thread(&xtreemfs::AsyncWriteHandler::ProcessCallbacks));
}

void ClientImplementation::Shutdown() {
  boost::mutex::scoped_lock lock(list_open_volumes_mutex_);

  // Issue Close() on every Volume and remove it's pointer.
  list<VolumeImplementation*>::iterator it;
  while (!list_open_volumes_.empty()) {
    it = list_open_volumes_.begin();
    (*it)->CloseInternal();
    delete *it;
    it = list_open_volumes_.erase(it);
  }

  if (async_write_callback_thread_->joinable()) {
    async_write_callback_thread_->interrupt();
    async_write_callback_thread_->join();
  }

  // Stop vivaldi thread if running
  if (vivaldi_thread_.get() && vivaldi_thread_->joinable()) {
    vivaldi_thread_->interrupt();
  }

}

Volume* ClientImplementation::OpenVolume(
    const std::string& volume_name,
    const xtreemfs::rpc::SSLOptions* ssl_options,
    const Options& options) {
  // TODO(mberlin): Fix possible leak through the use of scoped_ptr and swap().
  SimpleUUIDIterator* mrc_uuid_iterator = new SimpleUUIDIterator;
  VolumeNameToMRCUUID(volume_name, mrc_uuid_iterator);

  VolumeImplementation* volume = new VolumeImplementation(
      this,
      client_uuid_,
      mrc_uuid_iterator,
      volume_name,
      ssl_options,
      options);
  {
    boost::mutex::scoped_lock lock(list_open_volumes_mutex_);
    list_open_volumes_.push_back(volume);
  }
  volume->Start();
<commit_msg>client: fixed falsely merged client_implemenation.cpp.<commit_after>/*
 * Copyright (c) 2011-2012 by <NAME>, Zuse Institute Berlin
 *               2010-2011 by <NAME>, Zuse Institute Berlin
 *
 *
 * Licensed under the BSD License, see LICENSE file for details.
 *
 */

#include "libxtreemfs/client_implementation.h"

#include <cstdlib>

#include <boost/bind.hpp>
#include <boost/thread/thread.hpp>
#include <list>
#include <string>

#include "libxtreemfs/async_write_handler.h"
#include "libxtreemfs/execute_sync_request.h"
#include "libxtreemfs/helper.h"
#include "libxtreemfs/options.h"
#include "libxtreemfs/pbrpc_url.h"
#include "libxtreemfs/uuid_iterator.h"
#include "libxtreemfs/volume_implementation.h"
#include "libxtreemfs/xtreemfs_exception.h"
#include "util/error_log.h"
#include "xtreemfs/DIRServiceClient.h"
#include "xtreemfs/MRCServiceClient.h"
#include "xtreemfs/OSDServiceClient.h"

using namespace std;
using namespace xtreemfs::pbrpc;
using namespace xtreemfs::util;

namespace xtreemfs {

ClientImplementation::ClientImplementation(
    const std::string& dir_service_address,
    const xtreemfs::pbrpc::UserCredentials& user_credentials,
    const xtreemfs::rpc::SSLOptions* ssl_options,
    const Options& options)
  : was_shutdown_(false),
    dir_service_auth_(),
    dir_service_user_credentials_(user_credentials),
    dir_service_ssl_options_(ssl_options),
    options_(options),
    list_open_volumes_(),
    uuid_cache_(),
    network_client_(NULL),
    network_client_thread_(NULL),
    dir_service_client_(NULL),
    client_uuid_("") {
  dir_service_addresses.AddUUID(dir_service_address);

  // Set bogus auth object.
  auth_bogus_.set_auth_type(AUTH_NONE);

  // Currently no AUTH is needed to access the DIR.
  dir_service_auth_.set_auth_type(AUTH_NONE);

  initialize_logger(options.log_level_string,
                    options.log_file_path,
                    LEVEL_WARN);
  initialize_error_log(20);

  if (Logging::log->loggingActive(LEVEL_DEBUG)) {
    Logging::log->getLog(LEVEL_DEBUG) << "Created a new libxtreemfs Client "
        "object (version " << options.version_string << ")" << endl;
  }
}

ClientImplementation::~ClientImplementation() {
  Shutdown();
  if (!list_open_volumes_.empty()) {
    string error = "Client::~Client(): Not all XtreemFS volumes were closed."
        " Did you forget to call Client::Shutdown()? Memory leaks are the"
        " consequence.";
    Logging::log->getLog(LEVEL_ERROR) << error << endl;
    ErrorLog::error_log->AppendError(error);
  }

  network_client_->shutdown();
  network_client_thread_->join();

  // Since we wait for outstanding requests, the RPC client (network_client_)
  // has to shutdown first and then we can wait for the Vivaldi thread.
  // The other way around a deadlock might occur.
  if (vivaldi_thread_.get() && vivaldi_thread_->joinable()) {
    vivaldi_thread_->join();
  }

  atexit(google::protobuf::ShutdownProtobufLibrary);

  shutdown_logger();
  shutdown_error_log();
}

void ClientImplementation::Start() {
  // start network (rpc) client
  network_client_.reset(new xtreemfs::rpc::Client(
      options_.connect_timeout_s,
      options_.request_timeout_s,
      options_.linger_timeout_s,
      dir_service_ssl_options_));

  network_client_thread_.reset(
      new boost::thread(boost::bind(&xtreemfs::rpc::Client::run,
                                    network_client_.get())));

  dir_service_client_.reset(new DIRServiceClient(network_client_.get()));

  GenerateVersion4UUID(&client_uuid_);
  assert(!client_uuid_.empty());

  // Start vivaldi thread if configured
  if (options_.vivaldi_enable) {
    if (Logging::log->loggingActive(LEVEL_INFO)) {
      Logging::log->getLog(LEVEL_INFO)
          << "Starting vivaldi..." << endl;
    }
    vivaldi_.reset(new Vivaldi(network_client_.get(),
                               dir_service_client_.get(),
                               &dir_service_addresses,
                               this->GetUUIDResolver(),
                               options_));
    vivaldi_thread_.reset(new boost::thread(boost::bind(&xtreemfs::Vivaldi::Run,
                                                        vivaldi_.get())));
  }

  async_write_callback_thread_.reset(new boost::thread(&xtreemfs::AsyncWriteHandler::ProcessCallbacks));
}

void ClientImplementation::Shutdown() {
  if (!was_shutdown_) {
    was_shutdown_ = true;
    boost::mutex::scoped_lock lock(list_open_volumes_mutex_);

    // Issue Close() on every Volume and remove it's pointer.
    list<VolumeImplementation*>::iterator it;
    while (!list_open_volumes_.empty()) {
      it = list_open_volumes_.begin();
      (*it)->CloseInternal();
      delete *it;
      it = list_open_volumes_.erase(it);
    }

    if (async_write_callback_thread_->joinable()) {
      async_write_callback_thread_->interrupt();
      async_write_callback_thread_->join();
    }

    // Stop vivaldi thread if running
    if (vivaldi_thread_.get() && vivaldi_thread_->joinable()) {
      vivaldi_thread_->interrupt();
    }
  }
}

Volume* ClientImplementation::OpenVolume(
    const std::string& volume_name,
    const xtreemfs::rpc::SSLOptions* ssl_options,
    const Options& options) {
  // TODO(mberlin): Fix possible leak through the use of scoped_ptr and swap().
  SimpleUUIDIterator* mrc_uuid_iterator = new SimpleUUIDIterator;
  VolumeNameToMRCUUID(volume_name, mrc_uuid_iterator);

  VolumeImplementation* volume = new VolumeImplementation(
      this,
      client_uuid_,
      mrc_uuid_iterator,
      volume_name,
      ssl_options,
      options);
  {
    boost::mutex::scoped_lock lock(list_open_volumes_mutex_);
    list_open_volumes_.push_back(volume);
  }
  volume->Start();

{'fraction_non_alphanumeric': 0.11677660902121591, 'fraction_numerical': 0.0033874130861116063, 'mean_word_length': 3.778109028960818, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6123006', 'n_tokens_mistral': 3602, 'n_tokens_neox': 3510, 'n_words': 644}