#237 fix various bulk loader bugs related to nproc ulimit exceeded

ft/ftloader-internal.h

@@ -245,6 +245,7 @@ struct ft_loader_s {
     CACHETABLE cachetable;
     bool did_reserve_memory;
     bool compress_intermediates;
+    bool allow_puts;
     uint64_t   reserved_memory; // how much memory are we allowed to use?
 
     /* To make it easier to recover from errors, we don't use FILE*, instead we use an index into the file_infos. */
@@ -346,7 +347,8 @@ int toku_ft_loader_internal_init (/* out */ FTLOADER *blp,
                                    TOKUTXN txn,
                                    bool reserve_memory,
                                    uint64_t reserve_memory_size,
-                                   bool compress_intermediates);
+                                   bool compress_intermediates,
+                                   bool allow_puts);
 
 void toku_ft_loader_internal_destroy (FTLOADER bl, bool is_error);
 

ft/ftloader.cc

@@ -420,6 +420,10 @@ void toku_ft_loader_internal_destroy (FTLOADER bl, bool is_error) {
     }
 
     destroy_rowset(&bl->primary_rowset);
+    if (bl->primary_rowset_queue) {
+        queue_destroy(bl->primary_rowset_queue);
+        bl->primary_rowset_queue = nullptr;
+    }
 
     for (int i=0; i<bl->N; i++) {
         if ( bl->fractal_queues ) {
@@ -543,7 +547,8 @@ int toku_ft_loader_internal_init (/* out */ FTLOADER *blp,
                                    TOKUTXN txn,
                                    bool reserve_memory,
                                    uint64_t reserve_memory_size,
-                                   bool compress_intermediates)
+                                   bool compress_intermediates,
+                                   bool allow_puts)
 // Effect: Allocate and initialize a FTLOADER, but do not create the extractor thread.
 {
     FTLOADER CALLOC(bl); // initialized to all zeros (hence CALLOC)
@@ -560,10 +565,7 @@ int toku_ft_loader_internal_init (/* out */ FTLOADER *blp,
         bl->reserved_memory = 512*1024*1024; // if no cache table use 512MB.
     }
     bl->compress_intermediates = compress_intermediates;
-    if (0) { // debug
-        fprintf(stderr, "%s Reserved memory=%" PRId64 "\n", __FUNCTION__, bl->reserved_memory);
-    }
-
+    bl->allow_puts = allow_puts;
     bl->src_db = src_db;
     bl->N = N;
     bl->load_lsn = load_lsn;
@@ -628,7 +630,6 @@ int toku_ft_loader_internal_init (/* out */ FTLOADER *blp,
     {   int r = queue_create(&bl->primary_rowset_queue, EXTRACTOR_QUEUE_DEPTH); 
         if (r!=0) { toku_ft_loader_internal_destroy(bl, true); return r; }
     }
-    //printf("%s:%d toku_pthread_create\n", __FILE__, __LINE__);
     {
         ft_loader_lock_init(bl);
     }
@@ -650,34 +651,38 @@ int toku_ft_loader_open (/* out */ FTLOADER *blp,
                           TOKUTXN txn,
                           bool reserve_memory,
                           uint64_t reserve_memory_size,
-                          bool compress_intermediates)
-/* Effect: called by DB_ENV->create_loader to create an ft loader.
- * Arguments:
- *   blp                  Return the ft loader here.
- *   g                    The function for generating a row
- *   src_db               The source database.  Needed by g.  May be NULL if that's ok with g.
- *   N                    The number of dbs to create.
- *   dbs                  An array of open databases.  Used by g.  The data will be put in these database.
- *   new_fnames           The file names (these strings are owned by the caller: we make a copy for our own purposes).
- *   temp_file_template   A template suitable for mkstemp()
- * Return value: 0 on success, an error number otherwise.
- */
-{
+                          bool compress_intermediates,
+                          bool allow_puts) {
+// Effect: called by DB_ENV->create_loader to create a brt loader.
+// Arguments:
+//   blp                  Return the brt loader here.
+//   g                    The function for generating a row
+//   src_db               The source database.  Needed by g.  May be NULL if that's ok with g.
+//   N                    The number of dbs to create.
+//   dbs                  An array of open databases.  Used by g.  The data will be put in these database.
+//   new_fnames           The file names (these strings are owned by the caller: we make a copy for our own purposes).
+//   temp_file_template   A template suitable for mkstemp()
+//   reserve_memory       Cause the loader to reserve memory for its use from the cache table.
+//   compress_intermediates  Cause the loader to compress intermediate loader files.
+//   allow_puts           Prepare the loader for rows to insert.  When puts are disabled, the loader does not run the
+//                        extractor or the fractal tree writer threads.
+// Return value: 0 on success, an error number otherwise.
     int result = 0;
     {
         int r = toku_ft_loader_internal_init(blp, cachetable, g, src_db,
-                                             N, fts, dbs,
-                                             new_fnames_in_env,
-                                             bt_compare_functions,
-                                             temp_file_template,
-                                             load_lsn,
-                                             txn,
-                                             reserve_memory,
-                                             reserve_memory_size,
-                                             compress_intermediates);
+                                              N, fts, dbs,
+                                              new_fnames_in_env,
+                                              bt_compare_functions,
+                                              temp_file_template,
+                                              load_lsn,
+                                              txn,
+                                              reserve_memory,
+                                              reserve_memory_size,
+                                              compress_intermediates,
+                                              allow_puts);
         if (r!=0) result = r;
     }
-    if (result==0) {
+    if (result==0 && allow_puts) {
         FTLOADER bl = *blp;
         int r = toku_pthread_create(&bl->extractor_thread, NULL, extractor_thread, (void*)bl); 
         if (r==0) {
@@ -1213,6 +1218,7 @@ finish_extractor (FTLOADER bl) {
     {
         int r = queue_destroy(bl->primary_rowset_queue);
         invariant(r==0);
+        bl->primary_rowset_queue = nullptr;
     }
 
     rval = ft_loader_fi_close_all(&bl->file_infos);
@@ -1374,10 +1380,9 @@ int toku_ft_loader_put (FTLOADER bl, DBT *key, DBT *val)
  * Return value: 0 on success, an error number otherwise.
  */
 {
-    if (ft_loader_get_error(&bl->error_callback)) 
+    if (!bl->allow_puts || ft_loader_get_error(&bl->error_callback)) 
         return EINVAL; // previous panic
     bl->n_rows++;
-//    return loader_write_row(key, val, bl->fprimary_rows, &bl->fprimary_offset, bl);
     return loader_do_put(bl, key, val);
 }
 
@@ -2714,12 +2719,7 @@ static int loader_do_i (FTLOADER bl,
     struct rowset *rows = &(bl->rows[which_db]);
     invariant(rows->data==NULL); // the rows should be all cleaned up already
 
-    // a better allocation would be to figure out roughly how many merge passes we'll need.
-    int allocation_for_merge = (2*progress_allocation)/3;
-    progress_allocation -= allocation_for_merge;
-
-    int r;
-    r = queue_create(&bl->fractal_queues[which_db], FRACTAL_WRITER_QUEUE_DEPTH);
+    int r = queue_create(&bl->fractal_queues[which_db], FRACTAL_WRITER_QUEUE_DEPTH);
     if (r) goto error;
 
     {
@@ -2740,49 +2740,62 @@ static int loader_do_i (FTLOADER bl,
         r = dest_db->get_fanout(dest_db, &target_fanout);
         invariant_zero(r);
 
-        // This structure must stay live until the join below.
-        struct fractal_thread_args fta = { bl,
-                                           descriptor,
-                                           fd,
-                                           progress_allocation,
-                                           bl->fractal_queues[which_db],
-                                           bl->extracted_datasizes[which_db],
-                                           0,
-                                           which_db,
-                                           target_nodesize,
-                                           target_basementnodesize,
-                                           target_compression_method,
-                                           target_fanout
-        };
-
-        r = toku_pthread_create(bl->fractal_threads+which_db, NULL, fractal_thread, (void*)&fta);
-        if (r) {
-            int r2 __attribute__((__unused__)) = queue_destroy(bl->fractal_queues[which_db]);
-            // ignore r2, since we already have an error
-            goto error;
-        }
-        invariant(bl->fractal_threads_live[which_db]==false);
-        bl->fractal_threads_live[which_db] = true;
+        if (bl->allow_puts) {
+            // a better allocation would be to figure out roughly how many merge passes we'll need.
+            int allocation_for_merge = (2*progress_allocation)/3;
+            progress_allocation -= allocation_for_merge;
+            
+            // This structure must stay live until the join below.
+            struct fractal_thread_args fta = { 
+                bl,
+                descriptor,
+                fd,
+                progress_allocation,
+                bl->fractal_queues[which_db],
+                bl->extracted_datasizes[which_db],
+                0,
+                which_db,
+                target_nodesize,
+                target_basementnodesize,
+                target_compression_method,
+                target_fanout
+            };
+
+            r = toku_pthread_create(bl->fractal_threads+which_db, NULL, fractal_thread, (void*)&fta);
+            if (r) {
+                int r2 __attribute__((__unused__)) = queue_destroy(bl->fractal_queues[which_db]);            
+                // ignore r2, since we already have an error
+                bl->fractal_queues[which_db] = nullptr;
+                goto error;
+            }
+            invariant(bl->fractal_threads_live[which_db]==false);
+            bl->fractal_threads_live[which_db] = true;
 
-        r = merge_files(fs, bl, which_db, dest_db, compare, allocation_for_merge, bl->fractal_queues[which_db]);
+            r = merge_files(fs, bl, which_db, dest_db, compare, allocation_for_merge, bl->fractal_queues[which_db]);
 
-        {
-            void *toku_pthread_retval;
-            int r2 = toku_pthread_join(bl->fractal_threads[which_db], &toku_pthread_retval);
-            invariant(fta.bl==bl); // this is a gratuitous assertion to make sure that the fta struct is still live here.  A previous bug but that struct into a C block statement.
-            resource_assert_zero(r2);
-            invariant(toku_pthread_retval==NULL);
-            invariant(bl->fractal_threads_live[which_db]);
-            bl->fractal_threads_live[which_db] = false;
-            if (r == 0) r = fta.errno_result;
+            {
+                void *toku_pthread_retval;
+                int r2 = toku_pthread_join(bl->fractal_threads[which_db], &toku_pthread_retval);
+                invariant(fta.bl==bl); // this is a gratuitous assertion to make sure that the fta struct is still live here.  A previous bug put that struct into a C block statement.
+                resource_assert_zero(r2);
+                invariant(toku_pthread_retval==NULL);
+                invariant(bl->fractal_threads_live[which_db]);
+                bl->fractal_threads_live[which_db] = false;
+                if (r == 0) r = fta.errno_result;
+            }
+        } else {
+            queue_eof(bl->fractal_queues[which_db]);
+            r = toku_loader_write_ft_from_q(bl, descriptor, fd, progress_allocation, 
+                                            bl->fractal_queues[which_db], bl->extracted_datasizes[which_db], which_db, 
+                                            target_nodesize, target_basementnodesize, target_compression_method, target_fanout);
         }
     }
 
  error: // this is the cleanup code.  Even if r==0 (no error) we fall through to here.
-    {
+    if (bl->fractal_queues[which_db]) {
         int r2 = queue_destroy(bl->fractal_queues[which_db]);
         invariant(r2==0);
-        bl->fractal_queues[which_db]=NULL;
+        bl->fractal_queues[which_db] = nullptr;
     }
 
     // if we get here we need to free up the merge_fileset and the rowset, as well as the keys
@@ -2851,6 +2864,10 @@ int toku_ft_loader_close (FTLOADER bl,
         if (r)
             result = r;
         invariant(!bl->extractor_live);
+    } else {
+        r = finish_primary_rows(bl);
+        if (r)
+            result = r;
     }
 
     // check for an error during extraction

ft/ftloader.h

@@ -113,7 +113,8 @@ int toku_ft_loader_open (FTLOADER *bl,
                           TOKUTXN txn,
                           bool reserve_memory,
                           uint64_t reserve_memory_size,
-                          bool compress_intermediates);
+                          bool compress_intermediates,
+                          bool allow_puts);
 
 int toku_ft_loader_put (FTLOADER bl, DBT *key, DBT *val);
 

ft/tests/ftloader-test-bad-generate.cc

@@ -170,7 +170,7 @@ static void test_extractor(int nrows, int nrowsets, bool expect_fail) {
     }
 
     FTLOADER loader;
-    r = toku_ft_loader_open(&loader, NULL, generate, NULL, N, fts, dbs, fnames, compares, "tempXXXXXX", ZERO_LSN, nullptr, true, 0, false);
+    r = toku_ft_loader_open(&loader, NULL, generate, NULL, N, fts, dbs, fnames, compares, "tempXXXXXX", ZERO_LSN, nullptr, true, 0, false, true);
     assert(r == 0);
 
     struct rowset *rowset[nrowsets];

ft/tests/ftloader-test-extractor-errors.cc

@@ -180,7 +180,7 @@ static void test_extractor(int nrows, int nrowsets, bool expect_fail, const char
     sprintf(temp, "%s/%s", testdir, "tempXXXXXX");
 
     FTLOADER loader;
-    r = toku_ft_loader_open(&loader, NULL, generate, NULL, N, fts, dbs, fnames, compares, "tempXXXXXX", ZERO_LSN, nullptr, true, 0, false);
+    r = toku_ft_loader_open(&loader, NULL, generate, NULL, N, fts, dbs, fnames, compares, "tempXXXXXX", ZERO_LSN, nullptr, true, 0, false, true);
     assert(r == 0);
 
     struct rowset *rowset[nrowsets];

ft/tests/ftloader-test-extractor.cc

@@ -402,7 +402,7 @@ static void test_extractor(int nrows, int nrowsets, const char *testdir) {
     sprintf(temp, "%s/%s", testdir, "tempXXXXXX");
 
     FTLOADER loader;
-    r = toku_ft_loader_open(&loader, NULL, generate, NULL, N, fts, dbs, fnames, compares, temp, ZERO_LSN, nullptr, true, 0, false);
+    r = toku_ft_loader_open(&loader, NULL, generate, NULL, N, fts, dbs, fnames, compares, temp, ZERO_LSN, nullptr, true, 0, false, true);
     assert(r == 0);
 
     struct rowset *rowset[nrowsets];

ft/tests/ftloader-test-merge-files-dbufio.cc

@@ -412,7 +412,7 @@ static void test (const char *directory, bool is_error) {
 					       bt_compare_functions,
 					       "tempxxxxxx",
 					       *lsnp,
-                                               nullptr, true, 0, false);
+                                               nullptr, true, 0, false, true);
 	assert(r==0);
     }
 
@@ -500,11 +500,6 @@ static void test (const char *directory, bool is_error) {
 	    assert(cthunk.n_read == N_RECORDS);
 	}
     }
-    //printf("%s:%d Destroying\n", __FILE__, __LINE__);
-    {
-	int r = queue_destroy(bl->primary_rowset_queue);
-	assert(r==0);
-    }
     {
 	int r = queue_destroy(q);
 	assert(r==0);

ft/tests/ftloader-test-open.cc

@@ -143,7 +143,7 @@ static void test_loader_open(int ndbs) {
     for (i = 0; ; i++) {
         set_my_malloc_trigger(i+1);
 
-        r = toku_ft_loader_open(&loader, NULL, NULL, NULL, ndbs, fts, dbs, fnames, compares, "", ZERO_LSN, nullptr, true, 0, false);
+        r = toku_ft_loader_open(&loader, NULL, NULL, NULL, ndbs, fts, dbs, fnames, compares, "", ZERO_LSN, nullptr, true, 0, false, true);
         if (r == 0)
             break;
     }

src/loader.cc

@@ -172,6 +172,13 @@ struct __toku_loader_internal {
     char **inames_in_env; /* [N]  inames of new files to be created */
 };
 
+static void free_inames(char **inames, int n) {
+    for (int i = 0; i < n; i++) {
+        toku_free(inames[i]);
+    }
+    toku_free(inames);
+}
+
 /*
  *  free_loader_resources() frees all of the resources associated with
  *      struct __toku_loader_internal 
@@ -185,16 +192,15 @@ static void free_loader_resources(DB_LOADER *loader)
         toku_destroy_dbt(&loader->i->err_val);
 
         if (loader->i->inames_in_env) {
-            for (int i=0; i<loader->i->N; i++) {
-                if (loader->i->inames_in_env[i]) toku_free(loader->i->inames_in_env[i]);
-            }
-            toku_free(loader->i->inames_in_env);
+            free_inames(loader->i->inames_in_env, loader->i->N);
+            loader->i->inames_in_env = nullptr;
         }
-        if (loader->i->temp_file_template) toku_free(loader->i->temp_file_template);
+        toku_free(loader->i->temp_file_template);
+        loader->i->temp_file_template = nullptr;
 
         // loader->i
         toku_free(loader->i);
-        loader->i = NULL;
+        loader->i = nullptr;
     }
 }
 
@@ -306,6 +312,9 @@ toku_loader_create_loader(DB_ENV *env,
 
         // time to open the big kahuna
         char **XMALLOC_N(N, new_inames_in_env);
+        for (int i = 0; i < N; i++) {
+            new_inames_in_env[i] = nullptr;
+        }
         FT_HANDLE *XMALLOC_N(N, fts);
         for (int i=0; i<N; i++) {
             fts[i] = dbs[i]->i->ft_handle;
@@ -313,7 +322,7 @@ toku_loader_create_loader(DB_ENV *env,
         LSN load_lsn;
         rval = locked_load_inames(env, txn, N, dbs, new_inames_in_env, &load_lsn, puts_allowed);
         if ( rval!=0 ) {
-            toku_free(new_inames_in_env);
+            free_inames(new_inames_in_env, N);
             toku_free(fts);
             goto create_exit;
         }
@@ -331,12 +340,14 @@ toku_loader_create_loader(DB_ENV *env,
                                  ttxn,
                                  puts_allowed,
                                  env->get_loader_memory_size(env),
-                                 compress_intermediates);
+                                 compress_intermediates,
+                                 puts_allowed);
         if ( rval!=0 ) {
-            toku_free(new_inames_in_env);
+            free_inames(new_inames_in_env, N);
             toku_free(fts);
             goto create_exit;
         }
+
         loader->i->inames_in_env = new_inames_in_env;
         toku_free(fts);
 
@@ -441,7 +452,7 @@ static void redirect_loader_to_empty_dictionaries(DB_LOADER *loader) {
         loader->i->dbs,
         loader->i->db_flags,
         loader->i->dbt_flags,
-        0,
+        LOADER_DISALLOW_PUTS,
         false
         );
     lazy_assert_zero(r);

src/tests/loader-nproc-close.cc

+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+/*
+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) 2010-2013 Tokutek Inc.  All rights reserved."
+#ident "$Id$"
+
+#include "test.h"
+#include <db.h>
+#include <sys/resource.h>
+
+static int loader_flags = 0;
+static const char *envdir = TOKU_TEST_FILENAME;
+
+static int put_multiple_generate(DB *UU(dest_db), DB *UU(src_db), DBT_ARRAY *UU(dest_keys), DBT_ARRAY *UU(dest_vals), const DBT *UU(src_key), const DBT *UU(src_val)) {
+    return ENOMEM;
+}
+
+static void loader_open_close(int ndb) {
+    int r;
+
+    char rmcmd[32 + strlen(envdir)];
+    snprintf(rmcmd, sizeof rmcmd, "rm -rf %s", envdir);
+    r = system(rmcmd);                                                                             CKERR(r);
+    r = toku_os_mkdir(envdir, S_IRWXU+S_IRWXG+S_IRWXO);                                                       CKERR(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                                               CKERR(r);
+    r = env->set_generate_row_callback_for_put(env, put_multiple_generate);
+    CKERR(r);
+    int envflags = DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | DB_INIT_TXN | DB_CREATE | DB_PRIVATE;
+    r = env->open(env, envdir, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                                            CKERR(r);
+    env->set_errfile(env, stderr);
+
+    DB *dbs[ndb];
+    uint32_t db_flags[ndb];
+    uint32_t dbt_flags[ndb];
+    for (int i = 0; i < ndb; i++) {
+        db_flags[i] = DB_NOOVERWRITE;
+        dbt_flags[i] = 0;
+        r = db_create(&dbs[i], env, 0); CKERR(r);
+        char name[32];
+        sprintf(name, "db%d", i);
+        r = dbs[i]->open(dbs[i], NULL, name, NULL, DB_BTREE, DB_CREATE, 0666); CKERR(r);
+    }
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0); CKERR(r);
+
+    DB_LOADER *loader;
+    r = env->create_loader(env, txn, &loader, ndb > 0 ? dbs[0] : NULL, ndb, dbs, db_flags, dbt_flags, loader_flags); CKERR(r);
+
+    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 = 0;
+    r = setrlimit(RLIMIT_NPROC, &new_nproc_limit);
+    assert(r == 0);
+
+    r = loader->close(loader); 
+
+    if (loader_flags & LOADER_DISALLOW_PUTS)
+        CKERR(r);
+    else
+        CKERR2(r, EAGAIN);
+
+    r = setrlimit(RLIMIT_NPROC, &current_nproc_limit);
+    assert(r == 0);
+
+    r = txn->abort(txn); CKERR(r);
+
+    for (int i = 0; i < ndb; i++) {
+        r = dbs[i]->close(dbs[i], 0); CKERR(r);
+    }
+
+    r = env->close(env, 0); CKERR(r);
+}
+
+static void do_args(int argc, char * const argv[]) {
+    int resultcode;
+    char *cmd = argv[0];
+    argc--; argv++;
+    while (argc>0) {
+        if (strcmp(argv[0], "-h")==0) {
+	    resultcode=0;
+	do_usage:
+	    fprintf(stderr, "Usage: %s -h -v -q -p\n", cmd);
+	    exit(resultcode);
+	} else if (strcmp(argv[0], "-v")==0) {
+	    verbose++;
+	} else if (strcmp(argv[0],"-q")==0) {
+	    verbose--;
+	    if (verbose<0) verbose=0;
+        } else if (strcmp(argv[0], "-p") == 0) {
+            loader_flags |= LOADER_DISALLOW_PUTS;
+        } else if (strcmp(argv[0], "-z") == 0) {
+            loader_flags |= LOADER_COMPRESS_INTERMEDIATES;
+        } else if (strcmp(argv[0], "-e") == 0) {
+            argc--; argv++;
+            if (argc > 0)
+                envdir = argv[0];
+	} else {
+	    fprintf(stderr, "Unknown arg: %s\n", argv[0]);
+	    resultcode=1;
+	    goto do_usage;
+	}
+	argc--;
+	argv++;
+    }
+}
+
+int test_main(int argc, char * const *argv) {
+    do_args(argc, argv);
+    loader_open_close(1);
+    return 0;
+}

src/tests/loader-nproc-create.cc

+/* -*- mode: C++; c-basic-offset: 4; indent-tabs-mode: nil -*- */
+// vim: ft=cpp:expandtab:ts=8:sw=4:softtabstop=4:
+/*
+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) 2010-2013 Tokutek Inc.  All rights reserved."
+#ident "$Id$"
+
+#include "test.h"
+#include <db.h>
+#include <sys/resource.h>
+
+static int loader_flags = 0;
+static const char *envdir = TOKU_TEST_FILENAME;
+
+static int put_multiple_generate(DB *UU(dest_db), DB *UU(src_db), DBT_ARRAY *UU(dest_keys), DBT_ARRAY *UU(dest_vals), const DBT *UU(src_key), const DBT *UU(src_val)) {
+    return ENOMEM;
+}
+
+static void loader_open_close(int ndb) {
+    int r;
+
+    char rmcmd[32 + strlen(envdir)];
+    snprintf(rmcmd, sizeof rmcmd, "rm -rf %s", envdir);
+    r = system(rmcmd);                                                                             CKERR(r);
+    r = toku_os_mkdir(envdir, S_IRWXU+S_IRWXG+S_IRWXO);                                                       CKERR(r);
+
+    DB_ENV *env;
+    r = db_env_create(&env, 0);                                                                               CKERR(r);
+    r = env->set_generate_row_callback_for_put(env, put_multiple_generate);
+    CKERR(r);
+    int envflags = DB_INIT_LOCK | DB_INIT_LOG | DB_INIT_MPOOL | DB_INIT_TXN | DB_CREATE | DB_PRIVATE;
+    r = env->open(env, envdir, envflags, S_IRWXU+S_IRWXG+S_IRWXO);                                            CKERR(r);
+    env->set_errfile(env, stderr);
+
+    DB *dbs[ndb];
+    uint32_t db_flags[ndb];
+    uint32_t dbt_flags[ndb];
+    for (int i = 0; i < ndb; i++) {
+        db_flags[i] = DB_NOOVERWRITE;
+        dbt_flags[i] = 0;
+        r = db_create(&dbs[i], env, 0); CKERR(r);
+        char name[32];
+        sprintf(name, "db%d", i);
+        r = dbs[i]->open(dbs[i], NULL, name, NULL, DB_BTREE, DB_CREATE, 0666); CKERR(r);
+    }
+
+    DB_TXN *txn;
+    r = env->txn_begin(env, NULL, &txn, 0); CKERR(r);
+
+    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 = 0;
+    r = setrlimit(RLIMIT_NPROC, &new_nproc_limit);
+    assert(r == 0);
+
+    DB_LOADER *loader;
+    int loader_r = env->create_loader(env, txn, &loader, ndb > 0 ? dbs[0] : NULL, ndb, dbs, db_flags, dbt_flags, loader_flags);
+
+    r = setrlimit(RLIMIT_NPROC, &current_nproc_limit);
+    assert(r == 0);
+
+    if (loader_flags & LOADER_DISALLOW_PUTS)  {
+        CKERR(loader_r);
+        loader_r = loader->close(loader);
+        CKERR(loader_r);
+    } else {
+        CKERR2(loader_r, EAGAIN);
+    }
+
+    r = txn->abort(txn); CKERR(r);
+
+    for (int i = 0; i < ndb; i++) {
+        r = dbs[i]->close(dbs[i], 0); CKERR(r);
+    }
+
+    r = env->close(env, 0); CKERR(r);
+}
+
+static void do_args(int argc, char * const argv[]) {
+    int resultcode;
+    char *cmd = argv[0];
+    argc--; argv++;
+    while (argc>0) {
+        if (strcmp(argv[0], "-h")==0) {
+	    resultcode=0;
+	do_usage:
+	    fprintf(stderr, "Usage: %s -h -v -q -p\n", cmd);
+	    exit(resultcode);
+	} else if (strcmp(argv[0], "-v")==0) {
+	    verbose++;
+	} else if (strcmp(argv[0],"-q")==0) {
+	    verbose--;
+	    if (verbose<0) verbose=0;
+        } else if (strcmp(argv[0], "-p") == 0) {
+            loader_flags |= LOADER_DISALLOW_PUTS;
+        } else if (strcmp(argv[0], "-z") == 0) {
+            loader_flags |= LOADER_COMPRESS_INTERMEDIATES;
+        } else if (strcmp(argv[0], "-e") == 0) {
+            argc--; argv++;
+            if (argc > 0)
+                envdir = argv[0];
+	} else {
+	    fprintf(stderr, "Unknown arg: %s\n", argv[0]);
+	    resultcode=1;
+	    goto do_usage;
+	}
+	argc--;
+	argv++;
+    }
+}
+
+int test_main(int argc, char * const *argv) {
+    do_args(argc, argv);
+    loader_open_close(1);
+    return 0;
+}