Per-thread scheduling. Closes #682.

Tasks are spawned on a random thread. Currently they stay there, but
we should add task migration and load balancing in the future. This
should drammatically improve our task performance benchmarks.

Author: Eric Holk

src/rt/circular_buffer.cpp

@@ -5,34 +5,33 @@
 #include "rust_internal.h"
 
 circular_buffer::circular_buffer(rust_kernel *kernel, size_t unit_sz) :
-    sched(kernel->sched),
     kernel(kernel),
     unit_sz(unit_sz),
     _buffer_sz(initial_size()),
     _next(0),
     _unread(0),
     _buffer((uint8_t *)kernel->malloc(_buffer_sz, "circular_buffer")) {
 
-    A(sched, unit_sz, "Unit size must be larger than zero.");
+    // A(sched, unit_sz, "Unit size must be larger than zero.");
 
-    DLOG(sched, mem, "new circular_buffer(buffer_sz=%d, unread=%d)"
-         "-> circular_buffer=0x%" PRIxPTR,
-         _buffer_sz, _unread, this);
+    // DLOG(sched, mem, "new circular_buffer(buffer_sz=%d, unread=%d)"
+    //      "-> circular_buffer=0x%" PRIxPTR,
+    //      _buffer_sz, _unread, this);
 
-    A(sched, _buffer, "Failed to allocate buffer.");
+    // A(sched, _buffer, "Failed to allocate buffer.");
 }
 
 circular_buffer::~circular_buffer() {
-    DLOG(sched, mem, "~circular_buffer 0x%" PRIxPTR, this);
-    I(sched, _buffer);
-    W(sched, _unread == 0,
-      "freeing circular_buffer with %d unread bytes", _unread);
+    // DLOG(sched, mem, "~circular_buffer 0x%" PRIxPTR, this);
+    // I(sched, _buffer);
+    // W(sched, _unread == 0,
+    //   "freeing circular_buffer with %d unread bytes", _unread);
     kernel->free(_buffer);
 }
 
 size_t
 circular_buffer::initial_size() {
-    I(sched, unit_sz > 0);
+    // I(sched, unit_sz > 0);
     return INITIAL_CIRCULAR_BUFFER_SIZE_IN_UNITS * unit_sz;
 }
 
