Changeset: e73e5ebad5af for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=e73e5ebad5af
Modified Files:
monetdb5/mal/mal_dataflow.c
Branch: default
Log Message:
Merge with Jan2014 branch.
diffs (183 lines):
diff --git a/monetdb5/mal/mal_dataflow.c b/monetdb5/mal/mal_dataflow.c
--- a/monetdb5/mal/mal_dataflow.c
+++ b/monetdb5/mal/mal_dataflow.c
@@ -95,7 +95,12 @@ static struct worker {
MT_Sema s;
} workers[THREADS];
static Queue *todo = 0; /* pending instructions */
-static int volatile exiting = 0;
+#ifdef ATOMIC_LOCK
+static MT_Lock exitingLock MT_LOCK_INITIALIZER("exitingLock");
+#endif
+static volatile ATOMIC_TYPE exiting = 0;
+
+static MT_Lock dataflowLock MT_LOCK_INITIALIZER("dataflowLock");
/*
* Calculate the size of the dataflow dependency graph.
@@ -219,7 +224,7 @@ q_dequeue(Queue *q, Client cntxt)
assert(q);
MT_sema_down(&q->s, "q_dequeue");
- if (exiting)
+ if (ATOMIC_GET(exiting, exitingLock, "q_dequeue"))
return NULL;
MT_lock_set(&q->l, "q_dequeue");
if (cntxt) {
@@ -242,9 +247,9 @@ q_dequeue(Queue *q, Client cntxt)
if (q->exitcount > 0) {
q->exitcount--;
MT_lock_unset(&q->l, "q_dequeue");
- MT_lock_set(&mal_contextLock, "q_dequeue");
+ MT_lock_set(&dataflowLock, "q_dequeue");
q->exitedcount++;
- MT_lock_unset(&mal_contextLock, "q_dequeue");
+ MT_lock_unset(&dataflowLock, "q_dequeue");
return NULL;
}
assert(q->last > 0);
@@ -298,18 +303,24 @@ DFLOWworker(void *T)
Thread thr;
str error = 0;
int i,last;
+ Client cntxt;
thr = THRnew("DFLOWworker");
GDKsetbuf(GDKmalloc(GDKMAXERRLEN)); /* where to leave errors */
GDKerrbuf[0] = 0;
- if (t->cntxt) {
+ MT_lock_set(&dataflowLock, "DFLOWworker");
+ cntxt = t->cntxt;
+ MT_lock_unset(&dataflowLock, "DFLOWworker");
+ if (cntxt) {
/* wait until we are allowed to start working */
MT_sema_down(&t->s, "DFLOWworker");
}
while (1) {
if (fnxt == 0) {
- Client cntxt = t->cntxt;
+ MT_lock_set(&dataflowLock, "DFLOWworker");
+ cntxt = t->cntxt;
+ MT_lock_unset(&dataflowLock, "DFLOWworker");
fe = q_dequeue(todo, cntxt);
if (fe == NULL) {
if (cntxt) {
@@ -327,7 +338,7 @@ DFLOWworker(void *T)
}
} else
fe = fnxt;
- if (exiting) {
+ if (ATOMIC_GET(exiting, exitingLock, "DFLOWworker")) {
break;
}
fnxt = 0;
@@ -418,7 +429,9 @@ DFLOWworker(void *T)
GDKfree(GDKerrbuf);
GDKsetbuf(0);
THRdel(thr);
+ MT_lock_set(&dataflowLock, "DFLOWworker");
t->flag = EXITED;
+ MT_lock_unset(&dataflowLock, "DFLOWworker");
}
/*
@@ -463,6 +476,10 @@ DFLOWinitialize(void)
MT_lock_unset(&mal_contextLock, "DFLOWinitialize");
return -1;
}
+#ifdef NEED_MT_LOCK_INIT
+ ATOMIC_INIT(exitingLock, "exitingLock");
+ MT_lock_init(&dataflowLock, "dataflowLock");
+#endif
MT_lock_unset(&mal_contextLock, "DFLOWinitialize");
return 0;
}
@@ -655,7 +672,7 @@ DFLOWscheduler(DataFlow flow, struct wor
while (actions != tasks ) {
f = q_dequeue(flow->done, NULL);
- if (exiting)
+ if (ATOMIC_GET(exiting, exitingLock, "DFLOWscheduler"))
break;
if (f == NULL)
throw(MAL, "dataflow", "DFLOWscheduler():
q_dequeue(flow->done) returned NULL");
@@ -684,7 +701,9 @@ DFLOWscheduler(DataFlow flow, struct wor
}
/* release the worker from its specific task (turn it into a
* generic worker) */
+ MT_lock_set(&dataflowLock, "DFLOWscheduler");
w->cntxt = NULL;
+ MT_lock_unset(&dataflowLock, "DFLOWscheduler");
/* wrap up errors */
assert(flow->done->last == 0);
if (flow->error ) {
@@ -748,15 +767,22 @@ runMALdataflow(Client cntxt, MalBlkPtr m
/* in addition, create one more worker that will only execute
* tasks for the current client to compensate for our waiting
* until all work is done */
- MT_lock_set(&mal_contextLock, "runMALdataflow");
+ MT_lock_set(&dataflowLock, "runMALdataflow");
/* join with already exited threads */
- for (i = 0; i < THREADS && todo->exitedcount > 0; i++) {
- if (workers[i].flag == EXITED) {
- todo->exitedcount--;
- workers[i].flag = IDLE;
- workers[i].cntxt = NULL;
- MT_join_thread(workers[i].id);
+ while (todo->exitedcount > 0) {
+ for (i = 0; i < THREADS; i++) {
+ if (workers[i].flag == EXITED) {
+ todo->exitedcount--;
+ workers[i].flag = IDLE;
+ workers[i].cntxt = NULL;
+ MT_lock_unset(&dataflowLock, "runMALdataflow");
+ MT_join_thread(workers[i].id);
+ MT_lock_set(&dataflowLock, "runMALdataflow");
+ break;
+ }
}
+ if (i == THREADS)
+ break;
}
for (i = 0; i < THREADS; i++) {
if (workers[i].flag == IDLE) {
@@ -782,14 +808,14 @@ runMALdataflow(Client cntxt, MalBlkPtr m
if (MT_create_thread(&workers[i].id, DFLOWworker, (void
*) &workers[i], MT_THR_JOINABLE) < 0) {
/* cannot start new thread, run serially */
*ret = TRUE;
- MT_lock_unset(&mal_contextLock,
"runMALdataflow");
+ MT_lock_unset(&dataflowLock, "runMALdataflow");
return MAL_SUCCEED;
}
workers[i].flag = RUNNING;
break;
}
}
- MT_lock_unset(&mal_contextLock, "runMALdataflow");
+ MT_lock_unset(&dataflowLock, "runMALdataflow");
if (i == THREADS) {
/* no empty thread slots found, run serially */
*ret = TRUE;
@@ -870,14 +896,19 @@ stopMALdataflow(void)
{
int i;
- exiting = 1;
+ ATOMIC_SET(exiting, 1, exitingLock, "q_dequeue");
if (todo) {
for (i = 0; i < THREADS; i++)
MT_sema_up(&todo->s, "stopMALdataflow");
+ MT_lock_set(&dataflowLock, "stopMALdataflow");
for (i = 0; i < THREADS; i++) {
- if (workers[i].flag != IDLE)
+ if (workers[i].flag != IDLE) {
+ MT_lock_unset(&dataflowLock, "stopMALdataflow");
MT_join_thread(workers[i].id);
+ MT_lock_set(&dataflowLock, "stopMALdataflow");
+ }
workers[i].flag = IDLE;
}
+ MT_lock_unset(&dataflowLock, "stopMALdataflow");
}
}
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
