Implement object table API (#16)

Makefile

@@ -2,10 +2,6 @@ CC = gcc
 CFLAGS = -g -Wall --std=c99 -D_XOPEN_SOURCE=500 -D_POSIX_C_SOURCE=200809L -fPIC -I. -Ithirdparty
 BUILD = build
 
-CFLAGS += -Wmissing-prototypes
-CFLAGS += -Wstrict-prototypes
-CFLAGS += -Wmissing-declarations
-
 all: $(BUILD)/libcommon.a
 
 $(BUILD)/libcommon.a: event_loop.o common.o task.o io.o state/redis.o

common.h

@@ -38,4 +38,6 @@ unique_id globally_unique_id(void);
  * UNIQUE_ID_SIZE + 1 */
 char *sha1_to_hex(const unsigned char *sha1, char *buffer);
 
+typedef unique_id object_id;
+
 #endif

event_loop.c

@@ -12,6 +12,13 @@ void event_loop_init(event_loop *loop) {
   utarray_new(loop->waiting, &poll_icd);
 }
 
+/* Free the space associated to the event loop.
+ * Does not free the event_loop datastructure itself. */
+void event_loop_free(event_loop *loop) {
+  utarray_free(loop->items);
+  utarray_free(loop->waiting);
+}
+
 /* Add a new file descriptor fd to the event loop.
  * This function sets a user defined type and id for the file descriptor
  * which can be queried using event_loop_type and event_loop_id. The parameter
@@ -83,9 +90,9 @@ void *event_loop_get_data(event_loop *loop, int64_t index) {
   return item->data;
 }
 
-/* Free the space associated to the event loop.
- * Does not free the event_loop datastructure itself. */
-void event_loop_free(event_loop *loop) {
-  utarray_free(loop->items);
-  utarray_free(loop->waiting);
+/* Return the type of connection. */
+int event_loop_type(event_loop *loop, int64_t index) {
+  event_loop_item *item =
+      (event_loop_item *) utarray_eltptr(loop->items, index);
+  return item->type;
 }

event_loop.h

@@ -34,5 +34,6 @@ int64_t event_loop_size(event_loop *loop);
 struct pollfd *event_loop_get(event_loop *loop, int64_t index);
 void event_loop_set_data(event_loop *loop, int64_t index, void *data);
 void *event_loop_get_data(event_loop *loop, int64_t index);
+int event_loop_type(event_loop *loop, int64_t index);
 
 #endif

state/object_table.h

@@ -1,15 +1,21 @@
 #include "common.h"
 #include "db.h"
 
-typedef void (*lookup_callback)(void *);
+/* The callback that is called when the result of a lookup
+ * in the object table comes back. The callback should free
+ * the manager_vector array, but NOT the strings they are pointing to. */
+typedef void (*lookup_callback)(object_id object_id,
+                                int manager_count,
+                                const char *manager_vector[]);
 
 /* Register a new object with the directory. */
-void object_table_add(db_conn *db, unique_id object_id);
+/* TODO(pcm): Retry, print for each attempt. */
+void object_table_add(db_conn *db, object_id object_id);
 
-/* Remove object from the directory */
-void object_table_remove(db_conn *db, unique_id object_id);
+/* Remove object from the directory. */
+void object_table_remove(db_conn *db, object_id object_id, const char *manager);
 
 /* Look up entry from the directory */
 void object_table_lookup(db_conn *db,
-                         unique_id object_id,
+                         object_id object_id,
                          lookup_callback callback);

state/redis.c

@@ -5,6 +5,7 @@
 #include "common.h"
 #include "db.h"
 #include "object_table.h"
+#include "task_queue.h"
 #include "event_loop.h"
 #include "redis.h"
 
@@ -88,12 +89,13 @@ void db_connect(const char *address,
     }
     freeReplyObject(reply);
   }
-  redisFree(context);
 
   db->client_type = strdup(client_type);
   db->client_id = num_clients;
   db->reading = 0;
   db->writing = 0;
+  db->service_cache = NULL;
+  db->sync_context = context;
 
   /* Establish async connection */
   db->context = redisAsyncConnect(address, port);
@@ -102,6 +104,18 @@ void db_connect(const char *address,
   db->context->data = (void *) db;
 }
 
+void db_disconnect(db_conn *db) {
+  redisFree(db->sync_context);
+  redisAsyncFree(db->context);
+  service_cache_entry *e, *tmp;
+  HASH_ITER(hh, db->service_cache, e, tmp) {
+    free(e->addr);
+    HASH_DEL(db->service_cache, e);
+    free(e);
+  }
+  free(db->client_type);
+}
+
 void db_event(db_conn *db) {
   if (db->reading) {
     redisAsyncHandleRead(db->context);
@@ -137,51 +151,62 @@ void object_table_add(db_conn *db, unique_id object_id) {
   static char hex_object_id[2 * UNIQUE_ID_SIZE + 1];
   sha1_to_hex(&object_id.id[0], &hex_object_id[0]);
   redisAsyncCommand(db->context, NULL, NULL, "SADD obj:%s %d",
-                    &hex_object_id[0], 0);
+                    &hex_object_id[0], db->client_id);
   if (db->context->err) {
     LOG_REDIS_ERR(db->context, "could not add object_table entry");
   }
 }
 
-void object_table_lookup_callback(redisAsyncContext *c,
-                                  void *r,
-                                  void *privdata) {
+void object_table_get_entry(redisAsyncContext *c, void *r, void *privdata) {
+  db_conn *db = c->data;
+  lookup_callback_data *cb_data = privdata;
   redisReply *reply = r;
   if (reply == NULL)
     return;
-  lookup_callback callback = privdata;
-  char *str = malloc(reply->len);
-  memcpy(str, reply->str, reply->len);
-  callback(str);
-}
-
-void object_table_fetch_addr_port(redisAsyncContext *c,
-                                  void *r,
-                                  void *privdata) {
-  redisReply *reply = r;
-  if (reply == NULL)
-    return;
-  long long manager_id = -1;
-  if (reply->type == REDIS_REPLY_STRING) {
-    manager_id = strtoll(reply->str, NULL, 10);
-  } else if (reply->type != REDIS_REPLY_INTEGER) {
-    manager_id = reply->integer;
+  int *result = malloc(reply->elements * sizeof(int));
+  int64_t manager_count = reply->elements;
+  if (reply->type == REDIS_REPLY_ARRAY) {
+    for (int j = 0; j < reply->elements; j++) {
+      CHECK(reply->element[j]->type == REDIS_REPLY_STRING);
+      result[j] = atoi(reply->element[j]->str);
+      service_cache_entry *entry;
+      HASH_FIND_INT(db->service_cache, &result[j], entry);
+      if (!entry) {
+        redisReply *reply = redisCommand(db->sync_context, "HGET %s %lld",
+                                         db->client_type, result[j]);
+        CHECK(reply->type == REDIS_REPLY_STRING);
+        entry = malloc(sizeof(service_cache_entry));
+        entry->service_id = result[j];
+        entry->addr = strdup(reply->str);
+        HASH_ADD_INT(db->service_cache, service_id, entry);
+        freeReplyObject(reply);
+      }
+    }
   } else {
     LOG_ERR("expected integer or string, received type %d", reply->type);
     exit(-1);
   }
-  db_conn *db = c->data;
-  redisAsyncCommand(db->context, object_table_lookup_callback, privdata,
-                    "HGET %s %lld", db->client_type, manager_id);
+  const char **manager_vector = malloc(manager_count * sizeof(char *));
+  for (int j = 0; j < manager_count; ++j) {
+    service_cache_entry *entry;
+    HASH_FIND_INT(db->service_cache, &result[j], entry);
+    manager_vector[j] = entry->addr;
+  }
+  cb_data->callback(cb_data->object_id, manager_count, manager_vector);
+  free(privdata);
+  free(result);
 }
 
 void object_table_lookup(db_conn *db,
-                         unique_id object_id,
+                         object_id object_id,
                          lookup_callback callback) {
   static char hex_object_id[2 * UNIQUE_ID_SIZE + 1];
   sha1_to_hex(&object_id.id[0], &hex_object_id[0]);
-  redisAsyncCommand(db->context, object_table_fetch_addr_port, callback,
-                    "SRANDMEMBER obj:%s", &hex_object_id[0]);
+  lookup_callback_data *cb_data = malloc(sizeof(lookup_callback_data));
+  cb_data->callback = callback;
+  cb_data->object_id = object_id;
+  redisAsyncCommand(db->context, object_table_get_entry, cb_data,
+                    "SMEMBERS obj:%s", &hex_object_id[0]);
   if (db->context->err) {
     LOG_REDIS_ERR(db->context, "error in object_table lookup");
   }

state/redis.h

@@ -3,6 +3,16 @@
 
 #include "hiredis/hiredis.h"
 #include "hiredis/async.h"
+#include "uthash.h"
+
+typedef struct {
+  /* Unique ID for this service. */
+  int service_id;
+  /* IP address and port of this service. */
+  const char *addr;
+  /* Handle for the uthash table. */
+  UT_hash_handle hh;
+} service_cache_entry;
 
 struct db_conn_impl {
   /* String that identifies this client type. */
@@ -15,11 +25,21 @@ struct db_conn_impl {
   int reading, writing;
   /* The event loop this global state store connection is part of. */
   event_loop *loop;
+  /* Cache for the IP addresses of services. */
+  service_cache_entry *service_cache;
+  /* Redis context for synchronous connections.
+   * Should only be used very rarely, it is not asynchronous. */
+  redisContext *sync_context;
 };
 
-void object_table_fetch_addr_port(redisAsyncContext *c,
-                                  void *r,
-                                  void *privdata);
+typedef struct {
+  /* The callback that will be called. */
+  lookup_callback callback;
+  /* Object ID that is looked up. */
+  object_id object_id;
+} lookup_callback_data;
+
+void object_table_get_entry(redisAsyncContext *c, void *r, void *privdata);
 
 void object_table_lookup_callback(redisAsyncContext *c,
                                   void *r,

state/task_queue.h

@@ -0,0 +1,33 @@
+#ifndef TASK_QUEUE_H
+#define TASK_QUEUE_H
+
+#include "db.h"
+#include "task.h"
+
+/* The task ID is a deterministic hash of the function ID that
+ * the task executes and the argument IDs or argument values */
+typedef unique_id task_id;
+
+/* The task instance ID is a globally unique ID generated which
+ * identifies this particular execution of the task */
+typedef unique_id task_iid;
+
+/* The node id is an identifier for the node the task is
+ * scheduled on */
+typedef unique_id node_id;
+
+/* Callback for subscribing to the task queue. The only argument this
+ * callback gets is the task_id of the. */
+typedef void (*task_queue_callback)(task_iid *task_iid, task_spec *task);
+
+/* Submit task to the global scheduler. */
+void task_queue_submit_task(db_conn *db, task_iid task_iid, task_spec *task);
+
+/* Submit task to a local scheduler based on the decision made by the global
+ * scheduler. */
+void task_queue_schedule_task(db_conn *db, task_iid task_iid, node_id node);
+
+/* Subscribe to task queue. */
+void task_queue_register_callback(db_conn *db, task_queue_callback callback);
+
+#endif

state/task_table.h

@@ -0,0 +1,13 @@
+#ifndef TASK_TABLE_H
+#define TASK_TABLE_H
+
+#include "db.h"
+#include "task.h"
+
+/* Add task to the task table, handle errors here. */
+status task_table_add_task(db_conn *db, task_iid task_iid, task_spec *task);
+
+/* Get specific task from the task table. */
+status task_table_get_task(db_conn *db, task_iid task_iid, task_spec *task);
+
+#endif /* TASK_TABLE_H */

test/db_tests.c

@@ -6,37 +6,61 @@
 #include "state/db.h"
 #include "state/object_table.h"
 #include "state/redis.h"
+#include "task.h"
 
 SUITE(db_tests);
 
 int lookup_successful = 0;
 const char *manager_addr = "127.0.0.1";
-int manager_port = 12345;
-char received_addr[16] = {0};
-char received_port[6] = {0};
+int manager_port1 = 12345;
+int manager_port2 = 12346;
+char received_addr1[16] = {0};
+char received_port1[6] = {0};
+char received_addr2[16] = {0};
+char received_port2[6] = {0};
 
-void test_callback(void *userdata);
-
-void test_callback(void *userdata) {
-  char *reply = userdata;
+/* This is for synchronizing to make sure both entries have been written. */
+void sync_test_callback(object_id object_id,
+                        int manager_count,
+                        const char *manager_vector[]) {
   lookup_successful = 1;
-  if (!reply ||
-      sscanf(reply, "%15[0-9.]:%5[0-9]", received_addr, received_port) != 2) {
-    assert(0);
+  free(manager_vector);
+}
+
+/* This performs the actual test. */
+void test_callback(object_id object_id,
+                   int manager_count,
+                   const char *manager_vector[]) {
+  CHECK(manager_count == 2);
+  lookup_successful = 1;
+  if (!manager_vector[0] ||
+      sscanf(manager_vector[0], "%15[0-9.]:%5[0-9]", received_addr1,
+             received_port1) != 2) {
+    CHECK(0);
   }
-  free(reply);
+  if (!manager_vector[1] ||
+      sscanf(manager_vector[1], "%15[0-9.]:%5[0-9]", received_addr2,
+             received_port2) != 2) {
+    CHECK(0);
+  }
+  free(manager_vector);
 }
 
 TEST object_table_lookup_test(void) {
   event_loop loop;
   event_loop_init(&loop);
-  db_conn conn;
-  db_connect("127.0.0.1", 6379, "plasma_manager", manager_addr, manager_port,
-             &conn);
-  int64_t index = db_attach(&conn, &loop, 0);
-  unique_id id = {{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}};
-  object_table_add(&conn, id);
-  object_table_lookup(&conn, id, test_callback);
+  db_conn conn1;
+  db_connect("127.0.0.1", 6379, "plasma_manager", manager_addr, manager_port1,
+             &conn1);
+  db_conn conn2;
+  db_connect("127.0.0.1", 6379, "plasma_manager", manager_addr, manager_port2,
+             &conn2);
+  int64_t index1 = db_attach(&conn1, &loop, 0);
+  int64_t index2 = db_attach(&conn2, &loop, 1);
+  unique_id id = globally_unique_id();
+  object_table_add(&conn1, id);
+  object_table_add(&conn2, id);
+  object_table_lookup(&conn1, id, sync_test_callback);
   while (!lookup_successful) {
     int num_ready = event_loop_poll(&loop);
     if (num_ready < 0) {
@@ -46,18 +70,50 @@ TEST object_table_lookup_test(void) {
       struct pollfd *waiting = event_loop_get(&loop, i);
       if (waiting->revents == 0)
         continue;
-      if (i == index) {
-        db_event(&conn);
+      if (i == index1) {
+        db_event(&conn1);
+      }
+      if (i == index2) {
+        db_event(&conn2);
       }
     }
   }
-  ASSERT_STR_EQ(&received_addr[0], manager_addr);
-  ASSERT_EQ(atoi(received_port), manager_port);
+  lookup_successful = 0;
+  object_table_lookup(&conn1, id, test_callback);
+  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 == index1) {
+        db_event(&conn1);
+      }
+      if (i == index2) {
+        db_event(&conn2);
+      }
+    }
+  }
+  int port1 = atoi(received_port1);
+  int port2 = atoi(received_port2);
+  ASSERT_STR_EQ(&received_addr1[0], manager_addr);
+  ASSERT((port1 == manager_port1 && port2 == manager_port2) ||
+         (port2 == manager_port1 && port1 == manager_port2));
+
+  db_disconnect(&conn1);
+  db_disconnect(&conn2);
+
+  event_loop_free(&loop);
+
   PASS();
 }
 
 SUITE(db_tests) {
   RUN_TEST(object_table_lookup_test);
+  /* RUN_TEST(task_queue_test); */
 }
 
 GREATEST_MAIN_DEFS();