@@ -41,8 +40,8 @@ circular_buffer::initial_size() {
  */
 void
 circular_buffer::transfer(void *dst) {
-    I(sched, dst);
-    I(sched, _unread <= _buffer_sz);
+    // I(sched, dst);
+    // I(sched, _unread <= _buffer_sz);
 
     uint8_t *ptr = (uint8_t *) dst;
 
@@ -54,13 +53,13 @@ circular_buffer::transfer(void *dst) {
     } else {
         head_sz = _buffer_sz - _next;
     }
-    I(sched, _next + head_sz <= _buffer_sz);
+    // I(sched, _next + head_sz <= _buffer_sz);
     memcpy(ptr, _buffer + _next, head_sz);
 
     // Then copy any other items from the beginning of the buffer
-    I(sched, _unread >= head_sz);
+    // I(sched, _unread >= head_sz);
     size_t tail_sz = _unread - head_sz;
-    I(sched, head_sz + tail_sz <= _buffer_sz);
+    // I(sched, head_sz + tail_sz <= _buffer_sz);
     memcpy(ptr + head_sz, _buffer, tail_sz);
 }
 
@@ -70,37 +69,37 @@ circular_buffer::transfer(void *dst) {
  */
 void
 circular_buffer::enqueue(void *src) {
-    I(sched, src);
-    I(sched, _unread <= _buffer_sz);
-    I(sched, _buffer);
+    // I(sched, src);
+    // I(sched, _unread <= _buffer_sz);
+    // I(sched, _buffer);
 
     // Grow if necessary.
     if (_unread == _buffer_sz) {
         grow();
     }
 
-    DLOG(sched, mem, "circular_buffer enqueue "
-         "unread: %d, next: %d, buffer_sz: %d, unit_sz: %d",
-         _unread, _next, _buffer_sz, unit_sz);
+    // DLOG(sched, mem, "circular_buffer enqueue "
+    //      "unread: %d, next: %d, buffer_sz: %d, unit_sz: %d",
+    //      _unread, _next, _buffer_sz, unit_sz);
 
-    I(sched, _unread < _buffer_sz);
-    I(sched, _unread + unit_sz <= _buffer_sz);
+    // I(sched, _unread < _buffer_sz);
+    // I(sched, _unread + unit_sz <= _buffer_sz);
 
     // Copy data
     size_t dst_idx = _next + _unread;
     I(sched, dst_idx >= _buffer_sz || dst_idx + unit_sz <= _buffer_sz);
     if (dst_idx >= _buffer_sz) {
         dst_idx -= _buffer_sz;
 
-        I(sched, _next >= unit_sz);
-        I(sched, dst_idx <= _next - unit_sz);
+        // I(sched, _next >= unit_sz);
+        // I(sched, dst_idx <= _next - unit_sz);
     }
 
-    I(sched, dst_idx + unit_sz <= _buffer_sz);
+    // I(sched, dst_idx + unit_sz <= _buffer_sz);
     memcpy(&_buffer[dst_idx], src, unit_sz);
     _unread += unit_sz;
 
-    DLOG(sched, mem, "circular_buffer pushed data at index: %d", dst_idx);
+    // DLOG(sched, mem, "circular_buffer pushed data at index: %d", dst_idx);
 }
 
 /**
@@ -110,17 +109,17 @@ circular_buffer::enqueue(void *src) {
  */
 void
 circular_buffer::dequeue(void *dst) {
-    I(sched, unit_sz > 0);
-    I(sched, _unread >= unit_sz);
-    I(sched, _unread <= _buffer_sz);
-    I(sched, _buffer);
+    // I(sched, unit_sz > 0);
+    // I(sched, _unread >= unit_sz);
+    // I(sched, _unread <= _buffer_sz);
+    // I(sched, _buffer);
 
-    DLOG(sched, mem,
-             "circular_buffer dequeue "
-             "unread: %d, next: %d, buffer_sz: %d, unit_sz: %d",
-             _unread, _next, _buffer_sz, unit_sz);
+    // DLOG(sched, mem,
+    //          "circular_buffer dequeue "
+    //          "unread: %d, next: %d, buffer_sz: %d, unit_sz: %d",
+    //          _unread, _next, _buffer_sz, unit_sz);
 
-    I(sched, _next + unit_sz <= _buffer_sz);
+    // I(sched, _next + unit_sz <= _buffer_sz);
     if (dst != NULL) {
         memcpy(dst, &_buffer[_next], unit_sz);
     }
@@ -140,8 +139,9 @@ circular_buffer::dequeue(void *dst) {
 void
 circular_buffer::grow() {
     size_t new_buffer_sz = _buffer_sz * 2;
-    I(sched, new_buffer_sz <= MAX_CIRCULAR_BUFFER_SIZE);
-    DLOG(sched, mem, "circular_buffer is growing to %d bytes", new_buffer_sz);
+    // I(sched, new_buffer_sz <= MAX_CIRCULAR_BUFFER_SIZE);
+    // DLOG(sched, mem, "circular_buffer is growing to %d bytes",
+    //      new_buffer_sz);
     void *new_buffer = kernel->malloc(new_buffer_sz,
                                     "new circular_buffer (grow)");
     transfer(new_buffer);
@@ -154,9 +154,9 @@ circular_buffer::grow() {
 void
 circular_buffer::shrink() {
     size_t new_buffer_sz = _buffer_sz / 2;
-    I(sched, initial_size() <= new_buffer_sz);
-    DLOG(sched, mem, "circular_buffer is shrinking to %d bytes",
-         new_buffer_sz);
+    // I(sched, initial_size() <= new_buffer_sz);
+    // DLOG(sched, mem, "circular_buffer is shrinking to %d bytes",
+    //      new_buffer_sz);
     void *new_buffer = kernel->malloc(new_buffer_sz,
                                     "new circular_buffer (shrink)");
     transfer(new_buffer);

src/rt/rust.cpp

@@ -140,9 +140,10 @@ rust_start(uintptr_t main_fn, int argc, char **argv, void* crate_map) {
 
     update_log_settings(crate_map, getenv("RUST_LOG"));
     enable_claims(getenv("CHECK_CLAIMS"));
+    int num_threads = get_num_threads();
 
     rust_srv *srv = new rust_srv();
-    rust_kernel *kernel = new rust_kernel(srv);
+    rust_kernel *kernel = new rust_kernel(srv, num_threads);
     kernel->start();
     rust_task *root_task = kernel->create_task(NULL, "main");
     rust_scheduler *sched = root_task->sched;
@@ -158,11 +159,9 @@ rust_start(uintptr_t main_fn, int argc, char **argv, void* crate_map) {
 
     root_task->start(main_fn, (uintptr_t)args->args);
 
-    int num_threads = get_num_threads();
-
     DLOG(sched, dom, "Using %d worker threads.", num_threads);
 
-    int ret = kernel->start_task_threads(num_threads);
+    int ret = kernel->start_task_threads();
     delete args;
     delete kernel;
     delete srv;

src/rt/rust_chan.cpp

@@ -13,17 +13,17 @@ rust_chan::rust_chan(rust_kernel *kernel, maybe_proxy<rust_port> *port,
     if (port) {
         associate(port);
     }
-    DLOG(kernel->sched, comm, "new rust_chan(task=0x%" PRIxPTR
-        ", port=0x%" PRIxPTR ") -> chan=0x%" PRIxPTR,
-        (uintptr_t) task, (uintptr_t) port, (uintptr_t) this);
+    // DLOG(task->sched, comm, "new rust_chan(task=0x%" PRIxPTR
+    //     ", port=0x%" PRIxPTR ") -> chan=0x%" PRIxPTR,
+    //     (uintptr_t) task, (uintptr_t) port, (uintptr_t) this);
 }
 
 rust_chan::~rust_chan() {
-    DLOG(kernel->sched, comm, "del rust_chan(task=0x%" PRIxPTR ")",
-         (uintptr_t) this);
+    // DLOG(kernel->sched, comm, "del rust_chan(task=0x%" PRIxPTR ")",
+    //      (uintptr_t) this);
 
-    A(kernel->sched, is_associated() == false,
-      "Channel must be disassociated before being freed.");
+    // A(kernel->sched, is_associated() == false,
+    //   "Channel must be disassociated before being freed.");
 }
 
 /**
@@ -33,9 +33,9 @@ void rust_chan::associate(maybe_proxy<rust_port> *port) {
     this->port = port;
     if (port->is_proxy() == false) {
         scoped_lock with(port->referent()->lock);
-        DLOG(kernel->sched, task,
-            "associating chan: 0x%" PRIxPTR " with port: 0x%" PRIxPTR,
-            this, port);
+        // DLOG(kernel->sched, task,
+        //     "associating chan: 0x%" PRIxPTR " with port: 0x%" PRIxPTR,
+        //     this, port);
         ++this->ref_count;
         this->task = port->referent()->task;
         this->task->ref();
@@ -51,14 +51,14 @@ bool rust_chan::is_associated() {
  * Unlink this channel from its associated port.
  */
 void rust_chan::disassociate() {
-    A(kernel->sched, is_associated(),
-      "Channel must be associated with a port.");
+    // A(kernel->sched, is_associated(),
+    //   "Channel must be associated with a port.");
 
     if (port->is_proxy() == false) {
         scoped_lock with(port->referent()->lock);
-        DLOG(kernel->sched, task,
-            "disassociating chan: 0x%" PRIxPTR " from port: 0x%" PRIxPTR,
-            this, port->referent());
+        // DLOG(kernel->sched, task,
+        //     "disassociating chan: 0x%" PRIxPTR " from port: 0x%" PRIxPTR,
+        //     this, port->referent());
         --this->ref_count;
         --this->task->ref_count;
         this->task = NULL;
@@ -73,8 +73,8 @@ void rust_chan::disassociate() {
  * Attempt to send data to the associated port.
  */
 void rust_chan::send(void *sptr) {
-    rust_scheduler *sched = kernel->sched;
-    I(sched, !port->is_proxy());
+    // rust_scheduler *sched = kernel->sched;
+    // I(sched, !port->is_proxy());
 
     rust_port *target_port = port->referent();
     // TODO: We can probably avoid this lock by using atomic operations in
@@ -84,21 +84,21 @@ void rust_chan::send(void *sptr) {
     buffer.enqueue(sptr);
 
     if (!is_associated()) {
-        W(sched, is_associated(),
-          "rust_chan::transmit with no associated port.");
+        // W(sched, is_associated(),
+        //   "rust_chan::transmit with no associated port.");
         return;
     }
 
-    A(sched, !buffer.is_empty(),
-      "rust_chan::transmit with nothing to send.");
+    // A(sched, !buffer.is_empty(),
+    //   "rust_chan::transmit with nothing to send.");
 
     if (port->is_proxy()) {
         data_message::send(buffer.peek(), buffer.unit_sz, "send data",
                            task->get_handle(), port->as_proxy()->handle());
         buffer.dequeue(NULL);
     } else {
         if (target_port->task->blocked_on(target_port)) {
-            DLOG(sched, comm, "dequeued in rendezvous_ptr");
+            // DLOG(sched, comm, "dequeued in rendezvous_ptr");
             buffer.dequeue(target_port->task->rendezvous_ptr);
             target_port->task->rendezvous_ptr = 0;
             target_port->task->wakeup(target_port);
@@ -120,7 +120,7 @@ rust_chan *rust_chan::clone(maybe_proxy<rust_task> *target) {
         rust_handle<rust_port> *handle =
             task->sched->kernel->get_port_handle(port->as_referent());
         maybe_proxy<rust_port> *proxy = new rust_proxy<rust_port> (handle);
-        DLOG(kernel->sched, mem, "new proxy: " PTR, proxy);
+        DLOG(task->sched, mem, "new proxy: " PTR, proxy);
         port = proxy;
         target_task = target->as_proxy()->handle()->referent();
     }
@@ -133,8 +133,8 @@ rust_chan *rust_chan::clone(maybe_proxy<rust_task> *target) {
  * appear to be live, causing modify-after-free errors.
  */
 void rust_chan::destroy() {
-    A(kernel->sched, ref_count == 0,
-      "Channel's ref count should be zero.");
+    // A(kernel->sched, ref_count == 0,
+    //   "Channel's ref count should be zero.");
 
     if (is_associated()) {
         if (port->is_proxy()) {