Commit 6138ef4a authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Use done ptr to signal done in write_ptr function

parent d0eb3f65
...@@ -80,37 +80,30 @@ redis_request *redis__request_queue_pop_to_write(redis_request_queue *self) { ...@@ -80,37 +80,30 @@ redis_request *redis__request_queue_pop_to_write(redis_request_queue *self) {
} }
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) {
ngx_queue_t *q; ngx_queue_t *q = NULL;
redis_request *req; redis_request *req = NULL;
int done = 0;
/* We need at least one element in the wait_write queue */
if (ngx_queue_empty(&self->request_wait_write)) {
if (redis__request_queue_pop_to_write(self) == NULL) {
return -1;
}
}
while (1) { if (!ngx_queue_empty(&self->request_wait_write)) {
assert(!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, queue);
}
const char *auxbuf = NULL; /* We need one non-done request in the wait_write queue */
size_t auxlen = 0; while (req == NULL || req->write_ptr_done) {
assert(req->write_ptr);
req->write_ptr(req, &auxbuf, &auxlen);
if (auxbuf == NULL) {
if (redis__request_queue_pop_to_write(self) == NULL) { if (redis__request_queue_pop_to_write(self) == NULL) {
return -1; return -1;
} }
continue; q = ngx_queue_head(&self->request_wait_write);
req = ngx_queue_data(q, redis_request, queue);
} }
*buf = auxbuf; assert(req->write_ptr);
*len = auxlen; req->write_ptr(req, buf, len, &done);
break;
if (done) {
req->write_ptr_done = 1;
} }
return 0; return 0;
......
...@@ -14,12 +14,14 @@ typedef struct redis_request_queue_s redis_request_queue; ...@@ -14,12 +14,14 @@ typedef struct redis_request_queue_s redis_request_queue;
* *
* Arguments: * Arguments:
* self the request as previously inserted in the queue * self the request as previously inserted in the queue
* buf buffer to write, or NULL when there are no more * buf buffer with request data
* len length of the buffer to write * len length of the buffer with request data
* done set to non-zero by the request when it doesn't have more ptrs
*/ */
typedef void (redis_request_write_ptr)(redis_request *self, typedef void (redis_request_write_ptr)(redis_request *self,
const char **buf, const char **buf,
size_t *len); size_t *len,
int *done);
/* /*
* Let the request know that (some part of) it has been written. * Let the request know that (some part of) it has been written.
......
...@@ -138,15 +138,11 @@ struct t1_redis_request_s { ...@@ -138,15 +138,11 @@ struct t1_redis_request_s {
int free_calls; int free_calls;
}; };
void t1_write_ptr(redis_request *_self, const char **buf, size_t *len) { void t1_write_ptr(redis_request *_self, const char **buf, size_t *len, int *done) {
t1_redis_request *self = (t1_redis_request*)_self; t1_redis_request *self = (t1_redis_request*)_self;
size_t to_emit; size_t to_emit;
if (self->nemitted == self->len) { assert(self->nemitted < self->len);
*buf = NULL;
*len = 0;
return;
}
to_emit = self->len - self->nemitted; to_emit = self->len - self->nemitted;
if (to_emit > self->emit) { if (to_emit > self->emit) {
...@@ -156,6 +152,10 @@ void t1_write_ptr(redis_request *_self, const char **buf, size_t *len) { ...@@ -156,6 +152,10 @@ void t1_write_ptr(redis_request *_self, const char **buf, size_t *len) {
*buf = self->buf + self->nemitted; *buf = self->buf + self->nemitted;
*len = to_emit; *len = to_emit;
self->nemitted += to_emit; self->nemitted += to_emit;
if (self->nemitted == self->len) {
*done = 1;
}
} }
int t1_write_cb(redis_request *_self, int n) { int t1_write_cb(redis_request *_self, int n) {
......
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