From 00d1465d13980fc3acf650f182ee0723fbda0e06 Mon Sep 17 00:00:00 2001 From: Michael Bebenita Date: Mon, 19 Jul 2010 14:05:18 -0700 Subject: Added a message passing system based on lock free queues for inter-thread communication. Channels now buffer on the sending side, and no longer require blocking when sending. Lots of other refactoring and bug fixes. --- src/rt/rust_chan.cpp | 54 +++++++++++++++++++++++++++++++++++++++------------- 1 file changed, 41 insertions(+), 13 deletions(-) (limited to 'src/rt/rust_chan.cpp') diff --git a/src/rt/rust_chan.cpp b/src/rt/rust_chan.cpp index 38f93a7d..6aa9121a 100644 --- a/src/rt/rust_chan.cpp +++ b/src/rt/rust_chan.cpp @@ -1,29 +1,29 @@ - #include "rust_internal.h" #include "rust_chan.h" rust_chan::rust_chan(rust_task *task, rust_port *port) : - task(task), - port(port), - buffer(task->dom, port->unit_sz), - token(this) -{ - if (port) + task(task), port(port), buffer(task->dom, port->unit_sz), token(this) { + + if (port) { port->chans.push(this); + ref(); + } + + task->log(rust_log::MEM | rust_log::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() -{ +rust_chan::~rust_chan() { if (port) { if (token.pending()) token.withdraw(); - port->chans.swapdel(this); + port->chans.swap_delete(this); } } -void -rust_chan::disassociate() -{ +void rust_chan::disassociate() { I(task->dom, port); if (token.pending()) @@ -31,4 +31,32 @@ rust_chan::disassociate() // Delete reference to the port/ port = NULL; + + deref(); +} + +/** + * Attempt to transmit channel data to the associated port. + */ +int rust_chan::transmit() { + rust_dom *dom = task->dom; + + // TODO: Figure out how and why the port would become null. + if (port == NULL) { + dom->log(rust_log::COMM, "invalid port, transmission incomplete"); + return ERROR; + } + + if (buffer.is_empty()) { + dom->log(rust_log::COMM, "buffer is empty, transmission incomplete"); + return ERROR; + } + + if(port->task->blocked_on(port)) { + buffer.dequeue(port->task->rendezvous_ptr); + port->task->wakeup(port); + } + + return 0; + } -- cgit v1.2.3