[Tarantool-patches] [PATCH v4 05/11] xstream: get rid of an exception

[Georgy Kirichenko <georgy@tarantool.org>]

Refactoring: make xstream C-compliant

Part of #380
---
 src/box/CMakeLists.txt |  1 -
 src/box/box.cc         |  5 +++--
 src/box/relay.cc       | 25 +++++++++++++-----------
 src/box/xstream.cc     | 44 ------------------------------------------
 src/box/xstream.h      |  9 ++++++---
 5 files changed, 23 insertions(+), 61 deletions(-)
 delete mode 100644 src/box/xstream.cc

diff --git a/src/box/CMakeLists.txt b/src/box/CMakeLists.txt
index 56758bd2f..0cc154ba5 100644
--- a/src/box/CMakeLists.txt
+++ b/src/box/CMakeLists.txt
@@ -125,7 +125,6 @@ add_library(box STATIC
     authentication.cc
     replication.cc
     recovery.cc
-    xstream.cc
     applier.cc
     relay.cc
     journal.c
diff --git a/src/box/box.cc b/src/box/box.cc
index ca1696383..c5aaad295 100644
--- a/src/box/box.cc
+++ b/src/box/box.cc
@@ -327,7 +327,7 @@ recovery_journal_create(struct recovery_journal *journal, struct vclock *v)
 	journal->vclock = v;
 }
 
-static void
+static int
 apply_wal_row(struct xstream *stream, struct xrow_header *row)
 {
 	struct request request;
@@ -336,7 +336,7 @@ apply_wal_row(struct xstream *stream, struct xrow_header *row)
 		struct space *space = space_cache_find_xc(request.space_id);
 		if (box_process_rw(&request, space, NULL) != 0) {
 			say_error("error applying row: %s", request_str(&request));
-			diag_raise();
+			return -1;
 		}
 	}
 	struct wal_stream *xstream =
@@ -347,6 +347,7 @@ apply_wal_row(struct xstream *stream, struct xrow_header *row)
 	 */
 	if (++xstream->rows % WAL_ROWS_PER_YIELD == 0)
 		fiber_sleep(0);
+	return 0;
 }
 
 static void
diff --git a/src/box/relay.cc b/src/box/relay.cc
index bb7761b99..5b0d3f023 100644
--- a/src/box/relay.cc
+++ b/src/box/relay.cc
@@ -165,11 +165,11 @@ relay_last_row_time(const struct relay *relay)
 	return relay->last_row_time;
 }
 
-static void
+static int
 relay_send(struct relay *relay, struct xrow_header *packet);
-static void
+static int
 relay_send_initial_join_row(struct xstream *stream, struct xrow_header *row);
-static void
+static int
 relay_send_row(struct xstream *stream, struct xrow_header *row);
 
 struct relay *
@@ -192,7 +192,7 @@ relay_new(struct replica *replica)
 
 static void
 relay_start(struct relay *relay, int fd, uint64_t sync,
-	     void (*stream_write)(struct xstream *, struct xrow_header *))
+	     int (*stream_write)(struct xstream *, struct xrow_header *))
 {
 	xstream_create(&relay->stream, stream_write);
 	/*
@@ -716,7 +716,7 @@ relay_subscribe(struct replica *replica, int fd, uint64_t sync,
 		diag_raise();
 }
 
-static void
+static int
 relay_send(struct relay *relay, struct xrow_header *packet)
 {
 	ERROR_INJECT_YIELD(ERRINJ_RELAY_SEND_DELAY);
@@ -724,15 +724,16 @@ relay_send(struct relay *relay, struct xrow_header *packet)
 	packet->sync = relay->sync;
 	relay->last_row_time = ev_monotonic_now(loop());
 	if (coio_write_xrow(&relay->io, packet) < 0)
-		diag_raise();
+		return -1;
 	fiber_gc();
 
 	struct errinj *inj = errinj(ERRINJ_RELAY_TIMEOUT, ERRINJ_DOUBLE);
 	if (inj != NULL && inj->dparam > 0)
 		fiber_sleep(inj->dparam);
+	return 0;
 }
 
-static void
+static int
 relay_send_initial_join_row(struct xstream *stream, struct xrow_header *row)
 {
 	struct relay *relay = container_of(stream, struct relay, stream);
@@ -741,11 +742,12 @@ relay_send_initial_join_row(struct xstream *stream, struct xrow_header *row)
 	 * vclock while sending a snapshot.
 	 */
 	if (row->group_id != GROUP_LOCAL)
-		relay_send(relay, row);
+		return relay_send(relay, row);
+	return 0;
 }
 
 /** Send a single row to the client. */
-static void
+static int
 relay_send_row(struct xstream *stream, struct xrow_header *packet)
 {
 	struct relay *relay = container_of(stream, struct relay, stream);
@@ -762,7 +764,7 @@ relay_send_row(struct xstream *stream, struct xrow_header *packet)
 		 * skip all these rows.
 		 */
 		if (packet->replica_id == REPLICA_ID_NIL)
-			return;
+			return 0;
 		packet->type = IPROTO_NOP;
 		packet->group_id = GROUP_DEFAULT;
 		packet->bodycnt = 0;
@@ -790,6 +792,7 @@ relay_send_row(struct xstream *stream, struct xrow_header *packet)
 			say_warn("injected broken lsn: %lld",
 				 (long long) packet->lsn);
 		}
-		relay_send(relay, packet);
+		return relay_send(relay, packet);
 	}
+	return 0;
 }
diff --git a/src/box/xstream.cc b/src/box/xstream.cc
deleted file mode 100644
index c77e4360e..000000000
--- a/src/box/xstream.cc
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * Copyright 2010-2016, Tarantool AUTHORS, please see AUTHORS file.
- *
- * Redistribution and use in source and binary forms, with or
- * without modification, are permitted provided that the following
- * conditions are met:
- *
- * 1. Redistributions of source code must retain the above
- *    copyright notice, this list of conditions and the
- *    following disclaimer.
- *
- * 2. 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.
- *
- * THIS SOFTWARE IS PROVIDED BY AUTHORS ``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
- * <COPYRIGHT HOLDER> 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 "xstream.h"
-#include "exception.h"
-
-int
-xstream_write(struct xstream *stream, struct xrow_header *row)
-{
-	try {
-		stream->write(stream, row);
-	} catch (Exception *e) {
-		return -1;
-	}
-	return 0;
-}
diff --git a/src/box/xstream.h b/src/box/xstream.h
index d29ff4213..ae07c3e22 100644
--- a/src/box/xstream.h
+++ b/src/box/xstream.h
@@ -41,7 +41,7 @@ extern "C" {
 struct xrow_header;
 struct xstream;
 
-typedef void (*xstream_write_f)(struct xstream *, struct xrow_header *);
+typedef int (*xstream_write_f)(struct xstream *, struct xrow_header *);
 
 struct xstream {
 	xstream_write_f write;
@@ -53,8 +53,11 @@ xstream_create(struct xstream *xstream, xstream_write_f write)
 	xstream->write = write;
 }
 
-int
-xstream_write(struct xstream *stream, struct xrow_header *row);
+static inline int
+xstream_write(struct xstream *stream, struct xrow_header *row)
+{
+	return stream->write(stream, row);
+}
 
 #if defined(__cplusplus)
 } /* extern C */
-- 
2.25.0