Commit 10da4544 authored by Pieter Noordhuis's avatar Pieter Noordhuis
Browse files

Buffered read/write for handle

parent c06801cb
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
/* local */ /* local */
#include "handle.h" #include "handle.h"
#include "sds.h"
#define REDIS__READABLE 1 #define REDIS__READABLE 1
#define REDIS__WRITABLE 2 #define REDIS__WRITABLE 2
...@@ -38,6 +39,8 @@ int redis_handle_init(redis_handle *h) { ...@@ -38,6 +39,8 @@ int redis_handle_init(redis_handle *h) {
h->timeout.tv_sec = 5; h->timeout.tv_sec = 5;
h->timeout.tv_usec = 0; h->timeout.tv_usec = 0;
redis_parser_init(&h->parser, NULL); redis_parser_init(&h->parser, NULL);
h->wbuf = NULL;
h->rbuf = NULL;
return REDIS_OK; return REDIS_OK;
} }
...@@ -56,6 +59,16 @@ int redis_handle_close(redis_handle *h) { ...@@ -56,6 +59,16 @@ int redis_handle_close(redis_handle *h) {
h->fd = -1; h->fd = -1;
} }
if (h->wbuf) {
sdsfree(h->wbuf);
h->wbuf = NULL;
}
if (h->rbuf) {
sdsfree(h->rbuf);
h->rbuf = NULL;
}
return REDIS_OK; return REDIS_OK;
} }
...@@ -149,6 +162,8 @@ int redis_handle_connect_in(redis_handle *h, struct sockaddr_in addr) { ...@@ -149,6 +162,8 @@ int redis_handle_connect_in(redis_handle *h, struct sockaddr_in addr) {
} }
h->fd = fd; h->fd = fd;
h->wbuf = sdsempty();
h->rbuf = sdsempty();
return REDIS_OK; return REDIS_OK;
} }
...@@ -167,6 +182,8 @@ int redis_handle_connect_in6(redis_handle *h, struct sockaddr_in6 addr) { ...@@ -167,6 +182,8 @@ int redis_handle_connect_in6(redis_handle *h, struct sockaddr_in6 addr) {
} }
h->fd = fd; h->fd = fd;
h->wbuf = sdsempty();
h->rbuf = sdsempty();
return REDIS_OK; return REDIS_OK;
} }
...@@ -185,6 +202,8 @@ int redis_handle_connect_un(redis_handle *h, struct sockaddr_un addr) { ...@@ -185,6 +202,8 @@ int redis_handle_connect_un(redis_handle *h, struct sockaddr_un addr) {
} }
h->fd = fd; h->fd = fd;
h->wbuf = sdsempty();
h->rbuf = sdsempty();
return REDIS_OK; return REDIS_OK;
} }
...@@ -223,6 +242,8 @@ int redis_handle_connect_gai(redis_handle *h, ...@@ -223,6 +242,8 @@ int redis_handle_connect_gai(redis_handle *h,
freeaddrinfo(servinfo); freeaddrinfo(servinfo);
h->fd = fd; h->fd = fd;
h->wbuf = sdsempty();
h->rbuf = sdsempty();
return REDIS_OK; return REDIS_OK;
} }
...@@ -303,3 +324,87 @@ int redis_handle_wait_readable(redis_handle *h) { ...@@ -303,3 +324,87 @@ int redis_handle_wait_readable(redis_handle *h) {
int redis_handle_wait_writable(redis_handle *h) { int redis_handle_wait_writable(redis_handle *h) {
return redis__wait(h, REDIS__WRITABLE); return redis__wait(h, REDIS__WRITABLE);
} }
int redis_handle_write_from_buffer(redis_handle *h, int *drained) {
int nwritten;
if (h->fd < 0) {
errno = EINVAL;
return REDIS_ESYS;
}
if (sdslen(h->wbuf)) {
nwritten = write(h->fd, h->wbuf, sdslen(h->wbuf));
if (nwritten == -1) {
/* Let all errors bubble, including EAGAIN */
return REDIS_ESYS;
} else if (nwritten > 0) {
h->wbuf = sdsrange(h->wbuf, nwritten, -1);
}
}
if (drained) {
*drained = (sdslen(h->wbuf) == 0);
}
return REDIS_OK;
}
int redis_handle_write_to_buffer(redis_handle *h, const char *buf, size_t len) {
if (h->fd < 0) {
errno = EINVAL;
return REDIS_ESYS;
}
h->wbuf = sdscatlen(h->wbuf, buf, len);
return REDIS_OK;
}
int redis_handle_read_to_buffer(redis_handle *h) {
char buf[2048];
int nread;
if (h->fd < 0) {
errno = EINVAL;
return REDIS_ESYS;
}
nread = read(h->fd, buf, sizeof(buf));
if (nread == -1) {
/* Let all errors bubble, including EAGAIN */
return REDIS_ESYS;
} else if (nread == 0) {
return REDIS_EEOF;
}
h->rbuf = sdscatlen(h->rbuf, buf, nread);
return 0;
}
int redis_handle_read_from_buffer(redis_handle *h, redis_protocol **p) {
size_t navail, nparsed;
if (h->fd < 0) {
errno = EINVAL;
return REDIS_ESYS;
}
assert(p != NULL);
*p = NULL;
navail = sdslen(h->rbuf);
if (navail) {
nparsed = redis_parser_execute(&h->parser, p, h->rbuf, navail);
/* Trim read buffer */
h->rbuf = sdsrange(h->rbuf, nparsed, -1);
/* Test for parse error */
if (nparsed < navail && *p == NULL) {
errno = redis_parser_err(&h->parser);
return REDIS_EPARSER;
}
}
return REDIS_OK;
}
...@@ -15,6 +15,7 @@ ...@@ -15,6 +15,7 @@
#define REDIS_ESYS -1 #define REDIS_ESYS -1
#define REDIS_EGAI -2 #define REDIS_EGAI -2
#define REDIS_EPARSER -3 #define REDIS_EPARSER -3
#define REDIS_EEOF -4
typedef struct redis_handle_s redis_handle; typedef struct redis_handle_s redis_handle;
...@@ -22,6 +23,8 @@ struct redis_handle_s { ...@@ -22,6 +23,8 @@ struct redis_handle_s {
int fd; int fd;
struct timeval timeout; struct timeval timeout;
redis_parser parser; redis_parser parser;
char *wbuf;
char *rbuf;
}; };
int redis_handle_init(redis_handle *h); int redis_handle_init(redis_handle *h);
...@@ -38,4 +41,10 @@ int redis_handle_wait_connected(redis_handle *); ...@@ -38,4 +41,10 @@ int redis_handle_wait_connected(redis_handle *);
int redis_handle_wait_readable(redis_handle *); int redis_handle_wait_readable(redis_handle *);
int redis_handle_wait_writable(redis_handle *); int redis_handle_wait_writable(redis_handle *);
int redis_handle_write_from_buffer(redis_handle *, int *drained);
int redis_handle_write_to_buffer(redis_handle *, const char *buf, size_t len);
int redis_handle_read_to_buffer(redis_handle *);
int redis_handle_read_from_buffer(redis_handle *, redis_protocol **p);
#endif #endif
...@@ -147,6 +147,100 @@ int test_connect_gai_success(void) { ...@@ -147,6 +147,100 @@ int test_connect_gai_success(void) {
return 0; return 0;
} }
redis_handle *setup(void) {
redis_handle *h = malloc(sizeof(*h));
int rv;
rv = redis_handle_init(h);
assert(rv == REDIS_OK);
rv = redis_handle_set_timeout(h, 10000);
assert_equal_int(rv, REDIS_OK);
rv = redis_handle_connect_gai(h, AF_INET, "localhost", redis_port());
assert_equal_int(rv, REDIS_OK);
rv = redis_handle_wait_connected(h);
assert_equal_int(rv, REDIS_OK);
return h;
}
void teardown(redis_handle *h) {
redis_handle_destroy(h);
free(h);
}
void test_eof_after_quit(void) {
redis_handle *h = setup();
int rv, drained;
rv = redis_handle_write_to_buffer(h, "quit\r\n", 6);
assert(rv == REDIS_OK);
rv = redis_handle_write_from_buffer(h, &drained);
assert(rv == REDIS_OK && drained);
rv = redis_handle_wait_readable(h);
assert(rv == REDIS_OK);
rv = redis_handle_read_to_buffer(h);
assert(rv == REDIS_OK);
redis_protocol *p;
rv = redis_handle_read_from_buffer(h, &p);
assert(rv == REDIS_OK);
assert_equal_int(p->type, REDIS_STATUS);
assert_equal_int(p->plen, 5); /* +ok\r\n */
/* wait_readable should return REDIS_OK because EOF is readable */
rv = redis_handle_wait_readable(h);
assert_equal_int(rv, REDIS_OK);
/* 1: read EOF */
rv = redis_handle_read_to_buffer(h);
assert_equal_int(rv, REDIS_EEOF);
/* 2: reading EOF should be idempotent (user is responsible for closing the handle) */
rv = redis_handle_read_to_buffer(h);
assert_equal_int(rv, REDIS_EEOF);
teardown(h);
}
void test_read_timeout(void) {
redis_handle *h = setup();
int rv;
rv = redis_handle_wait_readable(h);
assert_equal_int(rv, REDIS_ESYS);
assert_equal_int(errno, ETIMEDOUT);
teardown(h);
}
void test_einval_against_closed_handle(void) {
redis_handle *h = setup();
int rv;
rv = redis_handle_close(h);
assert_equal_int(rv, REDIS_OK);
rv = redis_handle_write_to_buffer(h, "ping\r\n", 6);
assert_equal_int(rv, REDIS_ESYS);
assert_equal_int(errno, EINVAL);
int drained;
rv = redis_handle_write_from_buffer(h, &drained);
assert_equal_int(rv, REDIS_ESYS);
assert_equal_int(errno, EINVAL);
rv = redis_handle_read_to_buffer(h);
assert_equal_int(rv, REDIS_ESYS);
assert_equal_int(errno, EINVAL);
redis_protocol *p;
rv = redis_handle_read_from_buffer(h, &p);
assert_equal_int(rv, REDIS_ESYS);
assert_equal_int(errno, EINVAL);
teardown(h);
}
int main(void) { int main(void) {
test_connect_in_refused(); test_connect_in_refused();
test_connect_in6_refused(); test_connect_in6_refused();
...@@ -154,4 +248,8 @@ int main(void) { ...@@ -154,4 +248,8 @@ int main(void) {
test_connect_timeout(); test_connect_timeout();
test_connect_gai_unknown_host(); test_connect_gai_unknown_host();
test_connect_gai_success(); test_connect_gai_success();
test_eof_after_quit();
test_read_timeout();
test_einval_against_closed_handle();
} }
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