Commit 796b2f1f authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Add I/O hooks to request queue

parent b2dff318
...@@ -53,6 +53,24 @@ int redis_request_queue_destroy(redis_request_queue *self) { ...@@ -53,6 +53,24 @@ int redis_request_queue_destroy(redis_request_queue *self) {
return REDIS_OK; return REDIS_OK;
} }
void redis_request_queue_write(redis_request_queue *self) {
if (self->write_fn) {
self->write_fn(self);
}
}
void redis_request_queue_start_read(redis_request_queue *self) {
if (self->start_read_fn) {
self->start_read_fn(self);
}
}
void redis_request_queue_stop_read(redis_request_queue *self) {
if (self->stop_read_fn) {
self->stop_read_fn(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) {
request->request_queue = self; request->request_queue = self;
ngx_queue_insert_head(&self->request_to_write, &request->wq); ngx_queue_insert_head(&self->request_to_write, &request->wq);
...@@ -60,6 +78,12 @@ void redis_request_queue_insert(redis_request_queue *self, redis_request *reques ...@@ -60,6 +78,12 @@ void redis_request_queue_insert(redis_request_queue *self, redis_request *reques
if (self->request_to_write_cb) { if (self->request_to_write_cb) {
self->request_to_write_cb(self, request); self->request_to_write_cb(self, request);
} }
/* Kickstart writes when this is the first pending write */
self->pending_writes++;
if (self->pending_writes == 1) {
redis_request_queue_write(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) {
...@@ -95,6 +119,7 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s ...@@ -95,6 +119,7 @@ int redis_request_queue_write_ptr(redis_request_queue *self, const char **buf, s
if (done) { if (done) {
req->write_ptr_done = 1; req->write_ptr_done = 1;
self->pending_writes--;
} }
return 0; return 0;
......
...@@ -103,7 +103,26 @@ typedef void (redis_request_queue_wait_write_cb)(redis_request_queue *self, ...@@ -103,7 +103,26 @@ typedef void (redis_request_queue_wait_write_cb)(redis_request_queue *self,
typedef void (redis_request_queue_wait_read_cb)(redis_request_queue *self, typedef void (redis_request_queue_wait_read_cb)(redis_request_queue *self,
redis_request *request); redis_request *request);
/*
* This type of function is called when I/O needs to happen. All requests that
* a request queue takes in eventually need to be written to a socket. When a
* request is inserted while the I/O library was not yet busy draining the
* queue of requests to write, one of the I/O functions will be called to
* _kickstart_ writes. After the kickstart the I/O library is supposed to
* continue writing until it no longer can retrieve `char *`'s to write. After
* that, it should be kickstarted by user code when user code stopped emitting
* `char*`'s, or is automatically kickstarted when the queue of requests to
* write was drained.
*
* Arguments:
* self request queue
*/
typedef void (redis_request_queue_io_fn)(redis_request_queue *self);
struct redis_request_queue_s { struct redis_request_queue_s {
size_t pending_writes;
ngx_queue_t request_to_write; ngx_queue_t request_to_write;
ngx_queue_t request_wait_write; ngx_queue_t request_wait_write;
ngx_queue_t request_wait_read; ngx_queue_t request_wait_read;
...@@ -113,6 +132,10 @@ struct redis_request_queue_s { ...@@ -113,6 +132,10 @@ struct redis_request_queue_s {
redis_request_queue_wait_write_cb *request_wait_write_cb; redis_request_queue_wait_write_cb *request_wait_write_cb;
redis_request_queue_wait_read_cb *request_wait_read_cb; redis_request_queue_wait_read_cb *request_wait_read_cb;
redis_request_queue_io_fn *write_fn;
redis_request_queue_io_fn *start_read_fn;
redis_request_queue_io_fn *stop_read_fn;
void *data; void *data;
}; };
...@@ -122,6 +145,10 @@ void redis_request_destroy(redis_request *self); ...@@ -122,6 +145,10 @@ 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_start_write(redis_request_queue *self);
void redis_request_queue_start_read(redis_request_queue *self);
void redis_request_queue_stop_read(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, size_t len); int redis_request_queue_write_cb(redis_request_queue *self, size_t len);
......
...@@ -101,6 +101,34 @@ void request_wait_read_cb(redis_request_queue *q, redis_request *r) { ...@@ -101,6 +101,34 @@ void request_wait_read_cb(redis_request_queue *q, redis_request *r) {
s->callc++; s->callc++;
} }
static struct write_fn_s {
int callc;
struct {
redis_request_queue *q;
} *callv;
} write_fn_calls = { .callc = 0, .callv = NULL };
void write_fn_reset(void) {
struct write_fn_s *s = &write_fn_calls;
if (s->callc) {
free(s->callv);
}
s->callc = 0;
s->callv = NULL;
}
void write_fn(redis_request_queue *q) {
struct write_fn_s *s = &write_fn_calls;
s->callv = realloc(s->callv, sizeof(*s->callv) * (s->callc + 1));
assert(s->callv != NULL);
s->callv[s->callc].q = q;
s->callc++;
}
#define SETUP() \ #define SETUP() \
redis_request_queue q; \ redis_request_queue q; \
int rv; \ int rv; \
...@@ -108,13 +136,15 @@ void request_wait_read_cb(redis_request_queue *q, redis_request *r) { ...@@ -108,13 +136,15 @@ void request_wait_read_cb(redis_request_queue *q, redis_request *r) {
request_to_write_cb_reset(); \ request_to_write_cb_reset(); \
request_wait_write_cb_reset(); \ request_wait_write_cb_reset(); \
request_wait_read_cb_reset(); \ request_wait_read_cb_reset(); \
write_fn_reset(); \
\ \
rv = redis_request_queue_init(&q); \ rv = redis_request_queue_init(&q); \
assert_equal_return(rv, REDIS_OK); \ assert_equal_return(rv, REDIS_OK); \
\ \
q.request_to_write_cb = request_to_write_cb; \ q.request_to_write_cb = request_to_write_cb; \
q.request_wait_write_cb = request_wait_write_cb; \ q.request_wait_write_cb = request_wait_write_cb; \
q.request_wait_read_cb = request_wait_read_cb; q.request_wait_read_cb = request_wait_read_cb; \
q.write_fn = write_fn;
#define TEARDOWN() \ #define TEARDOWN() \
redis_request_queue_destroy(&q); redis_request_queue_destroy(&q);
...@@ -235,17 +265,17 @@ TEST(insert_request) { ...@@ -235,17 +265,17 @@ TEST(insert_request) {
TEARDOWN(); TEARDOWN();
} }
#define INIT_REQUEST(_var, _str) \
t1_redis_request (_var); \
t1_init(&(_var)); \
(_var).buf = (_str); \
(_var).len = strlen((_var).buf); \
(_var).emit = (_var).len;
#define SETUP_INSERTED() \ #define SETUP_INSERTED() \
SETUP(); \ SETUP(); \
t1_redis_request req1, req2; \ INIT_REQUEST(req1, "hello"); \
t1_init(&req1); \ INIT_REQUEST(req2, "world"); \
t1_init(&req2); \
req1.buf = "hello"; \
req1.len = strlen(req1.buf); \
req1.emit = req1.len; \
req2.buf = "world"; \
req2.len = strlen(req2.buf); \
req2.emit = req2.len; \
redis_request_queue_insert(&q, (redis_request*)&req1); \ redis_request_queue_insert(&q, (redis_request*)&req1); \
redis_request_queue_insert(&q, (redis_request*)&req2); \ redis_request_queue_insert(&q, (redis_request*)&req2); \
...@@ -457,6 +487,50 @@ TEST(read_cb_with_pending_write_ptr_emit) { ...@@ -457,6 +487,50 @@ TEST(read_cb_with_pending_write_ptr_emit) {
TEARDOWN(); TEARDOWN();
} }
TEST(write_fn_call_on_first_insert) {
SETUP();
INIT_REQUEST(req, "hello");
redis_request_queue_insert(&q, (redis_request*)&req);
/* Test that the write fn was called */
assert_equal_int(write_fn_calls.callc, 1);
assert(write_fn_calls.callv[0].q == &q);
TEARDOWN();
}
TEST(write_fn_no_call_on_second_insert) {
SETUP();
INIT_REQUEST(req1, "hello");
INIT_REQUEST(req2, "world");
redis_request_queue_insert(&q, (redis_request*)&req1);
redis_request_queue_insert(&q, (redis_request*)&req2);
/* Test that the write fn wasn't called twice */
assert_equal_int(write_fn_calls.callc, 1);
assert(write_fn_calls.callv[0].q == &q);
TEARDOWN();
}
TEST(write_fn_call_after_drain) {
SETUP_WRITTEN_CONFIRMED();
INIT_REQUEST(reqN, "hi");
redis_request_queue_insert(&q, (redis_request*)&reqN);
/* Test that the write fn was called */
assert_equal_int(write_fn_calls.callc, 2);
assert(write_fn_calls.callv[0].q == &q);
TEARDOWN();
}
int main(void) { int main(void) {
test_insert_request(); test_insert_request();
test_write_ptr(); test_write_ptr();
...@@ -464,4 +538,7 @@ int main(void) { ...@@ -464,4 +538,7 @@ int main(void) {
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(); test_read_cb_with_pending_write_ptr_emit();
test_write_fn_call_on_first_insert();
test_write_fn_no_call_on_second_insert();
test_write_fn_call_after_drain();
} }
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