aboutsummaryrefslogtreecommitdiff
path: root/src/rt/rust_upcall.cpp
diff options
context:
space:
mode:
Diffstat (limited to 'src/rt/rust_upcall.cpp')
-rw-r--r--src/rt/rust_upcall.cpp96
1 files changed, 64 insertions, 32 deletions
diff --git a/src/rt/rust_upcall.cpp b/src/rt/rust_upcall.cpp
index b0757e29..90d6f6d9 100644
--- a/src/rt/rust_upcall.cpp
+++ b/src/rt/rust_upcall.cpp
@@ -6,11 +6,13 @@
#define LOG_UPCALL_ENTRY(task) \
(task)->dom->get_log().reset_indent(0); \
(task)->log(rust_log::UPCALL, \
- "> UPCALL %s - task: %s @0x%" PRIxPTR \
- " retpc: x%" PRIxPTR, \
+ "> UPCALL %s - task: %s 0x%" PRIxPTR \
+ " retpc: x%" PRIxPTR \
+ " ref_count: %d", \
__FUNCTION__, \
(task)->name, (task), \
- __builtin_return_address(0)); \
+ __builtin_return_address(0), \
+ (task->ref_count)); \
(task)->dom->get_log().indent();
#else
#define LOG_UPCALL_ENTRY(task) \
@@ -21,6 +23,20 @@
(task)->dom->get_log().indent();
#endif
+void
+log_task_state(rust_task *task, maybe_proxy<rust_task> *target) {
+ rust_task *delegate = target->delegate();
+ if (target->is_proxy()) {
+ task->log(rust_log::TASK,
+ "remote task: 0x%" PRIxPTR ", ref_count: %d state: %s",
+ delegate, delegate->ref_count, delegate->state_str());
+ } else {
+ task->log(rust_log::TASK,
+ "local task: 0x%" PRIxPTR ", ref_count: %d state: %s",
+ delegate, delegate->ref_count, delegate->state_str());
+ }
+}
+
extern "C" CDECL char const *str_buf(rust_task *task, rust_str *s);
extern "C" void upcall_grow_task(rust_task *task, size_t n_frame_bytes) {
@@ -31,14 +47,13 @@ extern "C" void upcall_grow_task(rust_task *task, size_t n_frame_bytes) {
extern "C" CDECL void upcall_log_int(rust_task *task, int32_t i) {
LOG_UPCALL_ENTRY(task);
task->log(rust_log::UPCALL | rust_log::ULOG,
- "upcall log_int(0x%" PRIx32 " = %" PRId32 " = '%c')", i, i,
- (char) i);
+ "rust: %" PRId32 " (0x%" PRIx32 ")", i, i);
}
extern "C" CDECL void upcall_log_str(rust_task *task, rust_str *str) {
LOG_UPCALL_ENTRY(task);
const char *c = str_buf(task, str);
- task->log(rust_log::UPCALL | rust_log::ULOG, "upcall log_str(\"%s\")", c);
+ task->log(rust_log::UPCALL | rust_log::ULOG, "rust: %s", c);
}
extern "C" CDECL void upcall_trace_word(rust_task *task, uintptr_t i) {
@@ -86,26 +101,50 @@ upcall_new_chan(rust_task *task, rust_port *port) {
}
/**
+ * Called whenever this channel needs to be flushed. This can happen due to a
+ * flush statement, or automatically whenever a channel's ref count is
+ * about to drop to zero.
+ */
+extern "C" CDECL void
+upcall_flush_chan(rust_task *task, rust_chan *chan) {
+ LOG_UPCALL_ENTRY(task);
+ rust_dom *dom = task->dom;
+ task->log(rust_log::UPCALL | rust_log::COMM,
+ "flush chan: 0x%" PRIxPTR, chan);
+
+ if (chan->buffer.is_empty()) {
+ return;
+ }
+
+ A(dom, chan->port->is_proxy() == false,
+ "Channels to remote ports should be flushed automatically.");
+
+ // Block on the port until this channel has been completely drained
+ // by the port.
+ task->block(chan->port);
+ task->yield(2);
+}
+
+/**
* Called whenever the channel's ref count drops to zero.
+ *
+ * Cannot Yield: If the task were to unwind, the dropped ref would still
+ * appear to be live, causing modify-after-free errors.
*/
extern "C" CDECL void upcall_del_chan(rust_task *task, rust_chan *chan) {
LOG_UPCALL_ENTRY(task);
- rust_dom *dom = task->dom;
+
task->log(rust_log::UPCALL | rust_log::MEM | rust_log::COMM,
"upcall del_chan(0x%" PRIxPTR ")", (uintptr_t) chan);
- I(dom, !chan->ref_count);
-
- if (!chan->buffer.is_empty() && chan->is_associated()) {
- A(dom, !chan->port->is_proxy(),
- "Channels to remote ports should be flushed automatically.");
- // A target port may still be reading from this channel.
- // Block on this channel until it has been completely drained
- // by the port.
- task->block(chan);
- task->yield(2);
- return;
- }
+ A(task->dom, chan->ref_count == 0,
+ "Channel's ref count should be zero.");
+
+ if (chan->is_associated()) {
+ A(task->dom, chan->buffer.is_empty(),
+ "Channel's buffer should be empty.");
+ chan->disassociate();
+ }
delete chan;
}
@@ -171,8 +210,7 @@ upcall_send(rust_task *task, rust_chan *chan, void *sptr) {
"chan: 0x%" PRIxPTR ", sptr: 0x%" PRIxPTR ", size: %d",
(uintptr_t) chan, (uintptr_t) sptr,
chan->port->delegate()->unit_sz);
- chan->buffer.enqueue(sptr);
- chan->transmit();
+ chan->send(sptr);
task->log(rust_log::COMM, "=== sent data ===>");
}
@@ -185,17 +223,8 @@ upcall_recv(rust_task *task, uintptr_t *dptr, rust_port *port) {
(uintptr_t) port, (uintptr_t) dptr, port->unit_sz,
port->chans.length());
- for (uint32_t i = 0; i < port->chans.length(); i++) {
- rust_chan *chan = port->chans[i];
- if (chan->buffer.is_empty() == false) {
- chan->buffer.dequeue(dptr);
- if (chan->buffer.is_empty() && chan->task->blocked()) {
- chan->task->wakeup(chan);
- delete chan;
- }
- task->log(rust_log::COMM, "<=== read data ===");
- return;
- }
+ if (port->receive(dptr)) {
+ return;
}
// No data was buffered on any incoming channel, so block this task
@@ -222,6 +251,7 @@ extern "C" CDECL void upcall_fail(rust_task *task, char const *expr,
extern "C" CDECL void
upcall_kill(rust_task *task, maybe_proxy<rust_task> *target) {
LOG_UPCALL_ENTRY(task);
+ log_task_state(task, target);
rust_task *target_task = target->delegate();
task->log(rust_log::UPCALL | rust_log::TASK,
@@ -247,6 +277,8 @@ upcall_exit(rust_task *task) {
LOG_UPCALL_ENTRY(task);
task->log(rust_log::UPCALL | rust_log::TASK,
"task ref_count: %d", task->ref_count);
+ A(task->dom, task->ref_count >= 0,
+ "Task ref_count should not be negative on exit!");
task->die();
task->notify_tasks_waiting_to_join();
task->yield(1);
generated by cgit v1.2.3 (git 2.25.1) at 2026-04-21 23:16:20 +0000