Basic async IO module using libuv

6 files changed, 331 insertions, 14 deletions

diff --git a/src/rt/rust_log.cpp b/src/rt/rust_log.cpp
index c23a38bc1da..b4d26c780e6 100644
--- a/src/rt/rust_log.cpp
+++ b/src/rt/rust_log.cpp
@@ -197,6 +197,7 @@ size_t log_rt_gc;
 size_t log_rt_stdlib;
 size_t log_rt_kern;
 size_t log_rt_backtrace;
+size_t log_rt_callback;
 
 static const mod_entry _rt_module_map[] =
     {{"::rt::mem", &log_rt_mem},
@@ -211,6 +212,7 @@ static const mod_entry _rt_module_map[] =
      {"::rt::stdlib", &log_rt_stdlib},
      {"::rt::kern", &log_rt_kern},
      {"::rt::backtrace", &log_rt_backtrace},
+     {"::rt::callback", &log_rt_callback},
      {NULL, NULL}};
 
 void update_log_settings(void* crate_map, char* settings) {
diff --git a/src/rt/rust_log.h b/src/rt/rust_log.h
index 6021125705b..8770fc2f758 100644
--- a/src/rt/rust_log.h
+++ b/src/rt/rust_log.h
@@ -66,5 +66,6 @@ extern size_t log_rt_gc;
 extern size_t log_rt_stdlib;
 extern size_t log_rt_kern;
 extern size_t log_rt_backtrace;
+extern size_t log_rt_callback;
 
 #endif /* RUST_LOG_H */
diff --git a/src/rt/rust_upcall.cpp b/src/rt/rust_upcall.cpp
index 2554868c986..02fa7bc2846 100644
--- a/src/rt/rust_upcall.cpp
+++ b/src/rt/rust_upcall.cpp
@@ -1,21 +1,8 @@
 #include "rust_internal.h"
+#include "rust_upcall.h"
 
 // Upcalls.
 
-#ifdef __GNUC__
-#define LOG_UPCALL_ENTRY(task)                            \
-    LOG(task, upcall,                                     \
-        "> UPCALL %s - task: %s 0x%" PRIxPTR              \
-        " retpc: x%" PRIxPTR,                             \
-        __FUNCTION__,                                     \
-        (task)->name, (task),                             \
-        __builtin_return_address(0));
-#else
-#define LOG_UPCALL_ENTRY(task)                            \
-    LOG(task, upcall, "> UPCALL task: %s @x%" PRIxPTR,    \
-        (task)->name, (task));
-#endif
-
 extern "C" CDECL char const *
 str_buf(rust_task *task, rust_str *s);
 
diff --git a/src/rt/rust_upcall.h b/src/rt/rust_upcall.h
new file mode 100644
index 00000000000..e5ee5bb6948
--- /dev/null
+++ b/src/rt/rust_upcall.h
@@ -0,0 +1,17 @@
+#pragma once
+
+#ifdef __GNUC__
+#define LOG_UPCALL_ENTRY(task)                            \
+    LOG(task, upcall,                                     \
+        "> UPCALL %s - task: %s 0x%" PRIxPTR              \
+        " retpc: x%" PRIxPTR,                             \
+        __FUNCTION__,                                     \
+        (task)->name, (task),                             \
+        __builtin_return_address(0));
+#else
+#define LOG_UPCALL_ENTRY(task)                            \
+    LOG(task, upcall, "> UPCALL task: %s @x%" PRIxPTR,    \
+        (task)->name, (task));
+#endif
+
+
diff --git a/src/rt/rust_uv.cpp b/src/rt/rust_uv.cpp
new file mode 100644
index 00000000000..9e4e15b61aa
--- /dev/null
+++ b/src/rt/rust_uv.cpp
@@ -0,0 +1,300 @@
+#include "rust_internal.h"
+#include "rust_upcall.h"
+// Disable libev prototypes - they will make inline compatability functions
+// which are unused and so trigger a warning in gcc since -Wall is on.
+#define EV_PROTOTYPES 0
+#include "libuv/uv.h"
+
+#ifdef __GNUC__
+#define LOG_CALLBACK_ENTRY(p) \
+    LOG(iotask, callback, "> IO CALLBACK %s %p", __FUNCTION__, p)
+#else
+#define LOG_CALLBACK_ENTRY(p) \
+    LOG(iotask, callback, "> IO CALLBACK %s:%d %p", __FILE__, __LINE__, p)
+#endif
+
+// The primary task which is running the event loop. This is used to dispatch
+// all the notifications back to rust so we clone all passed in channels to
+// this task.
+static rust_task *iotask = NULL;
+
+struct socket_data : public task_owned<socket_data> {
+  // Either the task that the connection attempt was made from or the task
+  // that the server was spawned on.
+  rust_task *task;
+  // Channel for reporting the status of a connection attempt
+  // For connections from servers, this is always null
+  // For server sockets, this is used to send the notification that the server
+  // was closed.
+  rust_chan *chan;
+  // Channel to a port which receives bytes from this socket
+  rust_chan *reader;
+  uv_tcp_t socket;
+
+  ~socket_data() {
+    if (chan)
+      chan->deref();
+    if (reader)
+      reader->deref();
+  }
+};
+
+struct request : public uv_req_t, public task_owned<request> {
+  rust_task *task;
+  // Used for notifying about completion of connections, writes
+  rust_chan *chan;
+  request(socket_data *data, rust_chan *chan,
+          void (*cb)(request *req, int status)) {
+    uv_req_init(this, (uv_handle_t*)&data->socket, (void*)cb);
+    this->data = data;
+    this->task = data->task;
+    this->chan = chan->clone(iotask);
+  }
+  socket_data *socket() {
+    return (socket_data*)data;
+  }
+  void send_result(void *data) {
+    chan->send(&data);
+    chan->deref();
+    chan = NULL;
+  }
+};
+
+extern "C" CDECL void aio_close_socket(rust_task *task, socket_data *);
+
+static uv_idle_s idle_handler;
+
+static void idle_callback(uv_handle_t* handle, int status) {
+  rust_task *task = reinterpret_cast<rust_task*>(handle->data);
+  task->yield();
+}
+
+extern "C" CDECL void aio_init(rust_task *task) {
+  LOG_UPCALL_ENTRY(task);
+  iotask = task;
+  uv_init();
+  uv_idle_init(&idle_handler);
+  uv_idle_start(&idle_handler, idle_callback);
+}
+
+extern "C" CDECL void aio_run(rust_task *task) {
+  LOG_UPCALL_ENTRY(task);
+  idle_handler.data = task;
+  uv_run();
+}
+
+void nop_close(uv_handle_t* handle) {}
+
+extern "C" CDECL void aio_stop(rust_task *task) {
+  LOG_UPCALL_ENTRY(task);
+  uv_close((uv_handle_t*)&idle_handler, nop_close);
+}
+
+static socket_data *make_socket(rust_task *task, rust_chan *chan) {
+  socket_data *data = new (task, "make_socket") socket_data;
+  if (!data ||
+      uv_tcp_init(&data->socket)) {
+    return NULL;
+  }
+  data->task = task;
+  // Connections from servers don't have a channel
+  if (chan) {
+    data->chan = chan->clone(iotask);
+  } else {
+    data->chan = NULL;
+  }
+  data->socket.data = data;
+  data->reader = NULL;
+  return data;
+}
+
+// We allocate the requested space + rust_vec but return a pointer at a
+// +rust_vec offset so that it writes the bytes to the correct location.
+static uv_buf_t alloc_buffer(uv_tcp_t *socket, size_t suggested_size) {
+  LOG_CALLBACK_ENTRY(socket);
+  uv_buf_t buf;
+  size_t actual_size = suggested_size + sizeof (rust_ivec_heap);
+  socket_data *data = (socket_data*)socket->data;
+  char *base =
+    reinterpret_cast<char*>(data->task->kernel->malloc(actual_size,
+                                                       "read buffer"));
+  buf.base = base + sizeof (rust_ivec_heap);
+  buf.len = suggested_size;
+  return buf;
+}
+
+static void read_progress(uv_tcp_t *socket, ssize_t nread, uv_buf_t buf) {
+  LOG_CALLBACK_ENTRY(socket);
+  socket_data *data = (socket_data*)socket->data;
+  I(data->task->sched, data->reader != NULL);
+  I(data->task->sched, nread <= ssize_t(buf.len));
+
+  rust_ivec_heap *base = reinterpret_cast<rust_ivec_heap*>(
+      reinterpret_cast<char*>(buf.base) - sizeof (rust_ivec_heap));
+  rust_ivec v;
+  v.fill = 0;
+  v.alloc = buf.len;
+  v.payload.ptr = base;
+
+  switch (nread) {
+    case -1: // End of stream
+      base->fill = 0;
+      uv_read_stop(socket);
+      break;
+    case 0: // Nothing read
+      data->task->kernel->free(base);
+      return;
+    default: // Got nread bytes
+      base->fill = nread;
+      break;
+  }
+  data->reader->send(&v);
+}
+
+static void new_connection(uv_tcp_t *socket, int status) {
+  LOG_CALLBACK_ENTRY(socket);
+  socket_data *server = (socket_data*)socket->data;
+  I(server->task->sched, socket == &server->socket);
+  // Connections from servers don't have a channel
+  socket_data *client = make_socket(server->task, NULL);
+  if (!client) {
+    server->task->fail();
+    return;
+  }
+  if (uv_accept(socket, &client->socket)) {
+    aio_close_socket(client->task, client);
+    server->task->fail();
+    return;
+  }
+  server->chan->send(&client);
+}
+
+extern "C" CDECL socket_data *aio_serve(rust_task *task, const char *ip,
+                                        int port, rust_chan *chan) {
+  LOG_UPCALL_ENTRY(task);
+  struct sockaddr_in addr = uv_ip4_addr(const_cast<char*>(ip), port);
+  socket_data *server = make_socket(iotask, chan);
+  if (!server)
+    goto oom;
+  if (uv_bind(&server->socket, addr) ||
+      uv_listen(&server->socket, 128, new_connection)) {
+    aio_close_socket(task, server);
+    return NULL;
+  }
+  return server;
+oom:
+  task->fail();
+  return NULL;
+}
+
+static void free_socket(uv_handle_t *handle) {
+  LOG_CALLBACK_ENTRY(socket);
+  uv_tcp_t *socket = (uv_tcp_t*)handle;
+  socket_data *data = (socket_data*)socket->data;
+  I(data->task->sched, socket == &data->socket);
+  // For client sockets, send a 0-size buffer to indicate that we're done
+  // reading and should send the close notification.
+  if (data->reader) {
+    if (data->reader->is_associated()) {
+      uv_buf_t buf = alloc_buffer(socket, 0);
+      read_progress(socket, -1, buf);
+      uv_read_stop(socket);
+    }
+  } else {
+    // This is a server socket
+    bool closed = true;
+    I(data->task->sched, data->chan != NULL);
+    data->task->kill();
+    data->chan->send(&closed);
+  }
+  delete data;
+}
+
+extern "C" CDECL void aio_close_socket(rust_task *task, socket_data *client) {
+  LOG_UPCALL_ENTRY(task);
+  if (uv_close((uv_handle_t*)&client->socket, free_socket)) {
+    task->fail();
+  }
+}
+
+extern "C" CDECL void aio_close_server(rust_task *task, socket_data *server,
+                                       rust_chan *chan) {
+  LOG_UPCALL_ENTRY(task);
+  // XXX: hax until rust_task::kill
+  // send null and the receiver knows to call back into native code to check
+  void* null_client = NULL;
+  server->chan->send(&null_client);
+  server->chan->deref();
+  server->chan = chan->clone(iotask);
+  aio_close_socket(task, server);
+}
+
+extern "C" CDECL bool aio_is_null_client(rust_task *task,
+                                         socket_data *server) {
+  LOG_UPCALL_ENTRY(task);
+  return server == NULL;
+}
+
+static void connection_complete(request *req, int status) {
+  LOG_CALLBACK_ENTRY(socket);
+  socket_data *client = req->socket();
+  req->send_result(client);
+  delete req;
+}
+
+extern "C" CDECL void aio_connect(rust_task *task, const char *host,
+                                  int port, rust_chan *chan) {
+  LOG_UPCALL_ENTRY(task);
+  struct sockaddr_in addr = uv_ip4_addr(const_cast<char*>(host), port);
+  request *req;
+  socket_data *client = make_socket(iotask, NULL);
+  if (!client) {
+    goto oom_client;
+  }
+  req = new (client->task, "connection request")
+    request(client, chan, connection_complete);
+  if (!req) {
+    goto oom_req;
+  }
+  if (0 == uv_connect(req, addr)) {
+    return;
+  }
+oom_req:
+  aio_close_socket(task, client);
+oom_client:
+  task->fail();
+  return;
+}
+
+static void write_complete(request *req, int status) {
+  LOG_CALLBACK_ENTRY(socket);
+  bool success = status == 0;
+  req->send_result(&success);
+  delete req;
+}
+
+extern "C" CDECL void aio_writedata(rust_task *task, socket_data *data,
+                                    char *buf, size_t size, rust_chan *chan) {
+  LOG_UPCALL_ENTRY(task);
+  uv_buf_t buffer = { buf, size };
+  request *req = new (data->task, "write request")
+    request(data, chan, write_complete);
+  if (!req) {
+    goto fail;
+  }
+  if (uv_write(req, &buffer, 1)) {
+    delete req;
+    goto fail;
+  }
+  return;
+fail:
+  task->fail();
+}
+
+extern "C" CDECL void aio_read(rust_task *task, socket_data *data,
+                               rust_chan *reader) {
+  LOG_UPCALL_ENTRY(task);
+  I(task->sched, data->reader == NULL);
+  data->reader = reader->clone(iotask);
+  uv_read_start(&data->socket, alloc_buffer, read_progress);
+}
diff --git a/src/rt/rustrt.def.in b/src/rt/rustrt.def.in
index e818b9ac83d..6860dbee8d6 100644
--- a/src/rt/rustrt.def.in
+++ b/src/rt/rustrt.def.in
@@ -1,3 +1,13 @@
+aio_close_socket
+aio_close_server
+aio_connect
+aio_init
+aio_is_null_client
+aio_read
+aio_run
+aio_serve
+aio_stop
+aio_writedata
 align_of
 chan_send
 check_claims