#include #include #include #include "handle.h" /* return values */ #include "request.h" int redis_request_init(redis_request *self) { memset(self, 0, sizeof(*self)); ngx_queue_init(&self->wq); ngx_queue_init(&self->rq); return REDIS_OK; } void redis_request_destroy(redis_request *self) { ngx_queue_remove(&self->rq); ngx_queue_remove(&self->wq); self->free(self); } int redis_request_queue_init(redis_request_queue *self) { memset(self, 0, sizeof(*self)); ngx_queue_init(&self->request_to_write); ngx_queue_init(&self->request_wait_write); ngx_queue_init(&self->request_wait_read); redis_parser_init(&self->parser, NULL); return REDIS_OK; } void redis__free_queue(ngx_queue_t *h) { ngx_queue_t *q; redis_request *req; ngx_queue_foreach(q, h) { req = ngx_queue_data(q, redis_request, rq); redis_request_destroy(req); } } int redis_request_queue_destroy(redis_request_queue *self) { ngx_queue_t *q; redis_request *req; while (!ngx_queue_empty(&self->request_wait_read)) { q = ngx_queue_last(&self->request_wait_read); req = ngx_queue_data(q, redis_request, rq); redis_request_destroy(req); } while (!ngx_queue_empty(&self->request_wait_write)) { q = ngx_queue_last(&self->request_wait_write); req = ngx_queue_data(q, redis_request, wq); redis_request_destroy(req); } while (!ngx_queue_empty(&self->request_to_write)) { q = ngx_queue_last(&self->request_to_write); req = ngx_queue_data(q, redis_request, wq); redis_request_destroy(req); } 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->wq); if (self->request_to_write_cb) { self->request_to_write_cb(self, request); } } int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, size_t *len) { ngx_queue_t *q = NULL; redis_request *req = NULL; int done = 0; if (!ngx_queue_empty(&self->request_wait_write)) { q = ngx_queue_head(&self->request_wait_write); 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; } 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); if (req && self->request_wait_write_cb) { self->request_wait_write_cb(self, req); } } assert(req->write_ptr); req->write_ptr(req, buf, len, &done); if (done) { req->write_ptr_done = 1; } return 0; } int redis_request_queue_write_cb(redis_request_queue *self, size_t len) { ngx_queue_t *q = NULL; redis_request *req = NULL; int nwritten, done; while (len) { if (ngx_queue_empty(&self->request_wait_write)) { return -1; } q = ngx_queue_last(&self->request_wait_write); req = ngx_queue_data(q, redis_request, wq); done = 0; /* Add this request to `wait_read` if necessary */ if (ngx_queue_empty(&req->rq)) { ngx_queue_insert_head(&self->request_wait_read, &req->rq); if (req && self->request_wait_read_cb) { self->request_wait_read_cb(self, req); } } assert(req->write_cb); nwritten = req->write_cb(req, len, &done); /* Abort on error */ if (nwritten < 0) { return nwritten; } assert((unsigned)nwritten <= len); len -= nwritten; /* Remove this request from `wait_write` when done writing */ if (done) { ngx_queue_remove(q); ngx_queue_init(q); req->write_cb_done = 1; } } return 0; } int redis_request_queue_read_cb(redis_request_queue *self, const char *buf, size_t len) { ngx_queue_t *q = NULL; redis_request *req = NULL; int nread, done; while (len) { if (ngx_queue_empty(&self->request_wait_read)) { return -1; } q = ngx_queue_last(&self->request_wait_read); req = ngx_queue_data(q, redis_request, rq); done = 0; assert(req->read_cb); nread = req->read_cb(req, buf, len, &done); /* 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) { ngx_queue_remove(q); ngx_queue_init(q); req->read_cb_done = 1; redis_request_destroy(req); } } return 0; }