#240 make the toku thread pool handle transient thread creation errors

util/tests/threadpool-nproc-limit.cc

+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+#ident "$Id$"
+/*
+COPYING CONDITIONS NOTICE:
+
+  This program is free software; you can redistribute it and/or modify
+  it under the terms of version 2 of the GNU General Public License as
+  published by the Free Software Foundation, and provided that the
+  following conditions are met:
+
+      * Redistributions of source code must retain this COPYING
+        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
+        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
+        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
+        GRANT (below).
+
+      * Redistributions in binary form must reproduce this COPYING
+        CONDITIONS NOTICE, the COPYRIGHT NOTICE (below), the
+        DISCLAIMER (below), the UNIVERSITY PATENT NOTICE (below), the
+        PATENT MARKING NOTICE (below), and the PATENT RIGHTS
+        GRANT (below) in the documentation and/or other materials
+        provided with the distribution.
+
+  You should have received a copy of the GNU General Public License
+  along with this program; if not, write to the Free Software
+  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
+  02110-1301, USA.
+
+COPYRIGHT NOTICE:
+
+  TokuDB, Tokutek Fractal Tree Indexing Library.
+  Copyright (C) 2007-2013 Tokutek, Inc.
+
+DISCLAIMER:
+
+  This program is distributed in the hope that it will be useful, but
+  WITHOUT ANY WARRANTY; without even the implied warranty of
+  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+  General Public License for more details.
+
+UNIVERSITY PATENT NOTICE:
+
+  The technology is licensed by the Massachusetts Institute of
+  Technology, Rutgers State University of New Jersey, and the Research
+  Foundation of State University of New York at Stony Brook under
+  United States of America Serial No. 11/760379 and to the patents
+  and/or patent applications resulting from it.
+
+PATENT MARKING NOTICE:
+
+  This software is covered by US Patent No. 8,185,551.
+  This software is covered by US Patent No. 8,489,638.
+
+PATENT RIGHTS GRANT:
+
+  "THIS IMPLEMENTATION" means the copyrightable works distributed by
+  Tokutek as part of the Fractal Tree project.
+
+  "PATENT CLAIMS" means the claims of patents that are owned or
+  licensable by Tokutek, both currently or in the future; and that in
+  the absence of this license would be infringed by THIS
+  IMPLEMENTATION or by using or running THIS IMPLEMENTATION.
+
+  "PATENT CHALLENGE" shall mean a challenge to the validity,
+  patentability, enforceability and/or non-infringement of any of the
+  PATENT CLAIMS or otherwise opposing any of the PATENT CLAIMS.
+
+  Tokutek hereby grants to you, for the term and geographical scope of
+  the PATENT CLAIMS, a non-exclusive, no-charge, royalty-free,
+  irrevocable (except as stated in this section) patent license to
+  make, have made, use, offer to sell, sell, import, transfer, and
+  otherwise run, modify, and propagate the contents of THIS
+  IMPLEMENTATION, where such license applies only to the PATENT
+  CLAIMS.  This grant does not include claims that would be infringed
+  only as a consequence of further modifications of THIS
+  IMPLEMENTATION.  If you or your agent or licensee institute or order
+  or agree to the institution of patent litigation against any entity
+  (including a cross-claim or counterclaim in a lawsuit) alleging that
+  THIS IMPLEMENTATION constitutes direct or contributory patent
+  infringement, or inducement of patent infringement, then any rights
+  granted to you under this License shall terminate as of the date
+  such litigation is filed.  If you or your agent or exclusive
+  licensee institute or order or agree to the institution of a PATENT
+  CHALLENGE, then Tokutek may terminate any rights granted to you
+  under this License.
+*/
+
+#ident "Copyright (c) 2014 Tokutek Inc.  All rights reserved."
+
+// this test verifies that the toku thread pool is resilient when hitting the nproc limit.
+
+#include <util/threadpool.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+#include <string.h>
+#include <unistd.h>
+#include <errno.h>
+#include <sys/resource.h>
+
+int verbose = 0;
+
+static int usage(void) {
+    fprintf(stderr, "[-q] [-v] [--verbose] (%d)\n", verbose);
+    return 1;
+}
+
+static void *f(void *arg) {
+    return arg;
+}
+
+static int dotest(int the_limit) {
+    if (verbose)
+        fprintf(stderr, "%s:%u %d\n", __FILE__, __LINE__, the_limit);
+    int r;
+    struct toku_thread_pool *pool = nullptr;
+    r = toku_thread_pool_create(&pool, 10);
+    assert(r == 0 && pool != nullptr);
+
+    struct rlimit current_nproc_limit;
+    r = getrlimit(RLIMIT_NPROC, &current_nproc_limit);
+    assert(r == 0);
+    
+    struct rlimit new_nproc_limit = current_nproc_limit;
+    new_nproc_limit.rlim_cur = the_limit;
+    r = setrlimit(RLIMIT_NPROC, &new_nproc_limit);
+    assert(r == 0);
+
+    int want_n = 20;
+    int got_n = want_n;
+    r = toku_thread_pool_run(pool, 0, &got_n, f, nullptr);
+    if (r == 0)
+        assert(want_n == got_n);
+    else {
+        assert(r == EWOULDBLOCK);
+        assert(got_n <= want_n);
+    }
+
+    r = setrlimit(RLIMIT_NPROC, &current_nproc_limit);
+    assert(r == 0);
+
+    if (verbose)
+        toku_thread_pool_print(pool, stderr);
+    toku_thread_pool_destroy(&pool);
+    return got_n > 0;
+}
+
+int main(int argc, char *argv[]) {
+    // parse args
+    for (int i = 1; i < argc; i++) {
+        char *arg = argv[i];
+        if (arg[0] != '-')
+            break;
+        if (strcmp(arg, "-v") == 0 || strcmp(arg, "--verbose") == 0) {
+            verbose = verbose+1;
+            continue;
+        }
+        if (strcmp(arg, "-q") == 0) {
+            verbose = verbose > 0 ? verbose-1 : 0;
+            continue;
+        }
+        return usage();
+    }
+    // set increasing nproc limits until the test succeeds in hitting the limit after > 0 threads are created
+    for (int i = 0; 1; i++) {
+        if (dotest(i))
+            break;
+    }
+    return 0;
+}

