john/cell
1
0
Fork 0
Code Issues 31 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

quiesence exit

Browse Source
This commit is contained in:
John Alanbrook
2026-02-11 11:50:29 -06:00
parent da6f096a56
commit 2c55ae8cb2
4 changed files with 108 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
num_torture.cm Normal file
View File

@@ -0,0 +1,31 @@
// num_torture.cm — integer math torture test
// Pure integer arithmetic so it stays on the fast int path.
// Returns the final checksum so the caller can verify correctness.
var n = 5000000
var sum = 0
var i = 0
var a = 0
var b = 0
while (i < n) {
a = (i * 7 + 13) % 10007
b = (a * a) % 10007
sum = (sum + b) % 1000000007
i = i + 1
}
return function(n) {
var i = 0
var a = 0
var b = 0
var sum = 0
while (i < n) {
a = (i * 7 + 13) % 10007
b = (a * a) % 10007
sum = (sum + b) % 1000000007
i = i + 1
}
return sum
}

1
source/cell.c
View File

@@ -460,6 +460,7 @@ int cell_init(int argc, char **argv)
}
if (scheduler_actor_count() > 0) {
scheduler_enable_quiescence();
actor_loop();
exit_handler();
exit(0);

2
source/cell_internal.h
View File

@@ -54,6 +54,7 @@ typedef struct cell_rt {
double ar_secs; // time for unneeded
int disrupt;
int is_quiescent; // tracked by scheduler for quiescence detection
int main_thread_only;
int affinity;
@@ -82,6 +83,7 @@ void actor_loop();
void actor_initialize(void);
void actor_free(cell_rt *actor);
int scheduler_actor_count(void);
void scheduler_enable_quiescence(void);
uint64_t cell_ns();
void cell_sleep(double seconds);

92
source/scheduler.c
View File

@@ -47,9 +47,11 @@ static struct {
actor_node *main_head; // Main Thread Queue Head
actor_node *main_tail; // Main Thread Queue Tail
int shutting_down;
pthread_t *worker_threads;
int shutting_down;
int quiescence_enabled; // set after bootstrap, before actor_loop
_Atomic int quiescent_count; // actors idle with no messages and no timers
pthread_t *worker_threads;
int num_workers;
pthread_t timer_thread;
} engine;
@@ -258,6 +260,10 @@ void actor_initialize(void) {
void actor_free(cell_rt *actor)
{
if (actor->is_quiescent) {
actor->is_quiescent = 0;
atomic_fetch_sub(&engine.quiescent_count, 1);
}
lockless_shdel(actors, actor->id);
// Do not go forward with actor destruction until the actor is completely free
@@ -318,6 +324,21 @@ int scheduler_actor_count(void) {
return (int)lockless_shlen(actors);
}
void scheduler_enable_quiescence(void) {
engine.quiescence_enabled = 1;
// Check if all actors are already quiescent
int qc = atomic_load(&engine.quiescent_count);
int total = (int)lockless_shlen(actors);
if (qc >= total && total > 0) {
pthread_mutex_lock(&engine.lock);
engine.shutting_down = 1;
pthread_cond_broadcast(&engine.wake_cond);
pthread_cond_broadcast(&engine.timer_cond);
pthread_cond_broadcast(&engine.main_cond);
pthread_mutex_unlock(&engine.lock);
}
}
void exit_handler(void) {
static int already_exiting = 0;
if (already_exiting) return;
@@ -371,9 +392,13 @@ void set_actor_state(cell_rt *actor)
case ACTOR_IDLE:
if (arrlen(actor->letters)) {
if (actor->is_quiescent) {
actor->is_quiescent = 0;
atomic_fetch_sub(&engine.quiescent_count, 1);
}
actor->state = ACTOR_READY;
actor->ar = 0;
actor_node *n = malloc(sizeof(actor_node));
n->actor = actor;
n->next = NULL;
@@ -398,21 +423,46 @@ void set_actor_state(cell_rt *actor)
}
pthread_mutex_unlock(&engine.lock);
} else if (!arrlen(actor->letters) && !hmlen(actor->timers)) {
// Schedule remove timer
static uint32_t global_timer_id = 1;
uint32_t id = global_timer_id++;
actor->ar = id;
uint64_t now = cell_ns();
uint64_t execute_at = now + (uint64_t)(actor->ar_secs * 1e9);
pthread_mutex_lock(&engine.lock);
heap_push(execute_at, actor, id, TIMER_NATIVE_REMOVE);
if (timer_heap[0].timer_id == id) {
pthread_cond_signal(&engine.timer_cond);
} else if (!hmlen(actor->timers)) {
// No messages AND no timers
// Only count as quiescent if no $unneeded callback registered
int has_unneeded = !JS_IsNull(actor->unneeded_ref.val);
if (!actor->is_quiescent && actor->id && !has_unneeded) {
actor->is_quiescent = 1;
int qc = atomic_fetch_add(&engine.quiescent_count, 1) + 1;
int total = (int)lockless_shlen(actors);
if (qc >= total && total > 0 && engine.quiescence_enabled) {
pthread_mutex_lock(&engine.lock);
engine.shutting_down = 1;
pthread_cond_broadcast(&engine.wake_cond);
pthread_cond_broadcast(&engine.timer_cond);
pthread_cond_broadcast(&engine.main_cond);
pthread_mutex_unlock(&engine.lock);
}
}
if (!engine.shutting_down) {
// Schedule remove timer
static uint32_t global_timer_id = 1;
uint32_t id = global_timer_id++;
actor->ar = id;
uint64_t now = cell_ns();
uint64_t execute_at = now + (uint64_t)(actor->ar_secs * 1e9);
pthread_mutex_lock(&engine.lock);
heap_push(execute_at, actor, id, TIMER_NATIVE_REMOVE);
if (timer_heap[0].timer_id == id) {
pthread_cond_signal(&engine.timer_cond);
}
pthread_mutex_unlock(&engine.lock);
}
} else {
// Has timers but no letters — waiting, not quiescent
if (actor->is_quiescent) {
actor->is_quiescent = 0;
atomic_fetch_sub(&engine.quiescent_count, 1);
}
pthread_mutex_unlock(&engine.lock);
}
break;
}
@@ -539,6 +589,12 @@ const char *register_actor(const char *id, cell_rt *actor, int mainthread, doubl
free(actor->id);
return "Actor with given ID already exists.";
}
// Now that actor is in the registry, track its quiescent state
if (actor->state == ACTOR_IDLE && !arrlen(actor->letters)
&& !hmlen(actor->timers) && JS_IsNull(actor->unneeded_ref.val)) {
actor->is_quiescent = 1;
atomic_fetch_add(&engine.quiescent_count, 1);
}
return NULL;
}