Commit 125cf58c authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Simultaneous writes and reads for requests in the queue

Every request is now kept in a maximum of two queues. When a request is
initially inserted, it is inserted in the `to_write` queue. When at
least one write buffer is extracted from that request, it is *moved* to
the `wait_write` queue. When at least one byte of the request has been
written to the socket, it is *also* placed in the `wait_read` queue.

This means that it is possible for a request to be emitting write
buffers, while simultaneously receiving read callbacks.
parent 6138ef4a
...@@ -7,12 +7,15 @@ ...@@ -7,12 +7,15 @@
int redis_request_init(redis_request *self) { int redis_request_init(redis_request *self) {
memset(self, 0, sizeof(*self)); memset(self, 0, sizeof(*self));
ngx_queue_init(&self->wq);
ngx_queue_init(&self->rq);
return REDIS_OK; return REDIS_OK;
} }
int redis_request_destroy(redis_request *self) { void redis_request_destroy(redis_request *self) {
((void) self); ngx_queue_remove(&self->rq);
return REDIS_OK; ngx_queue_remove(&self->wq);
self->free(self);
} }
int redis_request_queue_init(redis_request_queue *self) { int redis_request_queue_init(redis_request_queue *self) {
...@@ -29,54 +32,44 @@ void redis__free_queue(ngx_queue_t *h) { ...@@ -29,54 +32,44 @@ void redis__free_queue(ngx_queue_t *h) {
redis_request *req; redis_request *req;
ngx_queue_foreach(q, h) { ngx_queue_foreach(q, h) {
ngx_queue_remove(q); req = ngx_queue_data(q, redis_request, rq);
req = ngx_queue_data(q, redis_request, queue); redis_request_destroy(req);
req->free(req);
} }
} }
int redis_request_queue_destroy(redis_request_queue *self) { int redis_request_queue_destroy(redis_request_queue *self) {
redis__free_queue(&self->request_to_write); ngx_queue_t *q;
redis__free_queue(&self->request_wait_write); redis_request *req;
redis__free_queue(&self->request_wait_read);
redis_parser_destroy(&self->parser);
return REDIS_OK;
}
void redis_request_queue_insert(redis_request_queue *self, redis_request *request) {
request->request_queue = self;
ngx_queue_insert_head(&self->request_to_write, &request->queue);
if (self->request_to_write_cb) { while (!ngx_queue_empty(&self->request_wait_read)) {
self->request_to_write_cb(self, request); q = ngx_queue_last(&self->request_wait_read);
req = ngx_queue_data(q, redis_request, rq);
redis_request_destroy(req);
} }
}
redis_request *redis__request_queue_move(ngx_queue_t *a, ngx_queue_t *b) { while (!ngx_queue_empty(&self->request_wait_write)) {
ngx_queue_t *q; q = ngx_queue_last(&self->request_wait_write);
req = ngx_queue_data(q, redis_request, wq);
redis_request_destroy(req);
}
/* Unable to move requests when there are none... */ while (!ngx_queue_empty(&self->request_to_write)) {
if (ngx_queue_empty(a)) { q = ngx_queue_last(&self->request_to_write);
return NULL; req = ngx_queue_data(q, redis_request, wq);
redis_request_destroy(req);
} }
q = ngx_queue_last(a); redis_parser_destroy(&self->parser);
ngx_queue_remove(q); return REDIS_OK;
ngx_queue_insert_head(b, q);
return ngx_queue_data(q, redis_request, queue);
} }
redis_request *redis__request_queue_pop_to_write(redis_request_queue *self) { void redis_request_queue_insert(redis_request_queue *self, redis_request *request) {
redis_request *req; request->request_queue = self;
ngx_queue_insert_head(&self->request_to_write, &request->wq);
req = redis__request_queue_move(&self->request_to_write,
&self->request_wait_write);
if (req && self->request_wait_write_cb) { if (self->request_to_write_cb) {
self->request_wait_write_cb(self, req); self->request_to_write_cb(self, request);
} }
return req;
} }
int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, size_t *len) { int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, size_t *len) {
...@@ -86,17 +79,25 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s ...@@ -86,17 +79,25 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s
if (!ngx_queue_empty(&self->request_wait_write)) { if (!ngx_queue_empty(&self->request_wait_write)) {
q = ngx_queue_head(&self->request_wait_write); q = ngx_queue_head(&self->request_wait_write);
req = ngx_queue_data(q, redis_request, queue); req = ngx_queue_data(q, redis_request, wq);
} }
/* We need one non-done request in the wait_write queue */ /* We need one non-done request in the wait_write queue */
while (req == NULL || req->write_ptr_done) { while (req == NULL || req->write_ptr_done) {
if (redis__request_queue_pop_to_write(self) == NULL) { if (ngx_queue_empty(&self->request_to_write)) {
return -1; return -1;
} }
q = ngx_queue_head(&self->request_wait_write); q = ngx_queue_last(&self->request_to_write);
req = ngx_queue_data(q, redis_request, queue); req = ngx_queue_data(q, redis_request, wq);
/* Remove from tail of `to_write`, insert on head of `wait_write` */
ngx_queue_remove(q);
ngx_queue_insert_head(&self->request_wait_write, q);
if (req && self->request_wait_write_cb) {
self->request_wait_write_cb(self, req);
}
} }
assert(req->write_ptr); assert(req->write_ptr);
...@@ -109,82 +110,84 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s ...@@ -109,82 +110,84 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s
return 0; return 0;
} }
redis_request *redis__request_queue_pop_wait_write(redis_request_queue *self) { int redis_request_queue_write_cb(redis_request_queue *self, size_t len) {
redis_request *req; ngx_queue_t *q = NULL;
redis_request *req = NULL;
req = redis__request_queue_move(&self->request_wait_write, int nwritten, done;
&self->request_wait_read);
if (req && self->request_wait_read_cb) { while (len) {
self->request_wait_read_cb(self, req); if (ngx_queue_empty(&self->request_wait_write)) {
} return -1;
}
return req; q = ngx_queue_last(&self->request_wait_write);
} req = ngx_queue_data(q, redis_request, wq);
done = 0;
int redis_request_queue_write_cb(redis_request_queue *self, int n) { /* Add this request to `wait_read` if necessary */
ngx_queue_t *q; if (ngx_queue_empty(&req->rq)) {
redis_request *req; ngx_queue_insert_head(&self->request_wait_read, &req->rq);
int wrote;
/* We need at least one element in the wait_read queue */ if (req && self->request_wait_read_cb) {
if (ngx_queue_empty(&self->request_wait_read)) { self->request_wait_read_cb(self, req);
if (redis__request_queue_pop_wait_write(self) == NULL) { }
/* The request cannot be NULL: it emitted bytes to write and should
* be present in the wait_write queue, waiting for a callback. */
return -1;
} }
}
while (n) {
assert(!ngx_queue_empty(&self->request_wait_read));
q = ngx_queue_head(&self->request_wait_read);
req = ngx_queue_data(q, redis_request, queue);
assert(req->write_cb); assert(req->write_cb);
wrote = req->write_cb(req, n); nwritten = req->write_cb(req, len, &done);
if (wrote == 0) {
if (redis__request_queue_pop_wait_write(self) == NULL) {
return -1;
}
continue; /* Abort on error */
if (nwritten < 0) {
return nwritten;
} }
assert(wrote <= n); assert((unsigned)nwritten <= len);
n -= wrote; len -= nwritten;
/* Remove this request from `wait_write` when done writing */
if (done) {
ngx_queue_remove(q);
ngx_queue_init(q);
}
} }
assert(n == 0);
return 0; return 0;
} }
int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size_t len) { int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size_t len) {
ngx_queue_t *q; ngx_queue_t *q = NULL;
redis_request *req; redis_request *req = NULL;
int n, done; int nread, done;
while (len) { while (len) {
assert(!ngx_queue_empty(&self->request_wait_read)); if (ngx_queue_empty(&self->request_wait_read)) {
return -1;
}
q = ngx_queue_last(&self->request_wait_read); q = ngx_queue_last(&self->request_wait_read);
req = ngx_queue_data(q, redis_request, queue); req = ngx_queue_data(q, redis_request, rq);
done = 0; done = 0;
/* Fire read callback */ assert(req->read_cb);
n = req->read_cb(req, buf, len, &done); nread = req->read_cb(req, buf, len, &done);
if (n < 0) {
return n; /* Abort on error */
if (nread < 0) {
return nread;
} }
assert((unsigned)nread <= len);
buf += nread;
len -= nread;
/* Remove this request from `wait_read` when done reading */
if (done) { if (done) {
ngx_queue_remove(q); ngx_queue_remove(q);
ngx_queue_init(q);
req->read_cb_done = 1; req->read_cb_done = 1;
req->free(req); redis_request_destroy(req);
} }
buf += n;
len -= n;
} }
return REDIS_OK; return 0;
} }
...@@ -29,12 +29,14 @@ typedef void (redis_request_write_ptr)(redis_request *self, ...@@ -29,12 +29,14 @@ typedef void (redis_request_write_ptr)(redis_request *self,
* Arguments: * Arguments:
* self the request as previously inserted in the queue * self the request as previously inserted in the queue
* n the number of bytes written * n the number of bytes written
* done set to non-zero by the request when it has been written in full
* *
* Return: * Return:
* the number of bytes (<= n) actually used * the number of bytes (<= n) actually used
*/ */
typedef int (redis_request_write_cb)(redis_request *self, typedef int (redis_request_write_cb)(redis_request *self,
int n); int n,
int *done);
/* /*
* Let the request know the wire-level data that was fed to the parser on its * Let the request know the wire-level data that was fed to the parser on its
...@@ -45,7 +47,7 @@ typedef int (redis_request_write_cb)(redis_request *self, ...@@ -45,7 +47,7 @@ typedef int (redis_request_write_cb)(redis_request *self,
* self the request as previously inserted in the queue * self the request as previously inserted in the queue
* buf buffer with reply data * buf buffer with reply data
* len length of buffer with reply data * len length of buffer with reply data
* done set to non-zero by the request when it doesn't need more bytes * done set to non-zero by the request when it has been read in full
* *
* Return: * Return:
* the number of bytes (<= n) actually used * the number of bytes (<= n) actually used
...@@ -66,8 +68,8 @@ typedef int (redis_request_read_cb)(redis_request *self, ...@@ -66,8 +68,8 @@ typedef int (redis_request_read_cb)(redis_request *self,
typedef void (redis_request_free)(redis_request *self); typedef void (redis_request_free)(redis_request *self);
#define REDIS_REQUEST_COMMON \ #define REDIS_REQUEST_COMMON \
ngx_queue_t queue; \
redis_request_queue *request_queue; \ redis_request_queue *request_queue; \
ngx_queue_t wq, rq; \
unsigned write_ptr_done:1; \ unsigned write_ptr_done:1; \
unsigned write_cb_done:1; \ unsigned write_cb_done:1; \
unsigned read_cb_done:1; \ unsigned read_cb_done:1; \
...@@ -86,9 +88,9 @@ struct redis_request_s { ...@@ -86,9 +88,9 @@ struct redis_request_s {
* function, it ends up in the `to_write` queue and the `to_write_cb` callback * function, it ends up in the `to_write` queue and the `to_write_cb` callback
* is called. Before the request emits one or more pointers to its buffers, it * is called. Before the request emits one or more pointers to its buffers, it
* is placed in the `wait_write` queue and the `wait_write_cb` callback is * is placed in the `wait_write` queue and the `wait_write_cb` callback is
* called. Finally, when the request is fully put on the wire and its * called. Finally, when one or more bytes of the request have been put on the
* `write_cb` function has indicated that it is done, the request is placed in * wire, the request is placed in the `wait_read` queue and the `wait_read_cb`
* the `wait_read` queue and the `wait_read_cb` callback is called. * callback is called.
* *
* Arguments: * Arguments:
* self request queue * self request queue
...@@ -115,14 +117,14 @@ struct redis_request_queue_s { ...@@ -115,14 +117,14 @@ struct redis_request_queue_s {
}; };
int redis_request_init(redis_request *self); int redis_request_init(redis_request *self);
int redis_request_destroy(redis_request *self); void redis_request_destroy(redis_request *self);
int redis_request_queue_init(redis_request_queue *self); int redis_request_queue_init(redis_request_queue *self);
int redis_request_queue_destroy(redis_request_queue *self); int redis_request_queue_destroy(redis_request_queue *self);
void redis_request_queue_insert(redis_request_queue *self, redis_request *request); void redis_request_queue_insert(redis_request_queue *self, redis_request *request);
int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, size_t *len); int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, size_t *len);
int redis_request_queue_write_cb(redis_request_queue *self, int n); int redis_request_queue_write_cb(redis_request_queue *self, size_t len);
int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size_t len); int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size_t len);
#endif #endif
...@@ -158,7 +158,7 @@ void t1_write_ptr(redis_request *_self, const char **buf, size_t *len, int *done ...@@ -158,7 +158,7 @@ void t1_write_ptr(redis_request *_self, const char **buf, size_t *len, int *done
} }
} }
int t1_write_cb(redis_request *_self, int n) { int t1_write_cb(redis_request *_self, int n, int *done) {
t1_redis_request *self = (t1_redis_request*)_self; t1_redis_request *self = (t1_redis_request*)_self;
int to_write; int to_write;
...@@ -168,6 +168,11 @@ int t1_write_cb(redis_request *_self, int n) { ...@@ -168,6 +168,11 @@ int t1_write_cb(redis_request *_self, int n) {
} }
self->nwritten += to_write; self->nwritten += to_write;
if (self->nwritten == self->len) {
*done = 1;
}
return to_write; return to_write;
} }
...@@ -393,10 +398,55 @@ TEST(read_cb_with_parse_error) { ...@@ -393,10 +398,55 @@ TEST(read_cb_with_parse_error) {
TEARDOWN(); TEARDOWN();
} }
TEST(read_cb_with_pending_write_ptr_emit) {
SETUP_INSERTED();
const char *buf;
size_t len;
req1.emit = 3;
/* Grab first 3 bytes from request */
rv = redis_request_queue_write_ptr(&q, &buf, &len);
assert_equal_size_t(rv, 0);
assert_equal_size_t(len, 3);
assert(strncmp(buf, "hel", len) == 0);
/* Assume 3 bytes were written ("hel" in req1) */
rv = redis_request_queue_write_cb(&q, 3);
assert_equal_size_t(rv, 0);
assert_equal_size_t(req1.nwritten, 3);
/* Feed part of the reply for request 1 */
rv = redis_request_queue_read_cb(&q, "+stat", 5);
assert_equal_size_t(rv, 0);
/* Grab remaining bytes from request */
rv = redis_request_queue_write_ptr(&q, &buf, &len);
assert_equal_size_t(rv, 0);
assert_equal_size_t(len, 2);
assert(strncmp(buf, "lo", len) == 0);
/* Assume remaining bytes were written */
rv = redis_request_queue_write_cb(&q, 2);
assert_equal_size_t(rv, 0);
assert_equal_size_t(req1.nwritten, 5);
/* Feed remaining part of the reply for request 1 */
rv = redis_request_queue_read_cb(&q, "us\r\n", 4);
assert_equal_size_t(rv, 0);
assert(req1.reply != NULL && req1.reply->type == REDIS_STATUS);
assert(req1.free_calls == 1);
TEARDOWN();
}
int main(void) { int main(void) {
test_insert_request(); test_insert_request();
test_write_ptr(); test_write_ptr();
test_write_cb(); test_write_cb();
test_read_cb(); test_read_cb();
test_read_cb_with_parse_error(); test_read_cb_with_parse_error();
test_read_cb_with_pending_write_ptr_emit();
} }
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment