<commit_before>/*
 * Copyright (C) 2015 deipi.com LLC and contributors. All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */

#include "client_binary.h"

#ifdef XAPIAND_CLUSTERING

#include "servers/server.h"
#include "servers/tcp_base.h"
#include "utils.h"
#include "length.h"
#include "io_utils.h"

#include <sysexits.h>
#include <sys/socket.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>



inline std::string::size_type
common_prefix_length(const std::string &a, const std::string &b)
{
	std::string::size_type minlen = std::min(a.size(), b.size());
	std::string::size_type common;
	for (common = 0; common < minlen; ++common) {
		if (a[common] != b[common]) break;
	}
	return common;
}

std::string
serialise_error(const Xapian::Error &exc)
{
	// The byte before the type name is the type code.
	std::string result(1, (exc.get_type())[-1]);
	result += serialise_length(exc.get_context().length());
	result += exc.get_context();
	result += serialise_length(exc.get_msg().length());
	result += exc.get_msg();
	// The "error string" goes last so we don't need to store its length.
	const char * err = exc.get_error_string();
	if (err) result += err;
	return result;
}


#define SWITCH_TO_REPL '\xfe'


//
// Xapian binary client
//

BinaryClient::BinaryClient(std::shared_ptr<BinaryServer> server_, ev::loop_ref *loop_, int sock_, double /*active_timeout_*/, double /*idle_timeout_*/)
	: BaseClient(std::move(server_), loop_, sock_),
	  running(0),
	  state(State::INIT),
	  writable(false),
	  flags(0),
	  database(nullptr)
{
	int binary_clients = ++XapiandServer::binary_clients;
	int total_clients = XapiandServer::total_clients;
	if (binary_clients > total_clients) {
		L_CRIT(this, "Inconsistency in number of binary clients");
		exit(EX_SOFTWARE);
	}

	L_CONN(this, "New Binary Client (sock=%d), %d client(s) of a total of %d connected.", sock, binary_clients, total_clients);

	L_OBJ(this, "CREATED BINARY CLIENT! (%d clients) [%llx]", binary_clients, this);
}


BinaryClient::~BinaryClient()
{
	checkin_database();

	int binary_clients = --XapiandServer::binary_clients;

	L_OBJ(this, "DELETED BINARY CLIENT! (%d clients left) [%llx]", binary_clients, this);
	if (binary_clients < 0) {
		L_CRIT(this, "Inconsistency in number of binary clients");
		exit(EX_SOFTWARE);
	}
}


bool
BinaryClient::init_remote()
{
	L_DEBUG(this, "init_remote");
	state = State::INIT;

	manager()->thread_pool.enqueue(share_this<BinaryClient>());
	return true;
}


