track the number of open FDs in order to not exceed max

Since keeping directory FDs alive in order to open child directories
with relative paths incurs the risk of exhausting available file
descriptors, this tracks the maximum number of FDs that should be
allocated to long-living directory handles.

When that maximum is reached, directory handles will no longer have
their lifetime extended, until the limit goes back down again.

Author: bertschinger

include/bf.h

@@ -379,11 +379,11 @@ void free_work(void *p);
 struct dir_rc {
    DIR *dir;
    uint64_t rc;
+   int dont_clone;
 };
 
 struct dir_rc *open_dir_rc(struct work *w);
 int get_dir_fd(struct dir_rc *dir);
-void dir_inc(struct dir_rc *dir);
 struct dir_rc *dir_clone(struct dir_rc *dir);
 void dir_dec(struct dir_rc *dir);
 

src/QueuePerThreadPool.c

@@ -63,15 +63,51 @@ OF SUCH DAMAGE.
 
 
 #include <assert.h>
+#include <errno.h>
 #include <pthread.h>
 #include <stdio.h>
+#include <string.h>
 #include <stdlib.h>
+#include <sys/resource.h>
 
 #include "QueuePerThreadPool.h"
 #include "SinglyLinkedList.h"
 #include "swap.h"
 #include "utils.h"
 
