syndicate-lang
/
hop-2012
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Make nap() interruptable. Use nap() in shoveller and exit if idle.

This commit is contained in:
Tony Garnock-Jones 2011-01-10 14:33:42 -05:00
parent 5913f2008a
commit e0ca281c32
3 changed files with 32 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
server/harness.c
View File

@ -117,24 +117,37 @@ Process *spawn(void (*f)(void *), void *arg) {
return p; return p;
} }
typedef struct nap_context_t_ {
Process *p;
int timeout_fired;
} nap_context_t;
void nap_isr(int fd, short what, void *arg) { void nap_isr(int fd, short what, void *arg) {
Process *p = (Process *) arg; nap_context_t *context = arg;
//info("nap_isr %p\n", p); //info("nap_isr %p\n", p);
assert((p->state == PROCESS_WAITING) && (p->wait_flags & EV_TIMEOUT)); assert((context->p->state == PROCESS_WAITING) && (context->p->wait_flags & EV_TIMEOUT));
p->wait_flags &= ~EV_TIMEOUT; context->timeout_fired = 1;
enqueue_runlist(p); enqueue_runlist(context->p);
} }
void nap(long millis) { int nap(long millis) {
struct event ev;
struct timeval tv; struct timeval tv;
nap_context_t context;
assert(current_process != NULL); assert(current_process != NULL);
assert(current_process->state == PROCESS_RUNNING); assert(current_process->state == PROCESS_RUNNING);
context.p = current_process;
context.timeout_fired = 0;
tv.tv_sec = millis / 1000; tv.tv_sec = millis / 1000;
tv.tv_usec = (millis % 1000) * 1000; tv.tv_usec = (millis % 1000) * 1000;
ICHECK(event_once(-1, EV_TIMEOUT, nap_isr, current_process, &tv), "event_once"); evtimer_set(&ev, nap_isr, &context);
ICHECK(evtimer_add(&ev, &tv), "evtimer_add");
current_process->state = PROCESS_WAITING; current_process->state = PROCESS_WAITING;
current_process->wait_flags |= EV_TIMEOUT; current_process->wait_flags |= EV_TIMEOUT;
schedule(); schedule();
current_process->wait_flags &= ~EV_TIMEOUT;
evtimer_del(&ev);
return context.timeout_fired;
} }
static void awaken_waiters(IOHandle *h, short mask) { static void awaken_waiters(IOHandle *h, short mask) {

2
server/harness.h
View File

@ -37,7 +37,7 @@ extern Process *current_process;
extern void yield(void); extern void yield(void);
extern Process *spawn(process_main_t f, void *arg); extern Process *spawn(process_main_t f, void *arg);
extern void nap(long millis); extern int nap(long millis); /* 1 for timeout expired; 0 for resumed early */
extern void suspend(void); extern void suspend(void);
extern void resume(Process *p); extern void resume(Process *p);

14
server/queue.c
View File

@ -91,6 +91,11 @@ static void shoveller(void *qv) {
sexp_t *body = NULL; /* held */ sexp_t *body = NULL; /* held */
subscription_t *sub = NULL; subscription_t *sub = NULL;
{
cmsg_bytes_t n = sexp_data(q->name);
info("Queue <<%.*s>> busy. Shoveller entering\n", n.len, n.bytes);
}
check_for_work: check_for_work:
//info("Checking for work\n"); //info("Checking for work\n");
@ -139,19 +144,24 @@ static void shoveller(void *qv) {
end_burst(q, &burst_count, total_count); end_burst(q, &burst_count, total_count);
//info("Waiting for throck\n"); //info("Waiting for throck\n");
q->shovel_awake = 0; q->shovel_awake = 0;
suspend(); if (nap(100)) {
cmsg_bytes_t n = sexp_data(q->name);
info("Queue <<%.*s>> idle. Shoveller exiting\n", n.len, n.bytes);
q->shovel = NULL;
return;
}
//info("Throck received!\n"); //info("Throck received!\n");
goto check_for_work; goto check_for_work;
} }
static void throck_shovel(queue_extension_t *q) { static void throck_shovel(queue_extension_t *q) {
if (!q->shovel_awake) { if (!q->shovel_awake) {
q->shovel_awake = 1;
if (!q->shovel) { if (!q->shovel) {
q->shovel = spawn(shoveller, q); q->shovel = spawn(shoveller, q);
} else { } else {
resume(q->shovel); resume(q->shovel);
} }
q->shovel_awake = 1;
} }
} }