bool
BinaryClient::init_replication(const Endpoint &src_endpoint, const Endpoint &dst_endpoint)
{
	L_REPLICATION(this, "init_replication: %s  -->  %s", src_endpoint.as_string().c_str(), dst_endpoint.as_string().c_str());
	state = State::REPLICATIONPROTOCOL_CLIENT;

	repl_endpoints.add(src_endpoint);
	endpoints.add(dst_endpoint);

	writable = true;

	if (!manager()->database_pool.checkout(database, endpoints, DB_WRITABLE | DB_SPAWN | DB_REPLICATION, [
		manager=manager(),
		src_endpoint,
		dst_endpoint
	] {
		L_DEBUG(manager.get(), "Triggering replication for %s after checkin!", dst_endpoint.as_string().c_str());
		manager->trigger_replication(src_endpoint, dst_endpoint);
	})) {
		L_ERR(this, "Cannot checkout %s", endpoints.as_string().c_str());
		return false;
	}

	int port = (src_endpoint.port == XAPIAND_BINARY_SERVERPORT) ? XAPIAND_BINARY_PROXY : src_endpoint.port;
<commit_msg>Release manager once the lambda no longer uses it<commit_after>/*
 * Copyright (C) 2015 deipi.com LLC and contributors. All rights reserved.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 */

#include "client_binary.h"

#ifdef XAPIAND_CLUSTERING

#include "servers/server.h"
#include "servers/tcp_base.h"
#include "utils.h"
#include "length.h"
#include "io_utils.h"

#include <sysexits.h>
#include <sys/socket.h>
#include <fcntl.h>
#include <unistd.h>
#include <sys/stat.h>



inline std::string::size_type
common_prefix_length(const std::string &a, const std::string &b)
{
	std::string::size_type minlen = std::min(a.size(), b.size());
	std::string::size_type common;
	for (common = 0; common < minlen; ++common) {
		if (a[common] != b[common]) break;
	}
	return common;
}

std::string
serialise_error(const Xapian::Error &exc)
{
	// The byte before the type name is the type code.
	std::string result(1, (exc.get_type())[-1]);
	result += serialise_length(exc.get_context().length());
	result += exc.get_context();
	result += serialise_length(exc.get_msg().length());
	result += exc.get_msg();
	// The "error string" goes last so we don't need to store its length.
	const char * err = exc.get_error_string();
	if (err) result += err;
	return result;
}


#define SWITCH_TO_REPL '\xfe'


//
// Xapian binary client
//

BinaryClient::BinaryClient(std::shared_ptr<BinaryServer> server_, ev::loop_ref *loop_, int sock_, double /*active_timeout_*/, double /*idle_timeout_*/)
	: BaseClient(std::move(server_), loop_, sock_),
	  running(0),
	  state(State::INIT),
	  writable(false),
	  flags(0),
	  database(nullptr)
{
	int binary_clients = ++XapiandServer::binary_clients;
	int total_clients = XapiandServer::total_clients;
	if (binary_clients > total_clients) {
		L_CRIT(this, "Inconsistency in number of binary clients");
		exit(EX_SOFTWARE);
	}

	L_CONN(this, "New Binary Client (sock=%d), %d client(s) of a total of %d connected.", sock, binary_clients, total_clients);

	L_OBJ(this, "CREATED BINARY CLIENT! (%d clients) [%llx]", binary_clients, this);
}


BinaryClient::~BinaryClient()
{
	checkin_database();

	int binary_clients = --XapiandServer::binary_clients;

	L_OBJ(this, "DELETED BINARY CLIENT! (%d clients left) [%llx]", binary_clients, this);
	if (binary_clients < 0) {
		L_CRIT(this, "Inconsistency in number of binary clients");
		exit(EX_SOFTWARE);
	}
}


bool
BinaryClient::init_remote()
{
	L_DEBUG(this, "init_remote");
	state = State::INIT;

	manager()->thread_pool.enqueue(share_this<BinaryClient>());
	return true;
}


bool
BinaryClient::init_replication(const Endpoint &src_endpoint, const Endpoint &dst_endpoint)
{
	L_REPLICATION(this, "init_replication: %s  -->  %s", src_endpoint.as_string().c_str(), dst_endpoint.as_string().c_str());
	state = State::REPLICATIONPROTOCOL_CLIENT;

	repl_endpoints.add(src_endpoint);
	endpoints.add(dst_endpoint);

	writable = true;

	if (!manager()->database_pool.checkout(database, endpoints, DB_WRITABLE | DB_SPAWN | DB_REPLICATION, [
		manager=manager(),
		src_endpoint,
		dst_endpoint
	] () mutable {
		L_DEBUG(manager.get(), "Triggering replication for %s after checkin!", dst_endpoint.as_string().c_str());
		manager->trigger_replication(src_endpoint, dst_endpoint);
		manager.reset();
	})) {
		L_ERR(this, "Cannot checkout %s", endpoints.as_string().c_str());
		return false;
	}

	int port = (src_endpoint.port == XAPIAND_BINARY_SERVERPORT) ? XAPIAND_BINARY_PROXY : src_endpoint.port;

{'fraction_non_alphanumeric': 0.12761647535449022, 'fraction_numerical': 0.0022507314877335135, 'mean_word_length': 5.254046446164673, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 27, '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': '769589', 'n_tokens_mistral': 3021, 'n_tokens_neox': 2895, 'n_words': 858}