diff --git a/src/box/engine_sophia.cc b/src/box/engine_sophia.cc
index a8c158dfbca82810f2bcf8bb7aafead9bc898f06..5489202748522a555627548f13878a070c31da5b 100644
--- a/src/box/engine_sophia.cc
+++ b/src/box/engine_sophia.cc
@@ -252,6 +252,7 @@ SophiaFactory::begin(struct txn *txn, struct space *space)
if (txn->n_stmts == 1) {
assert(tx == NULL);
SophiaIndex *index = (SophiaIndex *)index_find(space, 0);
+ (void) index;
assert(index->db != NULL);
tx = sp_begin(env);
if (tx == NULL)
diff --git a/src/box/error.cc b/src/box/error.cc
index 5b92e59b0424ae214a9be546e05aef45cf2a3f24..8f9bde98775ff00a043a70f2b73e73979686829b 100644
--- a/src/box/error.cc
+++ b/src/box/error.cc
@@ -54,8 +54,7 @@ ClientError::ClientError(const char *file, unsigned line, const char *msg,
void
ClientError::log() const
{
- _say(S_ERROR, m_file, m_line, NULL, "%s: %s", tnt_errcode_str(m_errcode),
- m_errmsg);
+ _say(S_ERROR, m_file, m_line, m_errmsg, "%s", tnt_errcode_str(m_errcode));
}
diff --git a/src/box/recovery.cc b/src/box/recovery.cc
index 619f4b6e212b18e0cc89746b632b731df416019c..2b38313271ccd9c4e8dd96f5146ed41ab43541c6 100644
--- a/src/box/recovery.cc
+++ b/src/box/recovery.cc
@@ -341,10 +341,12 @@ recover_xlog(struct recovery_state *r, struct xlog *l)
try {
recovery_process(r, &row);
} catch (SystemError *e) {
- say_error("can't apply row: %s", e->errmsg());
+ say_error("can't apply row: ");
+ e->log();
goto end;
} catch (Exception *e) {
- say_error("can't apply row: %s", e->errmsg());
+ say_error("can't apply row: ");
+ e->log();
if (l->dir->panic_if_error)
goto end;
}
diff --git a/src/box/replica.cc b/src/box/replica.cc
index 1b2819086d270d334545d457bf32b9d7d85e989d..7c17abf141ef743bad9a4eb8e9b7331e46e3b56a 100644
--- a/src/box/replica.cc
+++ b/src/box/replica.cc
@@ -220,20 +220,23 @@ pull_from_remote(va_list ap)
{
struct recovery_state *r = va_arg(ap, struct recovery_state *);
struct ev_io coio;
- struct iobuf *iobuf = NULL;
+ struct iobuf *iobuf = iobuf_new(fiber_name(fiber()));
ev_loop *loop = loop();
coio_init(&coio);
- for (;;) {
+ auto coio_guard = make_scoped_guard([&] {
+ iobuf_delete(iobuf);
+ evio_close(loop(), &coio);
+ });
+
+ while (true) {
const char *err = NULL;
try {
struct xrow_header row;
fiber_setcancellable(true);
if (! evio_is_active(&coio)) {
remote_set_status(&r->remote, "connecting");
- if (iobuf == NULL)
- iobuf = iobuf_new(fiber_name(fiber()));
err = "can't connect to master";
remote_connect(r, &coio, iobuf);
/* Send SUBSCRIBE request */
@@ -245,24 +248,31 @@ pull_from_remote(va_list ap)
remote_set_status(&r->remote, "connected");
}
err = "can't read row";
+ /**
+ * If there is an error in subscribe, it's
+ * sent directly in response to subscribe.
+ * If subscribe is successful, there is no
+ * "OK" response, but a stream of rows.
+ * from the binary log.
+ */
remote_read_row(&coio, iobuf, &row);
- if (!iproto_type_is_dml(row.type))
- xrow_decode_error(&row); /* error */
fiber_setcancellable(false);
err = NULL;
-
r->remote.recovery_lag = ev_now(loop) - row.tm;
r->remote.recovery_last_update_tstamp =
ev_now(loop);
+ if (iproto_type_is_error(row.type))
+ xrow_decode_error(&row); /* error */
recovery_process(r, &row);
iobuf_reset(iobuf);
fiber_gc();
+ } catch (ClientError *e) {
+ remote_set_status(&r->remote, "stopped");
+ throw;
} catch (FiberCancelException *e) {
remote_set_status(&r->remote, "failed");
- iobuf_delete(iobuf);
- evio_close(loop, &coio);
throw;
} catch (Exception *e) {
remote_set_status(&r->remote, "failed");
@@ -308,11 +318,7 @@ recovery_follow_remote(struct recovery_state *r)
say_crit("starting replication from %s", uri);
snprintf(name, sizeof(name), "replica/%s", uri);
- try {
- f = fiber_new(name, pull_from_remote);
- } catch (Exception *e) {
- return;
- }
+ f = fiber_new(name, pull_from_remote);
r->remote.reader = f;
fiber_call(f, r);
diff --git a/src/box/xrow.cc b/src/box/xrow.cc
index aedc8d19d3d9e91799703beb3041ec0753c4819e..b446a7bc8426be3a66bf060fc4254acd150f224e 100644
--- a/src/box/xrow.cc
+++ b/src/box/xrow.cc
@@ -161,8 +161,7 @@ xrow_encode_uuid(char *pos, const struct tt_uuid *in)
}
int
-xrow_to_iovec(const struct xrow_header *row,
- struct iovec *out)
+xrow_to_iovec(const struct xrow_header *row, struct iovec *out)
{
static const int iov0_len = mp_sizeof_uint(UINT32_MAX);
int iovcnt = xrow_header_encode(row, out + 1) + 1;
diff --git a/src/say.cc b/src/say.cc
index 3d213f6f361c6523516e582f0817ac165b5fcd32..ec96fddcf17b78bb3a1714cd5c116f27d15cb343 100644
--- a/src/say.cc
+++ b/src/say.cc
@@ -281,7 +281,7 @@ vsay(int level, const char *filename, int line, const char *error, const char *f
}
}
- if (level == S_WARN || level == S_ERROR)
+ if (level == S_WARN || level == S_ERROR || level == S_SYSERROR)
p += snprintf(buf + p, len - p, " %s:%i", filename, line);
p += snprintf(buf + p, len - p, " %c> ", level_to_char(level));
diff --git a/src/sio.cc b/src/sio.cc
index dc04c4eaea1b3ca8417caedc96b8363742f34ef1..3fbdde6171b67f5e46ce76414bd092e78b8ae343 100644
--- a/src/sio.cc
+++ b/src/sio.cc
@@ -304,7 +304,7 @@ sio_writev_all(int fd, struct iovec *iov, int iovcnt)
{
ssize_t bytes_total = 0;
struct iovec *iovend = iov + iovcnt;
- while(1) {
+ while (1) {
int cnt = iovend - iov;
if (cnt > IOV_MAX)
cnt = IOV_MAX;
@@ -315,8 +315,16 @@ sio_writev_all(int fd, struct iovec *iov, int iovcnt)
tnt_raise(SocketError, fd, "writev(%d)", cnt);
}
bytes_total += bytes_written;
- while (bytes_written >= iov->iov_len)
+ /*
+ * Check for iov < iovend, since otherwise
+ * if iovend->iov_len is 0, iov may go beyond
+ * iovend
+ */
+ while (bytes_written >= iov->iov_len) {
bytes_written -= (iov++)->iov_len;
+ if (iov == iovend)
+ break;
+ }
if (iov == iovend)
break;
iov->iov_base = (char *) iov->iov_base + bytes_written;