util/threadpool.cc

@@ -132,13 +132,18 @@ static int
 toku_thread_create(struct toku_thread_pool *pool, struct toku_thread **toku_thread_return) {
     int r;
     struct toku_thread *MALLOC(thread);
-    if (thread == NULL) {
+    if (thread == nullptr) {
         r = get_error_errno();
     } else {
         memset(thread, 0, sizeof *thread);
         thread->pool = pool;
-        toku_cond_init(&thread->wait, NULL);
-        r = toku_pthread_create(&thread->tid, NULL, toku_thread_run_internal, thread); resource_assert_zero(r);
+        toku_cond_init(&thread->wait, nullptr);
+        r = toku_pthread_create(&thread->tid, nullptr, toku_thread_run_internal, thread);
+        if (r) {
+            toku_cond_destroy(&thread->wait);
+            toku_free(thread);
+            thread = nullptr;
+        }
         *toku_thread_return = thread;
     }
     return r;
@@ -192,7 +197,7 @@ toku_thread_run_internal(void *arg) {
         if (doexit)
             break;
         toku_thread_pool_lock(pool);
-        thread->f = NULL;
+        thread->f = nullptr;
         toku_list_push(&pool->free_threads, &thread->free_link);
     }
     return arg;
@@ -202,13 +207,13 @@ int
 toku_thread_pool_create(struct toku_thread_pool **pool_return, int max_threads) {
     int r;
     struct toku_thread_pool *CALLOC(pool);
-    if (pool == NULL) {
+    if (pool == nullptr) {
         r = get_error_errno();
     } else {
-        toku_mutex_init(&pool->lock, NULL);
+        toku_mutex_init(&pool->lock, nullptr);
         toku_list_init(&pool->free_threads);
         toku_list_init(&pool->all_threads);
-        toku_cond_init(&pool->wait_free, NULL);
+        toku_cond_init(&pool->wait_free, nullptr);
         pool->cur_threads = 0;
         pool->max_threads = max_threads;
         *pool_return = pool;
@@ -230,7 +235,7 @@ toku_thread_pool_unlock(struct toku_thread_pool *pool) {
 void 
 toku_thread_pool_destroy(struct toku_thread_pool **poolptr) {
     struct toku_thread_pool *pool = *poolptr;
-    *poolptr = NULL;
+    *poolptr = nullptr;
 
     // ask the threads to exit
     toku_thread_pool_lock(pool);
@@ -260,7 +265,7 @@ toku_thread_pool_destroy(struct toku_thread_pool **poolptr) {
 
 static int 
 toku_thread_pool_add(struct toku_thread_pool *pool) {
-    struct toku_thread *thread = NULL;
+    struct toku_thread *thread = nullptr;
     int r = toku_thread_create(pool, &thread); 
     if (r == 0) {
         pool->cur_threads += 1;
@@ -294,7 +299,7 @@ toku_thread_pool_get_one(struct toku_thread_pool *pool, int dowait, struct toku_
         struct toku_thread *thread = toku_list_struct(list, struct toku_thread, free_link);
         *toku_thread_return = thread;
     } else
-        *toku_thread_return = NULL;
+        *toku_thread_return = nullptr;
     toku_thread_pool_unlock(pool);
     return r;
 }