diff --git a/src/box/iproto.cc b/src/box/iproto.cc
index f2952812f049f045e332f9b341e1edc8ebe6930c..7ef6c631cb153a2bbda74838874344c006737da1 100644
--- a/src/box/iproto.cc
+++ b/src/box/iproto.cc
@@ -949,7 +949,7 @@ iproto_connection_close(struct iproto_connection *con)
* parsed data is processed. It's important this
* is done only once.
*/
- con->p_ibuf->wpos -= con->parse_size;
+ ibuf_discard(con->p_ibuf, con->parse_size);
mh_int_t node;
mh_foreach(con->streams, node) {
struct iproto_stream *stream = (struct iproto_stream *)
@@ -1056,15 +1056,16 @@ iproto_connection_input_buffer(struct iproto_connection *con)
}
xibuf_reserve(new_ibuf, to_read + con->parse_size);
- /*
- * Discard unparsed data in the old buffer, otherwise it
- * won't be recycled when all parsed requests are processed.
- */
- old_ibuf->wpos -= con->parse_size;
if (con->parse_size != 0) {
/* Move the cached request prefix to the new buffer. */
- memcpy(new_ibuf->rpos, old_ibuf->wpos, con->parse_size);
- new_ibuf->wpos += con->parse_size;
+ void *wpos = ibuf_alloc(new_ibuf, con->parse_size);
+ assert(wpos != NULL);
+ memcpy(wpos, old_ibuf->wpos - con->parse_size, con->parse_size);
+ /*
+ * Discard unparsed data in the old buffer, otherwise it
+ * won't be recycled when all parsed requests are processed.
+ */
+ ibuf_discard(old_ibuf, con->parse_size);
/*
* We made ibuf idle. If obuf was already idle it
* makes the both ibuf and obuf idle, time to trim
diff --git a/src/box/lua/merger.c b/src/box/lua/merger.c
index fb860dce580aa3cea4b6352eeb1e3d469d7e3ab5..7efb3f3b64398378232adeca2399de49e0a86a3e 100644
--- a/src/box/lua/merger.c
+++ b/src/box/lua/merger.c
@@ -115,8 +115,8 @@ decode_header(struct ibuf *buf, size_t *len_p)
static void
encode_header(struct ibuf *output_buffer, uint32_t result_len)
{
- ibuf_reserve(output_buffer, mp_sizeof_array(result_len));
- output_buffer->wpos = mp_encode_array(output_buffer->wpos, result_len);
+ void *wpos = xibuf_alloc(output_buffer, mp_sizeof_array(result_len));
+ mp_encode_array(wpos, result_len);
}
/**
@@ -596,10 +596,10 @@ luaL_merge_source_buffer_next(struct merge_source *base,
return -1;
}
--source->remaining_tuple_count;
- source->buf->rpos = (char *) tuple_end;
if (format == NULL)
format = tuple_format_runtime;
struct tuple *tuple = tuple_new(format, tuple_beg, tuple_end);
+ ibuf_consume_before(source->buf, tuple_end);
if (tuple == NULL)
return -1;
@@ -1140,9 +1140,8 @@ encode_result_buffer(struct lua_State *L, struct merge_source *source,
merge_source_next(source, NULL, &tuple)) == 0 &&
tuple != NULL) {
uint32_t bsize = tuple_bsize(tuple);
- ibuf_reserve(output_buffer, bsize);
- memcpy(output_buffer->wpos, tuple_data(tuple), bsize);
- output_buffer->wpos += bsize;
+ void *wpos = xibuf_alloc(output_buffer, bsize);
+ memcpy(wpos, tuple_data(tuple), bsize);
result_len_offset += bsize;
++result_len;
diff --git a/src/box/lua/net_box.c b/src/box/lua/net_box.c
index 714766d920fd7b025fd111eec70820e4ac65154d..e98efd39972835c1be40b8fef0c3fd62aca11db2 100644
--- a/src/box/lua/net_box.c
+++ b/src/box/lua/net_box.c
@@ -254,6 +254,8 @@ struct netbox_transport {
struct ibuf send_buf;
/** Connection receive buffer. */
struct ibuf recv_buf;
+ /** Size of the last received message. */
+ size_t last_msg_size;
/** Signalled when send_buf becomes empty. */
struct fiber_cond on_send_buf_empty;
/** Next request id. */
@@ -506,6 +508,7 @@ netbox_transport_create(struct netbox_transport *transport)
iostream_clear(&transport->io);
ibuf_create(&transport->send_buf, &cord()->slabc, NETBOX_READAHEAD);
ibuf_create(&transport->recv_buf, &cord()->slabc, NETBOX_READAHEAD);
+ transport->last_msg_size = 0;
fiber_cond_create(&transport->on_send_buf_empty);
transport->next_sync = 1;
transport->requests = mh_i64ptr_new();
@@ -563,6 +566,7 @@ netbox_transport_set_error(struct netbox_transport *transport)
/* Reset buffers. */
ibuf_reinit(&transport->send_buf);
ibuf_reinit(&transport->recv_buf);
+ transport->last_msg_size = 0;
fiber_cond_broadcast(&transport->on_send_buf_empty);
/* Complete requests and clean up the hash. */
struct mh_i64ptr_t *h = transport->requests;
@@ -1120,7 +1124,7 @@ netbox_transport_communicate(struct netbox_transport *transport, size_t limit)
"Peer closed");
return -1;
} if (rc > 0) {
- recv_buf->wpos += rc;
+ VERIFY(ibuf_alloc(recv_buf, rc) != NULL);
} else if (rc == IOSTREAM_ERROR) {
goto io_error;
} else {
@@ -1134,7 +1138,7 @@ netbox_transport_communicate(struct netbox_transport *transport, size_t limit)
ssize_t rc = iostream_write(io, send_buf->rpos,
ibuf_used(send_buf));
if (rc >= 0) {
- send_buf->rpos += rc;
+ ibuf_consume(send_buf, rc);
if (ibuf_used(send_buf) == 0)
fiber_cond_broadcast(on_send_buf_empty);
} else if (rc == IOSTREAM_ERROR) {
@@ -1171,6 +1175,7 @@ static int
netbox_transport_send_and_recv(struct netbox_transport *transport,
struct xrow_header *hdr)
{
+ ibuf_consume(&transport->recv_buf, transport->last_msg_size);
while (true) {
size_t required;
size_t data_len = ibuf_used(&transport->recv_buf);
@@ -1190,10 +1195,11 @@ netbox_transport_send_and_recv(struct netbox_transport *transport,
required = size + len;
if (data_len >= required) {
const char *body_end = rpos + len;
- transport->recv_buf.rpos = (char *)body_end;
- return xrow_header_decode(
- hdr, &rpos, body_end,
- /*end_is_exact=*/true);
+ int rc = xrow_header_decode(
+ hdr, &rpos, body_end,
+ /*end_is_exact=*/true);
+ transport->last_msg_size = body_end - bufpos;
+ return rc;
}
}
if (netbox_transport_communicate(transport, required) != 0)
diff --git a/src/box/lua/tuple.lua b/src/box/lua/tuple.lua
index 33aa444849b45d84fc154b8197d40b56fde9e23f..3c952fdfccbd3e10aef71c4dd2f73d211989b5b3 100644
--- a/src/box/lua/tuple.lua
+++ b/src/box/lua/tuple.lua
@@ -270,9 +270,7 @@ end
local function tuple_to_msgpack(buf, tuple)
assert(ffi.istype(tuple_t, tuple))
local bsize = builtin.box_tuple_bsize(tuple)
- buf:reserve(bsize)
- builtin.box_tuple_to_buf(tuple, buf.wpos, bsize)
- buf.wpos = buf.wpos + bsize
+ builtin.box_tuple_to_buf(tuple, buf:alloc(bsize), bsize)
end
local function tuple_bsize(tuple)
diff --git a/src/httpc.c b/src/httpc.c
index 025346e149bfdf33357c4381d04ab95fdc3c30c4..14880200d1feeb1abd2210802505c7a1801abb97 100644
--- a/src/httpc.c
+++ b/src/httpc.c
@@ -68,7 +68,7 @@ curl_easy_io_read_cb(char *buffer, size_t size, size_t nitems, void *ctx)
size_t read_len = ibuf_len < buffer_size ? ibuf_len : buffer_size;
memcpy(buffer, req->send.rpos, read_len);
- req->send.rpos += read_len;
+ ibuf_consume(&req->send, read_len);
fiber_cond_broadcast(&req->io_send_cond);
return read_len;
@@ -541,7 +541,7 @@ httpc_request_io_read(struct httpc_request *req, char *buf, size_t len,
if (copied == ibuf_len)
ibuf_reset(&req->io_recv);
else
- req->io_recv.rpos += copied;
+ ibuf_consume(&req->io_recv, copied);
}
if (copied < len && recv_len > 0) {
diff --git a/src/lib/core/coio_buf.h b/src/lib/core/coio_buf.h
index 0d90a483c06d7d73aa6ee93a123c087651fcf702..110cc700c5b598bf722f3354f7113266bf2602d7 100644
--- a/src/lib/core/coio_buf.h
+++ b/src/lib/core/coio_buf.h
@@ -50,7 +50,7 @@ coio_bread(struct iostream *io, struct ibuf *buf, size_t sz)
ssize_t n = coio_read_ahead(io, buf->wpos, sz, ibuf_unused(buf));
if (n < 0)
diag_raise();
- buf->wpos += n;
+ VERIFY(ibuf_alloc(buf, n) != NULL);
return n;
}
@@ -68,7 +68,7 @@ coio_bread_timeout(struct iostream *io, struct ibuf *buf, size_t sz,
timeout);
if (n < 0)
diag_raise();
- buf->wpos += n;
+ VERIFY(ibuf_alloc(buf, n) != NULL);
return n;
}
@@ -80,7 +80,7 @@ coio_breadn(struct iostream *io, struct ibuf *buf, size_t sz)
ssize_t n = coio_readn_ahead(io, buf->wpos, sz, ibuf_unused(buf));
if (n < 0)
diag_raise();
- buf->wpos += n;
+ VERIFY(ibuf_alloc(buf, n) != NULL);
return n;
}
@@ -99,7 +99,7 @@ coio_breadn_timeout(struct iostream *io, struct ibuf *buf, size_t sz,
timeout);
if (n < 0)
diag_raise();
- buf->wpos += n;
+ VERIFY(ibuf_alloc(buf, n) != NULL);
return n;
}
diff --git a/src/lua/buffer.lua b/src/lua/buffer.lua
index 9fc5ee8fda2a5a09d9bb9dad0b95bcbdbf5e5b9b..6fa8457829c1d34b75ec9834c0d6177498b346c3 100644
--- a/src/lua/buffer.lua
+++ b/src/lua/buffer.lua
@@ -193,6 +193,13 @@ local function ibuf_read(buf, size)
return rpos
end
+local function ibuf_consume(buf, size)
+ checkibuf(buf, 'consume')
+ checksize(buf, size)
+ utils.poison_memory_region(buf.rpos, size);
+ buf.rpos = buf.rpos + size
+end
+
local function ibuf_serialize(buf)
local properties = { rpos = buf.rpos, wpos = buf.wpos }
return { ibuf = properties }
@@ -207,6 +214,7 @@ local ibuf_methods = {
checksize = ibuf_checksize;
read = ibuf_read;
+ consume = ibuf_consume;
__serialize = ibuf_serialize;
size = ibuf_used;
diff --git a/src/lua/httpc.lua b/src/lua/httpc.lua
index e694f1fc1db4e7d48e2a740fce11ce91258b24b3..a8f058bf6acb41ec0df033d9c96cec42845b90a7 100644
--- a/src/lua/httpc.lua
+++ b/src/lua/httpc.lua
@@ -424,7 +424,7 @@ local function io_read(self, opts, timeout)
local len = check(self, chunk, delimiter)
if len ~= nil then
local data = ffi.string(rbuf.rpos, len)
- rbuf.rpos = rbuf.rpos + len
+ rbuf:consume(len)
return data
end
@@ -438,15 +438,15 @@ local function io_read(self, opts, timeout)
self._errno = nil
local len = rbuf:size()
local data = ffi.string(rbuf.rpos, len)
- rbuf.rpos = rbuf.rpos + len
+ rbuf:consume(len)
return data
else
- rbuf.wpos = rbuf.wpos + res
+ rbuf:alloc(res)
local len = check(self, chunk, delimiter)
if len ~= nil then
self._errno = nil
local data = ffi.string(rbuf.rpos, len)
- rbuf.rpos = rbuf.rpos + len
+ rbuf:consume(len)
return data
end
end
diff --git a/src/lua/socket.lua b/src/lua/socket.lua
index 56671728a4732492a4ce36dccdfff6b25256130b..52694de379a761801f376fa4468a0109615a5b20 100644
--- a/src/lua/socket.lua
+++ b/src/lua/socket.lua
@@ -694,7 +694,7 @@ local function read(self, limit, timeout, check, ...)
if len ~= nil then
self._errno = nil
local data = ffi.string(rbuf.rpos, len)
- rbuf.rpos = rbuf.rpos + len
+ rbuf:consume(len)
return data
end
@@ -709,15 +709,15 @@ local function read(self, limit, timeout, check, ...)
self._errno = nil
local len = rbuf:size()
local data = ffi.string(rbuf.rpos, len)
- rbuf.rpos = rbuf.rpos + len
+ rbuf:consume(len)
return data
elseif res ~= nil then
- rbuf.wpos = rbuf.wpos + res
+ rbuf:alloc(res)
local len = check(self, limit, ...)
if len ~= nil then
self._errno = nil
local data = ffi.string(rbuf.rpos, len)
- rbuf.rpos = rbuf.rpos + len
+ rbuf:consume(len)
return data
end
elseif not errno_is_transient[self._errno] then