merge recovery: optimization: use copy-iterator & server-send for small/medium keys

bindings/python/elliptics_python.cpp

@@ -44,11 +44,13 @@ enum elliptics_iterator_types {
 };
 
 enum elliptics_iterator_flags {
-	iflag_default	= 0,
-	iflag_data	= DNET_IFLAGS_DATA,
-	iflag_key_range	= DNET_IFLAGS_KEY_RANGE,
-	iflag_ts_range	= DNET_IFLAGS_TS_RANGE,
-	iflag_no_meta	= DNET_IFLAGS_NO_META,
+	iflag_default		= 0,
+	iflag_data		= DNET_IFLAGS_DATA,
+	iflag_key_range		= DNET_IFLAGS_KEY_RANGE,
+	iflag_ts_range		= DNET_IFLAGS_TS_RANGE,
+	iflag_no_meta		= DNET_IFLAGS_NO_META,
+	iflags_move		= DNET_IFLAGS_MOVE,
+	iflags_overwrite	= DNET_IFLAGS_OVERWRITE
 };
 
 enum elliptics_cflags {
@@ -377,13 +379,20 @@ BOOST_PYTHON_MODULE(core)
 	    "default\n    There no filtering should be while iteration. All keys will be presented\n"
 	    "data\n    Iteration results should also includes objects datas\n"
 	    "key_range\n    elliptics.Id ranges should be used for filtering keys on the node while iteration\n"
-	    "ts_range\n    Time range should be used for filtering keys on the node while iteration"
-	    "no_meta\n    Iteration results will have empty key's metadata (user_flags and timestamp)")
+	    "ts_range\n    Time range should be used for filtering keys on the node while iteration\n"
+	    "no_meta\n    Iteration results will have empty key's metadata (user_flags and timestamp)\n"
+	    "move\n    Server-send iterator should move data not copy. This will force iterator/server-send logic\n"
+                  "    to queue REMOVE command locally if remote write has succeeeded.\n"
+	    "overwrite\n    Overwrite data. If this flag is NOT set, we only write data if remote timestamp is less\n"
+	               "    than in data being written. When NOT set, data will still be transferred over the network,\n"
+		       "    even if remote timestamp doesn't allow us to overwrite data.")
 		.value("default", iflag_default)
 		.value("data", iflag_data)
 		.value("key_range", iflag_key_range)
 		.value("ts_range", iflag_ts_range)
 		.value("no_meta", iflag_no_meta)
