Commit 2a37c4a1 authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Allow sentinel requests in request queue

parent d9915283
......@@ -87,9 +87,11 @@ void redis_request_queue_insert(redis_request_queue *self, redis_request *reques
}
}
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 **dstbuf, size_t *dstlen) {
ngx_queue_t *q = NULL;
redis_request *req = NULL;
const char *buf = NULL;
size_t len = 0;
int done = 0;
if (!ngx_queue_empty(&self->request_wait_write)) {
......@@ -97,32 +99,43 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s
req = ngx_queue_data(q, redis_request, wq);
}
/* We need one non-done request in the wait_write queue */
while (req == NULL || req->write_ptr_done) {
if (ngx_queue_empty(&self->request_to_write)) {
return -1;
}
/* Continue until write_ptr returned a non-empty buffer */
while (len == 0) {
q = ngx_queue_last(&self->request_to_write);
req = ngx_queue_data(q, redis_request, wq);
/* Make sure that `req` is assigned a non-done request */
if (req == NULL || req->write_ptr_done) {
if (ngx_queue_empty(&self->request_to_write)) {
return -1;
}
q = ngx_queue_last(&self->request_to_write);
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);
/* 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 (self->request_wait_write_cb) {
self->request_wait_write_cb(self, req);
if (self->request_wait_write_cb) {
self->request_wait_write_cb(self, req);
}
}
}
assert(req->write_ptr);
req->write_ptr(req, buf, len, &done);
buf = NULL;
len = 0;
done = 0;
assert(req->write_ptr);
req->write_ptr(req, &buf, &len, &done);
if (done) {
req->write_ptr_done = 1;
self->pending_writes--;
if (done) {
req->write_ptr_done = 1;
self->pending_writes--;
}
}
*dstbuf = buf;
*dstlen = len;
return 0;
}
......
......@@ -172,7 +172,7 @@ void t1_write_ptr(redis_request *_self, const char **buf, size_t *len, int *done
t1_redis_request *self = (t1_redis_request*)_self;
size_t to_emit;
assert(self->nemitted < self->len);
assert(self->nemitted <= self->len);
to_emit = self->len - self->nemitted;
if (to_emit > self->emit) {
......@@ -316,6 +316,30 @@ TEST(write_ptr) {
TEARDOWN();
}
TEST(write_ptr_skip_empty) {
SETUP();
INIT_REQUEST(req1, "");
INIT_REQUEST(req2, "hello");
redis_request_queue_insert(&q, (redis_request*)&req1);
redis_request_queue_insert(&q, (redis_request*)&req2);
const char *buf;
size_t len;
rv = redis_request_queue_write_ptr(&q, &buf, &len);
assert_equal_size_t(rv, 0);
assert_equal_size_t(len, 5);
assert(strncmp(buf, "hello", len) == 0);
/* Both req1 and req2 should be marked as done */
assert(req1.write_ptr_done == 1);
assert(req2.write_ptr_done == 1);
TEARDOWN();
}
#define SETUP_WRITTEN_UNCONFIRMED() \
SETUP_INSERTED(); \
do { \
......@@ -522,6 +546,7 @@ TEST(write_fn_call_after_drain) {
int main(void) {
test_insert_request();
test_write_ptr();
test_write_ptr_skip_empty();
test_write_cb();
test_read_cb();
test_read_cb_with_parse_error();
......
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