add ability to set destructor in QueuePerThreadPool

this is needed when the queues hold items that need additional work to
free above just calling free().

so that they can clean up everything on exit properly.

Author: bertschinger

include/QueuePerThreadPool.h

@@ -115,6 +115,7 @@ QPTPool_t *QPTPool_init_with_props(const size_t nthreads, void *args,
                                    const uint64_t queue_limit, const char *swap_prefix,
                                    const uint64_t steal_num, const uint64_t steal_denom);
 
+void QPTPool_set_destructor(QPTPool_t *pool, void (*destructor)(void *));
 /*
  * QPTPool_init only allocates memory - call this to start threads
  *

include/bf.h

@@ -360,7 +360,7 @@ struct work {
 size_t struct_work_size(struct work *w);
 struct work *new_work_with_name(const char *prefix, const size_t prefix_len,
                                 const char *basename, const size_t basename_len);
-void free_work(struct work *w);
+void free_work(void *p);
 
 /*
  * A reference-counted directory handle.

src/QueuePerThreadPool.c

@@ -119,6 +119,12 @@ struct QPTPool {
     uint64_t swapped;
 
     QPTPoolThreadData_t *data;
+
+    /*
+     * An optional pointer to a destructor for the work items in this QPTPool.
+     * If it is not specified, then the destructor is just free().
+     */
+    void (*destructor)(void *);
 };
 
 /* struct to pass into pthread_create */
@@ -551,6 +557,10 @@ int QPTPool_get_steal(QPTPool_t *ctx, uint64_t *num, uint64_t *denom) {
     return 0;
 }
 
+void QPTPool_set_destructor(QPTPool_t *pool, void (*destructor)(void *)) {
+    pool->destructor = destructor;
+}
+
 QPTPool_t *QPTPool_init_with_props(const size_t nthreads, void *args,
                                    QPTPoolNextFunc_t next_func, void *next_args,
                                    const uint64_t queue_limit, const char *swap_prefix,
@@ -574,6 +584,8 @@ QPTPool_t *QPTPool_init_with_props(const size_t nthreads, void *args,
     ctx->incomplete = 0;
     ctx->swapped = 0;
 
+    ctx->destructor = NULL;
+
     /* this can fail since nthreads is user input */
     ctx->data = calloc(nthreads, sizeof(QPTPoolThreadData_t));
     if (!ctx->data) {
@@ -1057,6 +1069,13 @@ void QPTPool_destroy(QPTPool_t *ctx) {
         return;
     }
 
+    void (*destructor)(void *);
+    if (ctx->destructor) {
+        destructor = ctx->destructor;
+    } else {
+        destructor = free;
+    }
+
     for(size_t i = 0; i < ctx->nthreads; i++) {
         QPTPoolThreadData_t *data = &ctx->data[i];
         data->threads_successful = 0;
@@ -1073,9 +1092,9 @@ void QPTPool_destroy(QPTPool_t *ctx) {
          * enqueuing work without starting the worker threads is allowed,
          * so free() is called to clear out any unprocessed work items
          */
-        sll_destroy(&data->waiting, free);
+        sll_destroy(&data->waiting, destructor);
         pthread_mutex_destroy(&data->claimed_mutex);
-        sll_destroy(&data->claimed, free);
+        sll_destroy(&data->claimed, destructor);
     }
 
     pthread_cond_destroy(&ctx->cv);

src/bf.c

@@ -695,7 +695,8 @@ struct work *new_work_with_name(const char *prefix, const size_t prefix_len,
     return w;
 }
 
-void free_work(struct work *w) {
+void free_work(void *p) {
+    struct work *w = (struct work *) p;
     dir_dec(w->parent_dir);
     free(w);
 }

test/unit/googletest/descend.cpp.in

@@ -100,6 +100,7 @@ TEST(descend, builddir) {
     QPTPool_t *pool = QPTPool_init(1, nullptr);
     ASSERT_NE(pool, nullptr);
     EXPECT_EQ(QPTPool_start(pool), 0);
+    QPTPool_set_destructor(pool, (void (*)(void *)) free_work);
 
     struct descend_counters ctrs;
 
@@ -130,7 +131,7 @@ TEST(descend, builddir) {
         work->level = 0;
         EXPECT_EQ(descend(pool, 0, nullptr, &in, work, 0, dir_rc, 0,
                           [](QPTPool_t *, const size_t, void *data, void *) -> int {
-                              free(data);
+                              free_work(data);
                               return 0;
                           },
                           nullptr, nullptr,
@@ -148,7 +149,7 @@ TEST(descend, builddir) {
         rewinddir(dir);
         EXPECT_EQ(descend(pool, 0, nullptr, &in, work, 0, dir_rc, 1,
                           [](QPTPool_t *, const size_t, void *data, void *) -> int {
-                              free(data);
+                              free_work(data);
                               return 0;
                           },
                           nullptr, nullptr,
@@ -166,7 +167,7 @@ TEST(descend, builddir) {
         in.subdir_limit = 1;
         EXPECT_EQ(descend(pool, 0, nullptr, &in, work, 0, dir_rc, 0,
                           [](QPTPool_t *, const size_t, void *data, void *) -> int {
-                              free(data);
+                              free_work(data);
                               return 0;
                           },
                           nullptr, nullptr,
@@ -181,9 +182,9 @@ TEST(descend, builddir) {
     QPTPool_stop(pool);
     QPTPool_destroy(pool);
 
-    EXPECT_EQ(closedir(dir),    0);
     free_work(work);
     dir_dec(dir_rc);
+
     EXPECT_EQ(unlink(pipename), 0);
     EXPECT_EQ(unlink(linkname), 0);
 }
@@ -210,6 +211,7 @@ TEST(descend, swap) {
     ASSERT_NE(pool, nullptr);
     EXPECT_EQ(QPTPool_set_queue_limit(pool, 1), 0); // if the WAIT queue already has 1 item, swap out the new item
     EXPECT_EQ(QPTPool_start(pool), 0);
+    QPTPool_set_destructor(pool, (void (*)(void *)) free_work);
 
     // enqueue stuck thread
     pthread_mutex_lock(&args.mutex);
@@ -258,7 +260,7 @@ TEST(descend, swap) {
 
     EXPECT_EQ(descend(pool, 0, nullptr, &in, work, 0, dir_rc, 0,
                       [](QPTPool_t *, const size_t, void *data, void *) -> int {
-                          free(data);
+                          free_work(data);
                           return 0;
                       },
                       nullptr, nullptr,
@@ -273,7 +275,6 @@ TEST(descend, swap) {
     QPTPool_stop(pool);
     QPTPool_destroy(pool);
 
-    EXPECT_EQ(closedir(dir), 0);
     free_work(work);
     dir_dec(dir_rc);