MonetDB: default - Better cleanup of dataflow threads.

[Sjoerd Mullender via checkin-list]

Changeset: 3e625a4c333d for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/3e625a4c333d
Modified Files:
        monetdb5/mal/mal_dataflow.c
Branch: default
Log Message:

Better cleanup of dataflow threads.


diffs (92 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
@@ -82,7 +82,7 @@ typedef struct DATAFLOW {
 
 struct worker {
        MT_Id id;
-       enum { WAITING, RUNNING, FREE, EXITED } flag;
+       enum { WAITING, RUNNING, FREE, EXITED, FINISHING } flag;
        ATOMIC_PTR_TYPE cntxt;          /* client we do work for (NULL -> any) 
*/
        MT_Sema s;
        struct worker *next;
@@ -245,6 +245,7 @@ static void
 DFLOWworker(void *T)
 {
        struct worker *t = (struct worker *) T;
+       bool locked = false;
 #ifdef _MSC_VER
        srand((unsigned int) GDKusec());
 #endif
@@ -411,20 +412,8 @@ DFLOWworker(void *T)
                        }
                }
                MT_lock_set(&dataflowLock);
-               if (GDKexiting() || ATOMIC_GET(&exiting)) {
-                       MT_lock_unset(&dataflowLock);
-                       break;
-               }
-               if (free_count >= free_max) {
-                       struct worker **tp = &workers;
-                       while (*tp && *tp != t)
-                               tp = &(*tp)->next;
-                       assert(*tp && *tp == t);
-                       *tp = t->next;
-                       t->flag = EXITED;
-                       t->next = exited_workers;
-                       exited_workers = t;
-                       MT_lock_unset(&dataflowLock);
+               if (GDKexiting() || ATOMIC_GET(&exiting) || free_count >= 
free_max) {
+                       locked = true;
                        break;
                }
                free_count++;
@@ -442,6 +431,19 @@ DFLOWworker(void *T)
                        break;
                assert(t->flag == WAITING);
        }
+       if (!locked)
+               MT_lock_set(&dataflowLock);
+       if (t->flag != FINISHING) {
+               struct worker **tp = t->flag == FREE ? &free_workers : &workers;
+               while (*tp && *tp != t)
+                       tp = &(*tp)->next;
+               assert(*tp && *tp == t);
+               *tp = t->next;
+               t->flag = EXITED;
+               t->next = exited_workers;
+               exited_workers = t;
+       }
+       MT_lock_unset(&dataflowLock);
        GDKsetbuf(NULL);
 }
 
@@ -747,6 +749,7 @@ DFLOWscheduler(DataFlow flow, struct wor
 static inline void
 finish_worker(struct worker *t)
 {
+       t->flag = FINISHING;
        MT_lock_unset(&dataflowLock);
        MT_join_thread(t->id);
        MT_sema_destroy(&t->s);
@@ -944,6 +947,8 @@ stopMALdataflow(void)
                }
                while (free_workers) {
                        struct worker *t = free_workers;
+                       assert(free_count > 0);
+                       free_count--;
                        free_workers = free_workers->next;
                        MT_sema_up(&t->s);
                        finish_worker(t);
@@ -953,6 +958,11 @@ stopMALdataflow(void)
                        workers = workers->next;
                        finish_worker(t);
                }
+               while (exited_workers) {
+                       struct worker *t = exited_workers;
+                       exited_workers = exited_workers->next;
+                       finish_worker(t);
+               }
                MT_lock_unset(&dataflowLock);
        }
 }
_______________________________________________
checkin-list mailing list -- checkin-list@monetdb.org
To unsubscribe send an email to checkin-list-le...@monetdb.org