From 313241e30321a5499fef51df461233c38cbe8261 Mon Sep 17 00:00:00 2001 From: Stephanie Wang Date: Tue, 20 Sep 2016 22:40:35 -0700 Subject: [PATCH] Asynchronous Redis IPC (#14) * Asynchronous Redis IPC * make valgrind happy * cleanup --- Makefile | 7 ++- common.h | 6 ++ io.c | 4 +- state/redis.c | 35 ++++++++++-- state/redis.h | 4 ++ test/db_tests.c | 7 ++- test/redis_tests.c | 138 +++++++++++++++++++++++++++++++++++++++++++++ 7 files changed, 189 insertions(+), 12 deletions(-) create mode 100644 test/redis_tests.c diff --git a/Makefile b/Makefile index 29b7befd9..f1659c882 100644 --- a/Makefile +++ b/Makefile @@ -16,6 +16,9 @@ $(BUILD)/io_tests: test/io_tests.c $(BUILD)/libcommon.a $(BUILD)/task_tests: test/task_tests.c $(BUILD)/libcommon.a $(CC) -o $@ $^ $(CFLAGS) +$(BUILD)/redis_tests: hiredis test/redis_tests.c $(BUILD)/libcommon.a + $(CC) -o $@ test/redis_tests.c $(BUILD)/libcommon.a thirdparty/hiredis/libhiredis.a $(CFLAGS) + clean: rm -f *.o state/*.o test/*.o rm -rf $(BUILD)/* @@ -26,8 +29,8 @@ redis: hiredis: git submodule update --init --recursive -- "thirdparty/hiredis" ; cd thirdparty/hiredis ; make -test: hiredis redis $(BUILD)/db_tests $(BUILD)/io_tests $(BUILD)/task_tests FORCE +test: hiredis redis $(BUILD)/db_tests $(BUILD)/io_tests $(BUILD)/task_tests $(BUILD)/redis_tests FORCE ./thirdparty/redis-3.2.3/src/redis-server & - sleep 1s ; ./build/db_tests ; ./build/io_tests ; ./build/task_tests + sleep 1s ; ./build/db_tests ; ./build/io_tests ; ./build/task_tests ; ./build/redis_tests FORCE: diff --git a/common.h b/common.h index 96e6402f0..3f30b5661 100644 --- a/common.h +++ b/common.h @@ -28,6 +28,12 @@ #define UNIQUE_ID_SIZE 20 +// Cleanup method for running tests with the greatest library. +// Runs the test, then clears the Redis database. +#define RUN_REDIS_TEST(context, test) \ + RUN_TEST(test); \ + freeReplyObject(redisCommand(context, "FLUSHALL")); + typedef struct { unsigned char id[UNIQUE_ID_SIZE]; } unique_id; /* Generate a globally unique ID. */ diff --git a/io.c b/io.c index 1d16a78e0..99295512b 100644 --- a/io.c +++ b/io.c @@ -10,7 +10,7 @@ #include "common.h" -/* Binds to a Unix domain datagram socket at the given +/* Binds to a Unix domain streaming socket at the given * pathname. Removes any existing file at the pathname. Returns * a file descriptor for the socket, or -1 if an error * occurred. */ @@ -44,7 +44,7 @@ int bind_ipc_sock(const char *socket_pathname) { return socket_fd; } -/* Connects to a Unix domain datagram socket at the given +/* Connects to a Unix domain streaming socket at the given * pathname. Returns a file descriptor for the socket, or -1 if * an error occurred. */ int connect_ipc_sock(const char *socket_pathname) { diff --git a/state/redis.c b/state/redis.c index c781b81ef..ae3fb1a6f 100644 --- a/state/redis.c +++ b/state/redis.c @@ -8,12 +8,13 @@ #include "task_queue.h" #include "event_loop.h" #include "redis.h" +#include "io.h" static void poll_add_read(void *privdata) { db_conn *conn = (db_conn *) privdata; if (!conn->reading) { conn->reading = 1; - event_loop_get(conn->loop, 0)->events |= POLLIN; + event_loop_get(conn->loop, conn->db_index)->events |= POLLIN; } } @@ -21,7 +22,7 @@ static void poll_del_read(void *privdata) { db_conn *conn = (db_conn *) privdata; if (conn->reading) { conn->reading = 0; - event_loop_get(conn->loop, 0)->events &= ~POLLIN; + event_loop_get(conn->loop, conn->db_index)->events &= ~POLLIN; } } @@ -29,7 +30,7 @@ static void poll_add_write(void *privdata) { db_conn *conn = (db_conn *) privdata; if (!conn->writing) { conn->writing = 1; - event_loop_get(conn->loop, 0)->events |= POLLOUT; + event_loop_get(conn->loop, conn->db_index)->events |= POLLOUT; } } @@ -37,7 +38,7 @@ static void poll_del_write(void *privdata) { db_conn *conn = (db_conn *) privdata; if (conn->writing) { conn->writing = 0; - event_loop_get(conn->loop, 0)->events &= ~POLLOUT; + event_loop_get(conn->loop, conn->db_index)->events &= ~POLLOUT; } } @@ -143,8 +144,10 @@ int64_t db_attach(db_conn *db, event_loop *loop, int connection_type) { ac->ev.data = db; - return event_loop_attach(loop, connection_type, NULL, c->fd, - POLLIN | POLLOUT); + int64_t index = + event_loop_attach(loop, connection_type, NULL, c->fd, POLLIN | POLLOUT); + db->db_index = index; + return index; } void object_table_add(db_conn *db, unique_id object_id) { @@ -211,3 +214,23 @@ void object_table_lookup(db_conn *db, LOG_REDIS_ERR(db->context, "error in object_table lookup"); } } + +void send_redis_command(int socket_fd, const char *format, ...) { + char *cmd; + va_list ap; + int len; + + va_start(ap, format); + len = redisvFormatCommand(&cmd, format, ap); + va_end(ap); + if (len == -1) { + LOG_ERR("Out of memory while formatting Redis command."); + return; + } else if (len == -2) { + LOG_ERR("Invalid Redis format string."); + return; + } + + write_string(socket_fd, cmd); + free(cmd); +} diff --git a/state/redis.h b/state/redis.h index ad8d5cbcf..132724e67 100644 --- a/state/redis.h +++ b/state/redis.h @@ -25,6 +25,8 @@ struct db_conn_impl { int reading, writing; /* The event loop this global state store connection is part of. */ event_loop *loop; + /* Index of the database connection in the event loop */ + int64_t db_index; /* Cache for the IP addresses of services. */ service_cache_entry *service_cache; /* Redis context for synchronous connections. @@ -44,3 +46,5 @@ void object_table_get_entry(redisAsyncContext *c, void *r, void *privdata); void object_table_lookup_callback(redisAsyncContext *c, void *r, void *privdata); + +void send_redis_command(int socket_fd, const char *format, ...); diff --git a/test/db_tests.c b/test/db_tests.c index 0788345f2..568eeaeb1 100644 --- a/test/db_tests.c +++ b/test/db_tests.c @@ -108,12 +108,15 @@ TEST object_table_lookup_test(void) { event_loop_free(&loop); + lookup_successful = 0; PASS(); } SUITE(db_tests) { - RUN_TEST(object_table_lookup_test); - /* RUN_TEST(task_queue_test); */ + redisContext *context = redisConnect("127.0.0.1", 6379); + redisCommand(context, "FLUSHALL"); + RUN_REDIS_TEST(context, object_table_lookup_test); + redisFree(context); } GREATEST_MAIN_DEFS(); diff --git a/test/redis_tests.c b/test/redis_tests.c new file mode 100644 index 000000000..9efd3bace --- /dev/null +++ b/test/redis_tests.c @@ -0,0 +1,138 @@ +#include "greatest.h" + +#include +#include + +#include "event_loop.h" +#include "state/db.h" +#include "state/redis.h" +#include "io.h" + +SUITE(redis_tests); + +int lookup_successful = 0; +const char *test_set_format = "SET %s %s"; +const char *test_get_format = "GET %s"; +const char *test_key = "foo"; +const char *test_value = "bar"; + +void async_redis_socket_test_callback(redisAsyncContext *ac, + void *r, + void *privdata) { + redisContext *context = redisConnect("127.0.0.1", 6379); + redisReply *reply = redisCommand(context, test_get_format, test_key); + redisFree(context); + assert(reply != NULL); + if (strcmp(reply->str, test_value)) { + freeReplyObject(reply); + assert(0); + } + freeReplyObject(reply); + lookup_successful = 1; +} + +TEST redis_socket_test(void) { + const char *socket_pathname = "redis-test-socket"; + redisContext *context = redisConnect("127.0.0.1", 6379); + ASSERT(context != NULL); + int socket_fd = bind_ipc_sock(socket_pathname); + ASSERT(socket_fd >= 0); + + int client_fd = connect_ipc_sock(socket_pathname); + ASSERT(client_fd >= 0); + + send_redis_command(client_fd, test_set_format, test_key, test_value); + + int server_fd = accept_client(socket_fd); + char *cmd = read_string(server_fd); + close(client_fd); + close(server_fd); + close(socket_fd); + unlink(socket_pathname); + + redisAppendFormattedCommand(context, cmd, strlen(cmd)); + redisReply *tmp; + redisGetReply(context, &tmp); + freeReplyObject(tmp); + redisReply *reply = redisCommand(context, "GET %s", test_key); + ASSERT(reply != NULL); + ASSERT_STR_EQ(reply->str, test_value); + freeReplyObject(reply); + + free(cmd); + redisFree(context); + PASS(); +} + +TEST async_redis_socket_test(void) { + int socket_fd, server_fd, client_fd; + event_loop loop; + event_loop_init(&loop); + /* Start IPC channel. */ + const char *socket_pathname = "async-redis-test-socket"; + socket_fd = bind_ipc_sock(socket_pathname); + ASSERT(socket_fd >= 0); + int64_t ipc_index = event_loop_attach(&loop, 1, NULL, socket_fd, POLLIN); + + /* Start connection to Redis. */ + db_conn conn; + db_connect("127.0.0.1", 6379, "", "", 0, &conn); + int64_t db_index = db_attach(&conn, &loop, 0); + + /* Send a command to the Redis process. */ + client_fd = connect_ipc_sock(socket_pathname); + ASSERT(client_fd >= 0); + send_redis_command(client_fd, test_set_format, test_key, test_value); + + while (!lookup_successful) { + int num_ready = event_loop_poll(&loop); + if (num_ready < 0) { + exit(-1); + } + for (int i = 0; i < event_loop_size(&loop); ++i) { + struct pollfd *waiting = event_loop_get(&loop, i); + if (waiting->revents == 0) + continue; + if (i == db_index) { + db_event(&conn); + } else if (i == ipc_index) { + /* For some reason, this check is necessary for Travis + * to pass these tests. */ + ASSERT(waiting->revents & POLLIN); + server_fd = accept_client(socket_fd); + ASSERT(server_fd >= 0); + event_loop_attach(&loop, 1, NULL, server_fd, POLLIN); + } else { + char *cmd = read_string(waiting->fd); + redisAsyncFormattedCommand(conn.context, + async_redis_socket_test_callback, NULL, cmd, + strlen(cmd)); + free(cmd); + } + } + } + db_disconnect(&conn); + event_loop_free(&loop); + close(server_fd); + close(client_fd); + close(socket_fd); + unlink(socket_pathname); + lookup_successful = 0; + PASS(); +} + +SUITE(redis_tests) { + redisContext *context = redisConnect("127.0.0.1", 6379); + freeReplyObject(redisCommand(context, "FLUSHALL")); + RUN_REDIS_TEST(context, redis_socket_test); + RUN_REDIS_TEST(context, async_redis_socket_test); + redisFree(context); +} + +GREATEST_MAIN_DEFS(); + +int main(int argc, char **argv) { + GREATEST_MAIN_BEGIN(); + RUN_SUITE(redis_tests); + GREATEST_MAIN_END(); +}