+		.value("move", iflags_move)
+		.value("overwrite", iflags_overwrite)
 	;
 
 	bp::enum_<elliptics_iterator_types>("iterator_types",

bindings/python/elliptics_session.cpp

@@ -502,6 +502,17 @@ class elliptics_session: public session, public bp::wrapper<session> {
 		return create_result(std::move(session::start_iterator(transform(id).id(), std_ranges, type, flags, time_begin.m_time, time_end.m_time)));
 	}
 
+	python_iterator_result start_copy_iterator(const bp::api::object &id, const bp::api::object &ranges,
+						   const bp::api::object &dst_groups,
+						   uint64_t flags,
+						   const elliptics_time& time_begin = elliptics_time(0, 0),
+						   const elliptics_time& time_end = elliptics_time(-1, -1)) {
+		auto std_ranges = convert_to_vector<dnet_iterator_range>(ranges);
+		auto std_dst_groups = convert_to_vector<int>(dst_groups);
+
+		return create_result(std::move(session::start_copy_iterator(transform(id).id(), std_ranges, flags, time_begin.m_time, time_end.m_time, std_dst_groups)));
+	}
+
 	python_iterator_result pause_iterator(const bp::api::object &id, const uint64_t &iterator_id) {
 		return create_result(std::move(session::pause_iterator(transform(id).id(), iterator_id)));
 	}
@@ -514,6 +525,18 @@ class elliptics_session: public session, public bp::wrapper<session> {
 		return create_result(std::move(session::cancel_iterator(transform(id).id(), iterator_id)));
 	}
 
+	python_iterator_result server_send(const bp::api::object &keys, uint64_t iflags, const bp::api::object &groups) {
+		auto std_groups = convert_to_vector<int>(groups);
+		std::vector<dnet_raw_id> std_keys;
+		std_keys.reserve(bp::len(keys));
+
+		for (bp::stl_input_iterator<bp::api::object> it(keys), end; it != end; ++it) {
+			std_keys.push_back(transform(*it).raw_id());
+		}
+
+		return create_result(std::move(session::server_send(std_keys, iflags, std_groups)));
+	}
+
 	python_exec_result exec(const bp::api::object &id_or_context, const std::string &event, const bp::api::object &data, const int src_key) {
 		dnet_id* raw_id = NULL;
 		dnet_id conv_id;
@@ -1521,6 +1544,41 @@ void init_elliptics_session() {
 		    "                       result.response.timestamp.tnsec,\n"
 		    "                       result.response_data))\n")
 
+		.def("start_copy_iterator", &elliptics_session::start_copy_iterator,
+		     bp::args("id", "ranges", "dst_groups", "flags", "time_begin", "time_end"),
+		    "start_copy_iterator(id, ranges, dst_groups, flags, time_begin, time_end)\n"
+		    "    Start copy iterator on the Elliptics node specified by @id. Return elliptics.AsyncResult.\n"
+		    "    -- id - elliptics.Id of the node where iteration should be executed\n"
+		    "    -- ranges - list of elliptics.IteratorRange by which keys on the node should be filtered\n"
+		    "    -- dst_groups - list of remote groups where data will be copied/moved\n"
+		    "    -- flags - bits set of elliptics.iterator_flags\n"
+		    "    -- time_begin - start of time range by which keys on the node should be filtered\n"
+		    "    -- time_end - end of time range by which keys on the node should be filtered\n\n"
+		    "    flags = elliptics.iterator_flags.key_range\n"
+		    "    id = session.routes.get_address_id(Address.from_host_port('host.com:1025'))\n"
+		    "    range = elliptics.IteratorRange()\n"
+		    "    range.key_begin = elliptics.Id([0] * 64, 1)\n"
+		    "    range.key_end = elliptics.Id([255] * 64, 1)\n"
+		    "    dst_groups = [2,3]\n"
+		    "    iterator = session.start_copy_iterator(id,\n"
+		    "                                      [range],\n"
+		    "                                      type,\n"
+		    "                                      flags,\n"
+		    "                                      elliptics.Time(0,0),\n"
+		    "                                      elliptics.Time(0,0),\n"
+		    "                                      dst_groups)\n\n"
+		    "    for result in iterator:\n"
+		    "        if result.status != 0:\n"
+		    "            raise AssertionError('Wrong status: {0}'.format(result.status))\n\n"
+		    "        iterator_id = result.id\n"
+		    "        print ('node: {0}, key: {1}, flags: {2}, ts: {3}/{4}, data: {5}'\n"
+		    "               .format(node,\n"
+		    "                       result.response.key,\n"
+		    "                       result.response.user_flags,\n"
+		    "                       result.response.timestamp.tsec,\n"
+		    "                       result.response.timestamp.tnsec,\n"
+		    "                       result.response_data))\n")
+
 		.def("pause_iterator", &elliptics_session::pause_iterator,
 		     bp::args("id", "iterator_id"),
 		    "pause_iterator(id, iterator_id)\n"
@@ -1561,6 +1619,19 @@ void init_elliptics_session() {
 		    "    iterator = session.cancel_iterator(id, iterator_id)\n"
 		    "    iterator.wait()\n")
 
+// Server send operations
+
+		.def("server_send", &elliptics_session::server_send,
+		     bp::args("keys", "iflags", "groups"),
+		    "server_send(keys, iflags, groups)\n"
+		    "    Similar to iterator, but instead of running over all keys on remote backend,\n"
+		    "    remote server nodes will read all specified keys (which live on local backends)\n"
+		    "    and send them to remote nodes.\n"
+		    "    Returns elliptics.AsyncResult.\n"
+		    "    -- keys - iterable object which provides set of elliptics keys (elliptics.Id)\n"
+		    "    -- iflags - bits set of elliptics.iterator_flags\n"
+		    "    -- groups - iterable object which specifies groups to which data should be send\n")
+
 // Index operations
 
 		.def("set_indexes", &elliptics_session::set_indexes,

example/eblob_backend.c

@@ -965,6 +965,7 @@ static int blob_send(struct eblob_backend_config *cfg, void *state, struct dnet_
 	struct dnet_server_send_ctl *ctl;
 	int *groups;
 	int i, err;
+	int backend_id;
 
 	struct dnet_ext_list elist;
 	static const size_t ehdr_size = sizeof(struct dnet_ext_list_hdr);
@@ -981,9 +982,6 @@ static int blob_send(struct eblob_backend_config *cfg, void *state, struct dnet_
 	ids = (struct dnet_raw_id *)(req + 1);
 	groups = (int *)(ids + req->id_num);
 
-	memset(&re, 0, sizeof(struct dnet_iterator_response));
-	re.total_keys = req->id_num;
-
 	/*
 	 * Set NEED_ACK bit to signal server-send controller that we want
 	 * to send final ACK when controller will be destroyed, which in turn
@@ -994,7 +992,8 @@ static int blob_send(struct eblob_backend_config *cfg, void *state, struct dnet_
 	 */
 	cmd->flags |= DNET_FLAGS_NEED_ACK;
 
-	ctl = dnet_server_send_alloc(state, cmd, req->iflags, groups, req->group_num);
+	backend_id = cfg->data.stat_id;
+	ctl = dnet_server_send_alloc(state, cmd, req->iflags, groups, req->group_num, backend_id);
 	if (!ctl) {
 		err = -ENOMEM;
 		goto err_out_exit;
@@ -1018,6 +1017,14 @@ static int blob_send(struct eblob_backend_config *cfg, void *state, struct dnet_
 
 
 	for (i = 0; i < req->id_num; ++i) {
+		memset(&re, 0, sizeof(struct dnet_iterator_response));
+		// set iterator response id to differentiate various commands
+		// client can use cmd->backend_id from reply though
+		re.id = cmd->backend_id;
+		re.key = ids[i];
+		re.iterated_keys = i;
+		re.total_keys = req->id_num;
+
 		memcpy(key.id, ids[i].id, EBLOB_ID_SIZE);
 
 		err = blob_lookup(b, &key, &wc);
@@ -1027,14 +1034,8 @@ static int blob_send(struct eblob_backend_config *cfg, void *state, struct dnet_
 			goto err_out_send_fail_reply;
 		}
 
-		re.key = ids[i];
 		re.flags = wc.flags; // these flags correspond to DNET_RECORD_FLAGS_*
-		re.status = 0;
-		re.iterated_keys = i;
 		re.size = wc.total_data_size;
-		// set iterator response id to differentiate various commands
-		// client can use cmd->backend_id from reply though
-		re.id = cmd->backend_id;
 
 		data_offset = wc.data_offset;
 		record_offset = 0;

include/elliptics/interface.h

@@ -943,7 +943,7 @@ int dnet_get_vm_stat(dnet_logger *l, struct dnet_vm_stat *st);
 
 struct dnet_server_send_ctl;
 struct dnet_server_send_ctl *dnet_server_send_alloc(void *state, struct dnet_cmd *cmd, uint64_t iflags,
-		int *groups, int group_num);
+						    int *groups, int group_num, int backend_id);
 struct dnet_server_send_ctl *dnet_server_send_get(struct dnet_server_send_ctl *ctl);
 int dnet_server_send_put(struct dnet_server_send_ctl *ctl);
 int dnet_server_send_write(struct dnet_server_send_ctl *send,

include/elliptics/packet.h

@@ -1025,7 +1025,7 @@ enum {
 #define DNET_IFLAGS_MOVE		(1<<4)
 /*
  * Overwrite data. If this flag is NOT set, we only write data if remote timestamp is less
- * that that in data being written. When NOT set, data will still be transferred over the network,
+ * than in data being written. When NOT set, data will still be transferred over the network,
  * even if remote timestamp doesn't allow us to overwrite data.
  */
 #define DNET_IFLAGS_OVERWRITE		(1<<5)

library/dnet.c

@@ -476,9 +476,9 @@ static int dnet_iterator_server_send_complete(struct dnet_addr *addr, struct dne
 					lc = r->header;
 					dnet_setup_id(&lc->id, send->cmd.id.group_id, cmd->id.id);
 					lc->cmd = DNET_CMD_DEL;
-					lc->backend_id = -1;
+					lc->backend_id = send->backend_id;
 					lc->trace_id = cmd->trace_id;
-					lc->flags = DNET_FLAGS_NOLOCK;
+					lc->flags = DNET_FLAGS_NOLOCK | DNET_FLAGS_DIRECT | DNET_FLAGS_DIRECT_BACKEND;
 					if (send->cmd.flags & DNET_FLAGS_TRACE_BIT)
 						lc->flags |= DNET_FLAGS_TRACE_BIT;
 					lc->size = sizeof(struct dnet_io_attr);
@@ -530,7 +530,7 @@ static int dnet_iterator_server_send_complete(struct dnet_addr *addr, struct dne
 }
 
 struct dnet_server_send_ctl *dnet_server_send_alloc(void *state, struct dnet_cmd *cmd, uint64_t iflags,
-		int *groups, int group_num)
+						    int *groups, int group_num, int backend_id)
 {
 	int err;
 	struct dnet_net_state *st = state;
@@ -549,6 +549,7 @@ struct dnet_server_send_ctl *dnet_server_send_alloc(void *state, struct dnet_cmd
 	ctl->state = state;
 	ctl->cmd = *cmd;
 	ctl->iflags = iflags;
+	ctl->backend_id = backend_id;
 	ctl->groups = (int *)(ctl + 1);
 	memcpy(ctl->groups, groups, sizeof(int) * group_num);
 	ctl->group_num = group_num;
@@ -1112,7 +1113,7 @@ static int dnet_iterator_start(struct dnet_backend_io *backend, struct dnet_net_
 		 * dnet_cmd, thus it will store command structure without NEED_ACK bit.
 		 */
 		cmd->flags &= ~DNET_FLAGS_NEED_ACK;
-		sspriv = dnet_server_send_alloc(st, cmd, ireq->flags, dst_groups, ireq->group_num);
+		sspriv = dnet_server_send_alloc(st, cmd, ireq->flags, dst_groups, ireq->group_num, backend->backend_id);
 		cmd->flags |= DNET_FLAGS_NEED_ACK;
 
 		if (!sspriv) {

library/elliptics.h

@@ -1049,6 +1049,7 @@ struct dnet_server_send_ctl {
 
 	uint64_t			iflags;		/* Iterator flags */
 
+	int                             backend_id;     /* Source backend_id */
 	int				*groups;	/* Groups to send WRITE commands */
 	int				group_num;
 

recovery/elliptics_recovery/iterator.py

@@ -239,7 +239,7 @@ def __init__(self, node, group, separately=False, trace_id=0):
         self.session.trace_id = trace_id
         self.separately = separately
 
-    def get_key_range_id(self, key):
+    def _get_key_range_id(self, key):
         if not self.separately:
             return 0
 
@@ -260,7 +260,6 @@ def get_key_range_id(self, key):
 
     def start(self,
               eid=IdRange.ID_MIN,
-              itype=elliptics.iterator_types.network,
               flags=elliptics.iterator_flags.key_range | elliptics.iterator_flags.ts_range,
               key_ranges=(IdRange(IdRange.ID_MIN, IdRange.ID_MAX),),
               timestamp_range=(Time.time_min().to_etime(), Time.time_max().to_etime()),
@@ -270,7 +269,6 @@ def start(self,
               group_id=0,
               leave_file=False,
               batch_size=1024):
-        assert itype == elliptics.iterator_types.network, "Only network iterator is supported for now"
         assert flags & elliptics.iterator_flags.data == 0, "Only metadata iterator is supported for now"
         assert len(key_ranges) > 0, "There should be at least one iteration range."
         self.ranges = key_ranges
@@ -300,12 +298,11 @@ def start(self,
                                                           leave_file=leave_file)
 
             ranges = [IdRange.elliptics_range(start, stop) for start, stop in key_ranges]
-            records = self.session.start_iterator(eid,
-                                                  ranges,
-                                                  itype,
-                                                  flags,
-                                                  timestamp_range[0],
-                                                  timestamp_range[1])
+            records = self._start_iterator(eid,
+                                           ranges,
+                                           flags,
+                                           timestamp_range)
+
             iterated_keys = 0
             total_keys = 0
 
@@ -322,9 +319,8 @@ def start(self,
 
                 if iterated_keys % batch_size == 0:
                     yield (iterated_keys, total_keys, start, end)
-                if record.response.status != 0:
-                    continue
-                results[self.get_key_range_id(record.response.key)].append(record)
+
+                self._on_key_response(results, record)
             end = time.time()
 
             elapsed_time = records.elapsed_time()
@@ -339,23 +335,34 @@ def start(self,
                            .format(address, backend_id, repr(e), traceback.format_exc()))
             yield None
 
-    @classmethod
-    def iterate_with_stats(cls, node, eid, timestamp_range,
+    def _start_iterator(self, eid, ranges, flags, timestamp_range):
+        return self.session.start_iterator(eid,
+                                           ranges,
+                                           elliptics.iterator_types.network,
+                                           flags,
+                                           timestamp_range[0],
+                                           timestamp_range[1])
+
+    def _on_key_response(self, results, record):
+        if record.response.status == 0:
+            self._save_record(results, record)
+
+    def _save_record(self, results, record):
+        results[self._get_key_range_id(record.response.key)].append(record)
+
+    def iterate_with_stats(self, eid, timestamp_range,
                            key_ranges, tmp_dir, address, group_id, backend_id, batch_size,
-                           stats, flags, leave_file=False,
-                           separately=False, trace_id=0):
-        iterator = cls(node, group_id, separately, trace_id=trace_id)
-        result = iterator.start(eid=eid,
-                                timestamp_range=timestamp_range,
-                                flags=flags,
-                                key_ranges=key_ranges,
-                                tmp_dir=tmp_dir,
-                                address=address,
-                                backend_id=backend_id,
-                                group_id=group_id,
-                                batch_size=batch_size,
-                                leave_file=leave_file,
-                                )
+                           stats, flags, leave_file=False):
+        result = self.start(eid=eid,
+                            flags=flags,
+                            key_ranges=key_ranges,
+                            timestamp_range=timestamp_range,
+                            tmp_dir=tmp_dir,
+                            address=address,
+                            backend_id=backend_id,
+                            group_id=group_id,
+                            leave_file=leave_file,
+                            batch_size=batch_size,)
         result_len = 0
         for it in result:
             if it is None:
@@ -379,6 +386,24 @@ def iterate_with_stats(cls, node, eid, timestamp_range,
         return result, result_len
 
 
+class MergeRecoveryIterator(Iterator):
+    '''
+    This class is used in merge recovery for backend iteratation on ranges which are not belong to
+    it using copy iterator. Every iterated key is moved to the backend, where it should exists.
+    If moving of some key was failed, then it saves the key to the results container.
+    '''
+    def __init__(self, *args,  **kwargs):
+        super(MergeRecoveryIterator, self).__init__(*args, **kwargs)
+
+    def _start_iterator(self, eid, ranges, flags, timestamp_range):
+        flags |= elliptics.iterator_flags.move
+        return self.session.start_copy_iterator(eid, ranges, [eid.group_id], flags, timestamp_range[0], timestamp_range[1])
+
+    def _on_key_response(self, results, record):
+        if record.response.status != 0:
+            self._save_record(results, record)
+
+
 class MergeData(object):
     """
     Assist class for IteratorResult.__merge__