+extern rlim_t MAX_OPEN_FILES;
+/*
+ * Determine how many file descriptors to allocate to potentially long-lived
+ * directory handles. Since the number of directory handles alive at once could
+ * be unbounded, it would risk file descriptor exhaustion if a limit is not
+ * imposed.
+ */
+void init_open_file_limit(size_t nthreads) {
+    struct rlimit rl;
+    int res = getrlimit(RLIMIT_NOFILE, &rl);
+    if (res) {
+        fprintf(stderr, "Warning: could not get open file limit: %s\n", strerror(errno));
+        MAX_OPEN_FILES = 0;
+        return;
+    }
+
+    /*
+     * Reserve 3 file descriptors per thread so that they always can always open
+     * the files they are currently working on. (This factor was determined
+     * experimentally.)
+     */
+    size_t reserve_fds = nthreads * 3;
+
+    if (rl.rlim_cur <= reserve_fds) {
+        // fprintf(stderr, "Warning: system may not allow enough open files for the number of requested threads.\n");
+        // fprintf(stderr, "Max number of open files: %llu; Number of threads requested: %llu\n", rl.rlim_cur, nthreads);
+        MAX_OPEN_FILES = 0;
+        return;
+    };
+
+    MAX_OPEN_FILES = rl.rlim_cur - reserve_fds;
+}
+
 typedef enum {
     INITIALIZED,
     RUNNING,
@@ -571,6 +607,8 @@ QPTPool_t *QPTPool_init_with_props(const size_t nthreads, void *args,
         return NULL;
     }
 
+    init_open_file_limit(nthreads);
+
     QPTPool_t *ctx = malloc(sizeof(QPTPool_t));
     ctx->nthreads = nthreads;
     ctx->queue_limit = queue_limit;

src/bf.c

@@ -62,13 +62,15 @@ OF SUCH DAMAGE.
 
 
 
+#include <errno.h>
 #include <fcntl.h>
 #include <pwd.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <unistd.h>
+#include <sys/resource.h>
 
 #include "bf.h"
 #include "dbutils.h"
@@ -590,6 +592,34 @@ INSTALL_NUMBER(INT, int, "%d")
 INSTALL_NUMBER(SIZE, size_t, "%zu")
 INSTALL_NUMBER(UINT64, uint64_t, "%" PRIu64)
 
+rlim_t MAX_OPEN_FILES;   /* Maximum number of files that may be held open by dir_rc objects */
+uint64_t CUR_OPEN_FILES; /* Current number of open files held by dir_rc objects */
+
+/*
+ * For understanding performance, track how many directories were opened
+ * with relative vs. absolute paths.
+ */
+static uint64_t n_openats = 0;
+static uint64_t n_opens = 0;
+
+#if defined(DEBUG) && 0
+__attribute__((destructor)) static void print_openat_stats(void) {
+    uint64_t tot = n_openats + n_opens;
+    if (tot > 0) {
+        printf("%" PRIu64 "/%" PRIu64 " (%" PRIu64 "%%) of directories used openat() over open()\n",
+                n_openats, tot, n_openats * 100 / tot);
+    }
+}
+#endif
+
+/*
+ * Increment the reference count for a dir_rc.
+ */
+static void dir_inc(struct dir_rc *dir) {
+    // XXX: is relaxed OK here?
+    __atomic_fetch_add(&dir->rc, 1, __ATOMIC_ACQ_REL);
+}
+
 /*
  * Create a new reference-counted DIR object for the given `struct work`.
  *
@@ -599,31 +629,49 @@ INSTALL_NUMBER(UINT64, uint64_t, "%" PRIu64)
  * Otherwise, just opendir() the path.
  *
  * Increments the refcount on the new object.
+ *
+ * If we are hitting the limit on the maximum number of open files available,
+ * then designate the new dir_rc as "non-clonable" so that it is closed as soon as
+ * processing the current directory is complete.
  */
 struct dir_rc *open_dir_rc(struct work *w) {
     DIR *dir;
 
     if (w->parent_dir) {
+        __atomic_add_fetch(&n_openats, 1, __ATOMIC_RELAXED);
         int d_fd = get_dir_fd(w->parent_dir);
         char *basename = w->name + w->name_len - w->basename_len;
         int fd = openat(d_fd, basename, O_RDONLY|O_DIRECTORY);
         if (fd < 0) {
-            return NULL;
+            goto err;
         }
         dir = fdopendir(fd);
     } else {
+        __atomic_add_fetch(&n_opens, 1, __ATOMIC_RELAXED);
         dir = opendir(w->name);
     }
 
     if (!dir) {
-        return NULL;
+        goto err;
     }
 
     struct dir_rc *new = calloc(1, sizeof(*new));
-    // printf("creating dir at %p\n", new);
     new->dir = dir;
     dir_inc(new);
+
+    uint64_t cur_open_files = __atomic_add_fetch(&CUR_OPEN_FILES, 1, __ATOMIC_ACQ_REL);
+    if (cur_open_files >= MAX_OPEN_FILES) {
+        new->dont_clone = 1;
+    }
+
     return new;
+
+  err:
+    if (errno == EMFILE) {
+        fprintf(stderr, "Warning: too many open files, index may not be complete!\n");
+    }
+
+    return NULL;
 }
 
 /*
@@ -634,19 +682,20 @@ int get_dir_fd(struct dir_rc *dir) {
 }
 
 /*
- * Increment the reference count for a dir_rc.
- */
-void dir_inc(struct dir_rc *dir) {
-    // printf("incrementing dir at %p\n", dir);
-    // XXX: is relaxed OK here?
-    __atomic_fetch_add(&dir->rc, 1, __ATOMIC_ACQ_REL);
-}
-
-/*
- * Clone a `dir_rc`.
- * This increments the refcount and returns a pointer to the dir_rc to the caller.
+ * Attempt to clone a `dir_rc`.
+ *
+ * If succesful, this increments the refcount and returns a pointer to the
+ * dir_rc to the caller.
+ *
+ * If unsuccesful, returns NULL.
+ *
+ * The caller MUST be prepared for cloning to fail!
  */
 struct dir_rc *dir_clone(struct dir_rc *dir) {
+    if (dir->dont_clone) {
+        return NULL;
+    }
+
     dir_inc(dir);
     return dir;
 }
@@ -656,10 +705,9 @@ struct dir_rc *dir_clone(struct dir_rc *dir) {
  */
 void dir_dec(struct dir_rc *dir) {
     if (dir) {
-        // printf("decrementing dir at %p\n", dir);
         if (__atomic_sub_fetch(&dir->rc, 1, __ATOMIC_ACQ_REL) == 0) {
-            // printf("freeing dir at %p\n", dir);
             closedir(dir->dir);
+            __atomic_sub_fetch(&CUR_OPEN_FILES, 1, __ATOMIC_ACQ_REL);
             free(dir);
         }
     }