Windows compatibility (#57)

* Add Python and Redis submodules, and remove old third-party modules

* Update VS projects (WARNING: references files that do not exist yet)

* Update code & add shims for APIs except AF_UNIX/{send,recv}msg()

* Minor style changes.

src/plasma/malloc.c

@@ -46,13 +46,33 @@ struct mmap_record *records_by_pointer = NULL;
 
 const int GRANULARITY_MULTIPLIER = 2;
 
+static void *pointer_advance(void *p, ptrdiff_t n) {
+  return (unsigned char *) p + n;
+}
+
+static void *pointer_retreat(void *p, ptrdiff_t n) {
+  return (unsigned char *) p - n;
+}
+
+static ptrdiff_t pointer_distance(void const *pfrom, void const *pto) {
+  return (unsigned char const *) pto - (unsigned char const *) pfrom;
+}
+
 /* Create a buffer. This is creating a temporary file and then
  * immediately unlinking it so we do not leave traces in the system. */
 int create_buffer(int64_t size) {
+  int fd;
+#ifdef _WIN32
+  if (!CreateFileMapping(INVALID_HANDLE_VALUE, NULL, PAGE_READWRITE,
+                         (DWORD)((uint64_t) size >> (CHAR_BIT * sizeof(DWORD))),
+                         (DWORD)(uint64_t) size, NULL)) {
+    fd = -1;
+  }
+#else
   static char template[] = "/tmp/plasmaXXXXXX";
   char file_name[32];
   strncpy(file_name, template, 32);
-  int fd = mkstemp(file_name);
+  fd = mkstemp(file_name);
   if (fd < 0)
     return -1;
   FILE *file = fdopen(fd, "a+");
@@ -68,6 +88,7 @@ int create_buffer(int64_t size) {
     LOG_ERROR("ftruncate error");
     return -1;
   }
+#endif
   return fd;
 }
 
@@ -95,14 +116,14 @@ void *fake_mmap(size_t size) {
   HASH_ADD(hh_pointer, records_by_pointer, pointer, sizeof(pointer), record);
 
   /* We lie to dlmalloc about where mapped memory actually lives. */
-  pointer += sizeof(size_t);
+  pointer = pointer_advance(pointer, sizeof(size_t));
   LOG_DEBUG("%p = fake_mmap(%lu)", pointer, size);
   return pointer;
 }
 
 int fake_munmap(void *addr, size_t size) {
   LOG_DEBUG("fake_munmap(%p, %lu)", addr, size);
-  addr -= sizeof(size_t);
+  addr = pointer_retreat(addr, sizeof(size_t));
   size += sizeof(size_t);
 
   struct mmap_record *record;
@@ -113,12 +134,15 @@ int fake_munmap(void *addr, size_t size) {
      * calls to mmap, to prevent dlmalloc from trimming. */
     return -1;
   }
-  close(record->fd);
 
   HASH_DELETE(hh_fd, records_by_fd, record);
   HASH_DELETE(hh_pointer, records_by_pointer, record);
 
-  return munmap(addr, size);
+  int r = munmap(addr, size);
+  if (r == 0) {
+    close(record->fd);
+  }
+  return r;
 }
 
 void get_malloc_mapinfo(void *addr,
@@ -128,10 +152,11 @@ void get_malloc_mapinfo(void *addr,
   struct mmap_record *record;
   /* TODO(rshin): Implement a more efficient search through records_by_fd. */
   for (record = records_by_fd; record != NULL; record = record->hh_fd.next) {
-    if (addr >= record->pointer && addr < record->pointer + record->size) {
+    if (addr >= record->pointer &&
+        addr < pointer_advance(record->pointer, record->size)) {
       *fd = record->fd;
       *map_size = record->size;
-      *offset = addr - record->pointer;
+      *offset = pointer_distance(record->pointer, addr);
       return;
     }
   }

src/plasma/plasma.h

@@ -6,6 +6,7 @@
 #include <errno.h>
 #include <stddef.h>
 #include <string.h>
+#include <unistd.h> /* pid_t */
 
 #include "common.h"
 

src/plasma/plasma_client.c

@@ -1,5 +1,9 @@
 /* PLASMA CLIENT: Client library for using the plasma store and manager */
 
+#ifdef _WIN32
+#include <Win32_Interop/win32_types.h>
+#endif
+
 #include <assert.h>
 #include <fcntl.h>
 #include <stdlib.h>
@@ -103,7 +107,10 @@ void plasma_send_request(int fd, int type, plasma_request *req) {
 }
 
 plasma_request make_plasma_request(object_id object_id) {
-  plasma_request req = {.num_object_ids = 1, .object_ids = {object_id}};
+  plasma_request req;
+  memset(&req, 0, sizeof(req));
+  req.num_object_ids = 1;
+  req.object_ids[0] = object_id;
   return req;
 }
 
@@ -326,7 +333,9 @@ void plasma_delete(plasma_connection *conn, object_id object_id) {
 
 int64_t plasma_evict(plasma_connection *conn, int64_t num_bytes) {
   /* Send a request to the store to evict objects. */
-  plasma_request req = {.num_bytes = num_bytes};
+  plasma_request req;
+  memset(&req, 0, sizeof(req));
+  req.num_bytes = num_bytes;
   plasma_send_request(conn->store_conn, PLASMA_EVICT, &req);
   /* Wait for a response with the number of bytes actually evicted. */
   plasma_reply reply;
@@ -338,6 +347,8 @@ int64_t plasma_evict(plasma_connection *conn, int64_t num_bytes) {
 
 int plasma_subscribe(plasma_connection *conn) {
   int fd[2];
+  /* TODO: Just create 1 socket, bind it to port 0 to find a free port, and
+   * send the port number instead, and let the client connect. */
   /* Create a non-blocking socket pair. This will only be used to send
    * notifications from the Plasma store to the client. */
   socketpair(AF_UNIX, SOCK_STREAM, 0, fd);
@@ -345,7 +356,7 @@ int plasma_subscribe(plasma_connection *conn) {
   int flags = fcntl(fd[1], F_GETFL, 0);
   CHECK(fcntl(fd[1], F_SETFL, flags | O_NONBLOCK) == 0);
   /* Tell the Plasma store about the subscription. */
-  plasma_request req = {};
+  plasma_request req = {0};
   plasma_send_request(conn->store_conn, PLASMA_SUBSCRIBE, &req);
   /* Send the file descriptor that the Plasma store should use to push
    * notifications about sealed objects to this client. We include a one byte

src/plasma/plasma_manager.c

@@ -156,8 +156,11 @@ int send_client_reply(client_connection *conn, plasma_reply *reply) {
 }
 
 int send_client_failure_reply(object_id object_id, client_connection *conn) {
-  plasma_reply reply = {
-      .object_ids = {object_id}, .num_object_ids = 1, .has_object = 0};
+  plasma_reply reply;
+  memset(&reply, 0, sizeof(reply));
+  reply.object_ids[0] = object_id;
+  reply.num_object_ids = 1;
+  reply.has_object = 0;
   return send_client_reply(conn, &reply);
 }
 
@@ -584,9 +587,11 @@ int manager_timeout_handler(event_loop *loop, timer_id id, void *context) {
     object_conn->num_retries--;
     return MANAGER_TIMEOUT;
   }
-  plasma_reply reply = {.object_ids = {object_conn->object_id},
-                        .num_object_ids = 1,
-                        .has_object = 0};
+  plasma_reply reply;
+  memset(&reply, 0, sizeof(reply));
+  reply.object_ids[0] = object_conn->object_id;
+  reply.num_object_ids = 1;
+  reply.has_object = 0;
   send_client_reply(client_conn, &reply);
   remove_object_connection(client_conn, object_conn);
   return EVENT_LOOP_TIMER_DONE;
@@ -647,7 +652,10 @@ void process_fetch_request(client_connection *client_conn,
                            object_id object_id) {
   client_conn->is_wait = false;
   client_conn->wait_reply = NULL;
-  plasma_reply reply = {.object_ids = {object_id}, .num_object_ids = 1};
+  plasma_reply reply;
+  memset(&reply, 0, sizeof(reply));
+  reply.object_ids[0] = object_id;
+  reply.num_object_ids = 1;
   if (client_conn->manager_state->db == NULL) {
     reply.has_object = 0;
     send_client_reply(client_conn, &reply);
@@ -747,7 +755,7 @@ void process_object_notification(event_loop *loop,
   plasma_manager_state *state = context;
   object_id obj_id;
   /* Read the notification from Plasma. */
-  int n = recv(client_sock, &obj_id, sizeof(object_id), MSG_WAITALL);
+  int n = recv(client_sock, (char *) &obj_id, sizeof(object_id), MSG_WAITALL);
   if (n == 0) {
     /* The store has closed the socket. */
     LOG_DEBUG("The plasma store has closed the object notification socket.");
@@ -769,8 +777,11 @@ void process_object_notification(event_loop *loop,
   client_connection *client_conn;
   HASH_FIND(fetch_hh, state->fetch_connections, &obj_id, sizeof(object_id),
             object_conn);
-  plasma_reply reply = {
-      .object_ids = {obj_id}, .num_object_ids = 1, .has_object = 1};
+  plasma_reply reply;
+  memset(&reply, 0, sizeof(reply));
+  reply.object_ids[0] = obj_id;
+  reply.num_object_ids = 1;
+  reply.has_object = 1;
   while (object_conn) {
     next = object_conn->next;
     client_conn = object_conn->client_conn;

src/plasma/plasma_store.c

@@ -380,7 +380,7 @@ void send_notifications(event_loop *loop,
   for (int i = 0; i < utarray_len(queue->object_ids); ++i) {
     object_id *obj_id = (object_id *) utarray_eltptr(queue->object_ids, i);
     /* Attempt to send a notification about this object ID. */
-    int nbytes = send(client_sock, obj_id, sizeof(*obj_id), 0);
+    int nbytes = send(client_sock, (char const *) obj_id, sizeof(*obj_id), 0);
     if (nbytes >= 0) {
       CHECK(nbytes == sizeof(*obj_id));
     } else if (nbytes == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {