start work: callback_result_data, basic handler

Author: sabramkin

bindings/cpp/CMakeLists.txt

@@ -39,6 +39,7 @@ set(ELLIPTICS_CLIENT_SRCS
     ../../library/request_queue.cpp
     ../../library/rbtree.c
     ../../library/trans.c
+    ../../library/trans.cpp
     ../../library/tests.c
     ../../library/common.cpp
     ../../library/access_context.cpp

bindings/cpp/async_result.cpp

@@ -199,7 +199,7 @@ bool async_result<T>::get(T &entry)
 {
 	wait(session::throw_at_get);
 	for (auto it = m_data->results.begin(); it != m_data->results.end(); ++it) {
-		if (it->status() == 0 && !it->data().empty()) {
+		if (it->is_valid() && it->status() == 0) {
 			entry = *it;
 			return true;
 		}

bindings/cpp/callback.cpp

@@ -40,24 +40,42 @@ class basic_handler
 		return 0;
 	}
 
+	// Used by old (protocol-dependent) mechanic, must be removed after refactoring
 	basic_handler(std::unique_ptr<dnet_logger> logger, async_generic_result &result) :
 		m_logger(std::move(logger)),
 		m_handler(result), m_completed(0), m_total(0)
 	{
+		memset(&m_addr, 0, sizeof(dnet_addr));
+		memset(&m_cmd, 0, sizeof(dnet_cmd));
 	}
 
-	bool handle(dnet_addr *addr, dnet_cmd *cmd)
+	basic_handler(const dnet_addr &addr, const dnet_cmd &cmd,
+	              std::unique_ptr<dnet_logger> logger, async_generic_result &result) :
+		m_addr(addr),
+		m_cmd(cmd),
+		m_logger(std::move(logger)),
+		m_handler(result), m_completed(0), m_total(0)
 	{
-		if (is_trans_destroyed(cmd)) {
-			return increment_completed();
-		}
+	}
 
+	void log_reply_info(dnet_addr *addr, dnet_cmd *cmd)
+	{
 		DNET_LOG(m_logger, cmd->status ? DNET_LOG_ERROR : DNET_LOG_NOTICE, "{}: {}: handled reply from: {}, "
 		                                                                   "trans: {}, cflags: {}, status: {}, "
 		                                                                   "size: {}, client: {}, last: {}",
 		         dnet_dump_id(&cmd->id), dnet_cmd_string(cmd->cmd), addr ? dnet_addr_string(addr) : "<unknown>",
 		         cmd->trans, dnet_flags_dump_cflags(cmd->flags), int(cmd->status), cmd->size,
 		         !(cmd->flags & DNET_FLAGS_REPLY), !(cmd->flags & DNET_FLAGS_MORE));
+	}
+
+	// Used by old (protocol-dependent) mechanic, must be removed after refactoring
+	bool handle(dnet_addr *addr, dnet_cmd *cmd)
+	{
+		if (is_trans_destroyed(cmd)) {
+			return increment_completed();
+		}
+
+		log_reply_info(addr, cmd);
 
 		auto data = std::make_shared<callback_result_data>(addr, cmd);
 
@@ -71,6 +89,29 @@ class basic_handler
 		return false;
 	}
 
+	int on_reply(const std::shared_ptr<void> &result)
+	{
+		// TODO(sabramkin): Output only protocol-independent known info (currently old-mechanic logging used)
+		log_reply_info(&m_addr, &m_cmd);
+
+		auto data = std::make_shared<n2_callback_result_data>(m_addr, m_cmd, result, 0);
+		callback_result_entry entry(data);
+		m_handler.process(entry);
+		return 0;
+	}
+
+	int on_reply_error(int err)
+	{
+		// TODO(sabramkin): Output only protocol-independent known info (currently old-mechanic logging used)
+		log_reply_info(&m_addr, &m_cmd);
+
+		auto data = std::make_shared<n2_callback_result_data>(m_addr, m_cmd, nullptr, err);
+		data->error = create_error(err, "n2 lookup_new error"); // TODO(sabramkin): rework error
+		callback_result_entry entry(data);
+		m_handler.process(entry);
+		return 0;
+	}
+
 	// how many independent transactions share this handler plus call below
 	// call below and corresponding +1 is needed, since transactions can be completed
 	// before send_impl() calls this method to setup this 'reference counter'
@@ -92,6 +133,8 @@ class basic_handler
 		return false;
 	}
 
+	dnet_addr m_addr;
+	dnet_cmd m_cmd;
 	std::unique_ptr<dnet_logger> m_logger;
 	async_result_handler<callback_result_entry> m_handler;
 	std::atomic_size_t m_completed;
@@ -100,13 +143,20 @@ class basic_handler
 
 } // namespace detail
 
+const dnet_addr &get_destination_addr(session &sess)
+{
+	const auto fwd = sess.get_forward();
+	if (fwd.is_valid())
+		return fwd.to_raw();
+	return sess.get_direct_address().to_raw();
+}
+
 template <typename Method, typename T>
 async_generic_result send_impl(session &sess, T &control, Method method)
 {
 	async_generic_result result(sess);
 
 	detail::basic_handler *handler = new detail::basic_handler(sess.get_logger(), result);
-
 	control.complete = detail::basic_handler::handler;
 	control.priv = handler;
 
@@ -142,6 +192,42 @@ async_generic_result send_to_single_state(session &sess, dnet_io_control &contro
 	return send_impl(sess, control, send_to_single_state_io_impl);
 }
 
+template <typename Method>
+async_generic_result n2_send_impl(session &sess, const n2_request &request, Method method)
+{
+	async_generic_result result(sess);
+
+	auto handler = std::make_shared<detail::basic_handler>(get_destination_addr(sess), request.cmd,
+	                                                       sess.get_logger(), result);
+	n2_request_info request_info{ request, n2_repliers() };
+
+	request_info.repliers.on_reply =
+		[handler](const std::shared_ptr<void> &result){
+			return handler->on_reply(result);
+		};
+	request_info.repliers.on_reply_error =
+		[handler](int err){
+			return handler->on_reply_error(err);
+		};
+
+	const size_t count = method(sess, std::move(request_info));
+	handler->set_total(count);
+	return result;
+}
+
+int n2_trans_alloc_send(dnet_session *s, n2_request_info &&request_info); // implemented in trans.cpp
+
+static size_t n2_send_to_single_state_impl(session &sess, n2_request_info &&request_info)
+{
+	n2_trans_alloc_send(sess.get_native(), std::move(request_info));
+	return 1;
+}
+
+async_generic_result n2_send_to_single_state(session &sess, const n2_request &request)
+{
+	return n2_send_impl(sess, request, n2_send_to_single_state_impl);
+}
+
 static size_t send_to_each_backend_impl(session &sess, dnet_trans_control &ctl)
 {
 	return dnet_request_cmd(sess.get_native(), &ctl);

bindings/cpp/callback_p.h

@@ -30,6 +30,27 @@
 #include <thread>
 
 #include "elliptics/async_result_cast.hpp"
+#include "library/n2_protocol.hpp"
+
+/*
+ * This macroses should be used surrounding all entry::methods which work directly
+ * with m_data or data() to ensure that meaningful exceptions are thrown
+ */
+#define DNET_DATA_BEGIN_2() try { \
+	do {} while (false)
+
+#define DNET_DATA_END_2(SIZE) \
+	} catch (not_found_error &) { \
+		if (!is_valid()) { \
+			throw_error(-ENOENT, "entry::%s(): entry is null", __FUNCTION__); \
+		} else {\
+			dnet_cmd *cmd = command(); \
+			throw_error(-ENOENT, cmd->id, "entry::%s(): data.size is too small, expected: %zu, actual: %zu, status: %d", \
+				__FUNCTION__, size_t(SIZE), data.size(), cmd->status); \
+		} \
+		throw; \
+	} \
+	do {} while (false)
 
 namespace ioremap { namespace elliptics {
 
@@ -63,7 +84,25 @@ class session_scope
 		uint32_t m_policy;
 };
 
-class callback_result_data
+// TODO(sabramkin): This abstraction is temporary and used while refactoring in progress.
+// TODO(sabramkin): After refactoring only n2_callback_result_data should stay, no base is needed.
+class callback_result_data_base
+{
+	public:
+		virtual dnet_addr *address() const = 0;
+		virtual dnet_cmd *command() const = 0;
+		virtual int status() const = 0;
+		virtual bool is_valid() const = 0;
+		virtual bool is_ack() const = 0;
+		virtual bool is_final() const = 0;
+		virtual bool is_client() const = 0;
+
+		virtual ~callback_result_data_base() = default;
+
+		error_info error;
+};
+
+class callback_result_data : public callback_result_data_base
 {
 	public:
 		callback_result_data()
@@ -81,12 +120,112 @@ class callback_result_data
 			memcpy(data.data<char>() + sizeof(dnet_addr), cmd, sizeof(dnet_cmd) + cmd->size);
 		}
 
-		virtual ~callback_result_data()
+		dnet_addr *address() const override
 		{
+			DNET_DATA_BEGIN_2();
+			return data
+				.data<dnet_addr>();
+			DNET_DATA_END_2(0);
+		}
+
+		dnet_cmd *command() const override
+		{
+			DNET_DATA_BEGIN_2();
+			return data
+				.skip<dnet_addr>()
+				.data<dnet_cmd>();
+			DNET_DATA_END_2(0);
+		}
+
+		int status() const override
+		{
+			return command()->status;
+		}
+
+		bool is_valid() const override
+		{
+			return !data.empty();
+		}
+
+		bool is_ack() const override
+		{
+			return status() == 0 && data.empty();
+		}
+
+		bool is_final() const override
+		{
+			return !(command()->flags & DNET_FLAGS_MORE);
+		}
+
+		bool is_client() const override
+		{
+			return !(command()->flags & DNET_FLAGS_REPLY);
 		}
 
 		data_pointer data;
-		error_info error;
+};
+
+class n2_callback_result_data : public callback_result_data_base
+{
+	public:
+		n2_callback_result_data() = default;
+
+		n2_callback_result_data(const dnet_addr &addr_in, const dnet_cmd &cmd_in,
+		                        const std::shared_ptr<void> &result_in, int result_status_in)
+		: addr(addr_in)
+		, cmd(cmd_in)
+		, result(result_in)
+		, result_status(result_status_in)
+		{
+			// TODO(sabramkin):
+			// Here is emulated protocol logic for single-response commands. It is hardcode that we must
+			// resolve when we introduce bulk commands. See also is_final() method. Note that protocol
+			// mustn't provide its inner structures (such as dnet_cmd), so we must remove command() method
+			// in the future, and must remove cmd member.
+			cmd.flags = (cmd.flags & ~(DNET_FLAGS_NEED_ACK)) | DNET_FLAGS_REPLY;
+		}
+
+		dnet_addr *address() const override
+		{
+			return const_cast<dnet_addr *>(&addr);
+		}
+
+		dnet_cmd *command() const override
+		{
+			return const_cast<dnet_cmd *>(&cmd);
+		}
+
+		int status() const override
+		{
+			return result_status;
+		}
+
+		bool is_valid() const override
+		{
+			return true;
+		}
+
+		bool is_ack() const override
+		{
+			return result_status == 0 && !result;
+		}
+
+		bool is_final() const override
+		{
+			return true; // TODO(sabramkin): rework when any bulk command is supported
+		}
+
+		bool is_client() const override
+		{
+			return false;
+		}
+
+		dnet_addr addr;
+		dnet_cmd cmd;
+
+		// Either result or nonzero result_status must be set
+		std::shared_ptr<void> result;
+		int result_status;
 };
 
 struct dnet_net_state_deleter
@@ -103,6 +242,7 @@ typedef std::unique_ptr<dnet_net_state, dnet_net_state_deleter> net_state_ptr;
 // Send request to specific state
 async_generic_result send_to_single_state(session &sess, const transport_control &control);
 async_generic_result send_to_single_state(session &sess, dnet_io_control &control);
+async_generic_result n2_send_to_single_state(session &sess, const n2_request &request);
 
 // Send request to each backend
 async_generic_result send_to_each_backend(session &sess, const transport_control &control);

bindings/cpp/newapi/result_entry.cpp

@@ -1,4 +1,6 @@
 #include "elliptics/newapi/result_entry.hpp"
+#include "bindings/cpp/callback_p.h"
+#include "library/n2_protocol.hpp"
 #include "library/protocol.hpp"
 #include "library/elliptics.h"
 
@@ -20,20 +22,21 @@ bool callback_result_entry::empty() const {
     return raw_data().empty();
 }
 
-std::string lookup_result_entry::path() const {
-	dnet_lookup_response response;
+lookup_result_entry::lookup_result_entry()
+: ioremap::elliptics::callback_result_entry(std::make_shared<ioremap::elliptics::n2_callback_result_data>()) {
+}
 
-	deserialize(raw_data(), response);
+std::string lookup_result_entry::path() const {
+	auto &response = *static_cast<ioremap::elliptics::n2::lookup_response *>(message());
 	return response.path;
 }
 
 dnet_record_info lookup_result_entry::record_info() const {
+	auto &response = *static_cast<ioremap::elliptics::n2::lookup_response *>(message());
+
 	dnet_record_info info;
 	memset(&info, 0, sizeof(info));
 
-	dnet_lookup_response response;
-	deserialize(raw_data(), response);
-
 	info.record_flags = response.record_flags;
 	info.user_flags = response.user_flags;
 
@@ -61,16 +64,12 @@ lookup_result_entry::checksum_t convert_checksum(const std::vector<unsigned char
 }
 
 lookup_result_entry::checksum_t lookup_result_entry::json_checksum() const {
-	dnet_lookup_response response;
-	deserialize(raw_data(), response);
-
+	auto &response = *static_cast<ioremap::elliptics::n2::lookup_response *>(message());
 	return convert_checksum(response.json_checksum);
 }
 
 lookup_result_entry::checksum_t lookup_result_entry::data_checksum() const {
-	dnet_lookup_response response;
-	deserialize(raw_data(), response);
-
+	auto &response = *static_cast<ioremap::elliptics::n2::lookup_response *>(message());
 	return convert_checksum(response.data_checksum);
 }