Add fuse-read-spin-us: poll /dev/fuse briefly before blocking

After handling a request, a sync read thread polls its /dev/fuse fd for
up to fuse-read-spin-us microseconds before blocking in read(). Default
0 (disabled). Measured at 50us (tmpfs branch, direct_io): sequential
128K/1M reads +27-32%, 128K writes +14%, metadata walks +24-32%; no
change on 8 parallel streams or 1M writes. Costs roughly one core on a
busy-but-quiet mount, nothing when idle.

mkdocs/docs/config/options.md

@@ -155,6 +155,9 @@ config file.
   multiplied by the number of process threads plus read thread
   count. 0 sets the depth to the same as the process thread
   count. (default: 2)
+* **[fuse-read-spin-us](threads.md)=UINT**: Microseconds a read thread
+  polls for the next request before sleeping. Trades CPU for lower
+  latency on synchronous workloads. (default: 0)
 * **[pin-threads](pin-threads.md)=STR**: Selects a strategy to pin
   threads to CPUs (default: unset)
 * **[flush-on-close](flush-on-close.md)=never|always|opened-for-write**:

mkdocs/docs/config/threads.md

@@ -59,3 +59,26 @@ in the pool.
   block in order to limit memory growth.
 * `process-thread-queue-depth<=0`: Sets the queue depth to 2. May be
   used in the future to set dynamically.
+
+
+## fuse-read-spin-us
+
+Defaults to `0` (disabled)
+
+Applies only when `process-thread-count=-1` (read threads also process
+requests). After handling a request, each read thread polls for the
+next request for up to this many microseconds before going to sleep.
+
+This lowers per-request latency for synchronous, single-stream workloads
+(sequential reads, small writes, metadata-heavy walks such as `ls -l`,
+`find`, or `rsync` scans) at the cost of CPU: while requests keep
+arriving, a read thread stays busy instead of sleeping. On a quiet,
+low-rate request stream this costs roughly one CPU core's worth of
+polling with no benefit. An idle mount costs nothing.
+
+`50` is the only value tested. Larger values increase CPU use. Parallel
+multi-stream workloads and large (1MiB) writes showed no change.
+
+* `fuse-read-spin-us=0`: Disabled. Read threads block immediately.
+* `fuse-read-spin-us=N`: Poll for up to `N` microseconds after each
+  request.

src/config.cpp

@@ -116,6 +116,7 @@ Config::Config()
   posix_acl(false),
   process_thread_count(fuse_cfg.process_thread_count),
   process_thread_queue_depth(fuse_cfg.process_thread_queue_depth),
+  fuse_read_spin_us(fuse_cfg.read_spin_us),
   proxy_ioprio(false),
   read_thread_count(fuse_cfg.read_thread_count),
   readahead(0),
@@ -172,6 +173,7 @@ Config::Config()
     posix_acl.ro =
     process_thread_count.ro =
     process_thread_queue_depth.ro =
+    fuse_read_spin_us.ro =
     read_thread_count.ro =
     scheduling_priority.ro =
     true;
@@ -276,6 +278,7 @@ Config::Config()
   _map["posix-acl"]                   = &posix_acl;
   _map["process-thread-count"]        = &process_thread_count;
   _map["process-thread-queue-depth"]  = &process_thread_queue_depth;
+  _map["fuse-read-spin-us"]           = &fuse_read_spin_us;
   _map["proxy-ioprio"]                = &proxy_ioprio;
   _map["read-thread-count"]           = &read_thread_count;
   _map["readahead"]                   = &readahead;

src/config.hpp

@@ -153,6 +153,7 @@ public:
   ConfigBOOL     posix_acl;
   TFSRef<int>    process_thread_count;
   TFSRef<int>    process_thread_queue_depth;
+  TFSRef<u64>    fuse_read_spin_us;
   ProxyIOPrio    proxy_ioprio;
   TFSRef<int>    read_thread_count;
   ConfigU64      readahead;

vendored/libfuse/include/fuse_cfg.hpp

@@ -47,6 +47,11 @@ public:
   int read_thread_count = 0;
   int process_thread_count = -1;
   int process_thread_queue_depth = 2;
+  // After each request a sync FUSE reader polls /dev/fuse for up to this many
+  // microseconds before blocking in read(). Trades CPU while the mount is
+  // busy for lower per-request latency on synchronous request streams.
+  // 0 disables.
+  u64 read_spin_us = 0;
   std::string pin_threads = "false";
 
   u16 request_timeout = 0;

vendored/libfuse/include/fuse_lowlevel.h

@@ -572,7 +572,8 @@ int fuse_session_loop_mt(struct fuse_session *se,
                          const int            read_thread_count,
                          const int            process_thread_count,
                          const int            process_thread_queue_depth,
-                         const char          *pin_threads_type);
+                         const char          *pin_threads_type,
+                         const uint64_t       read_spin_us);
 
 /* Direct file descriptor and buffer size accessors
  * (channel fields inlined into session for single-mount simplicity) */

vendored/libfuse/lib/fuse_loop.cpp

@@ -21,6 +21,8 @@
 #include <vector>
 
 #include <errno.h>
+#include <poll.h>
+#include <time.h>
 #include <pthread.h>
 #include <semaphore.h>
 #include <signal.h>
@@ -134,14 +136,17 @@ struct SyncWorker
   fuse_session *_se;
   int _fd;
   sem_t *_finished;
+  u64 _spin_us;
 
   SyncWorker(fuse_session *se_,
              int           fd_,
-             sem_t        *finished_)
+             sem_t        *finished_,
+             u64           spin_us_)
 
     : _se(se_),
       _fd(fd_),
-      _finished(finished_)
+      _finished(finished_),
+      _spin_us(spin_us_)
   {
   }
 
@@ -152,6 +157,7 @@ struct SyncWorker
     DEFER{ fuse_session_exit(_se); };
     DEFER{ sem_post(_finished); };
 
+    bool processed_one = false;
     while(!fuse_session_exited(_se))
       {
         int rv;
@@ -164,6 +170,35 @@ struct SyncWorker
             return;
           }
 
+        /* After processing a request, poll /dev/fuse for up to _spin_us
+           before blocking in receive. On synchronous request streams (one
+           dd, one metadata walk) this measurably lowers per-request latency
+           versus blocking immediately; the exact scheduler mechanism is not
+           established. Cost: up to _spin_us of CPU per worker after each
+           request, so a busy-but-quiet mount pays roughly a core; nothing on
+           an idle mount. 0 disables. Cancellation stays disabled during the
+           spin (it is only enabled around the blocking receive) and the spin
+           is bounded, so shutdown is unaffected. */
+        if(processed_one && (_spin_us > 0))
+          {
+            struct pollfd pfd = {_fd,POLLIN,0};
+            struct timespec ts;
+            uint64_t end;
+
+            ::clock_gettime(CLOCK_MONOTONIC,&ts);
+            end = ((uint64_t)ts.tv_sec * 1000000000ULL +
+                   (uint64_t)ts.tv_nsec +
+                   (_spin_us * 1000ULL));
+            while(::poll(&pfd,1,0) == 0)
+              {
+                ::clock_gettime(CLOCK_MONOTONIC,&ts);
+                if(((uint64_t)ts.tv_sec * 1000000000ULL +
+                    (uint64_t)ts.tv_nsec) >= end)
+                  break;
+              }
+          }
+        processed_one = true;
+
         while(true)
           {
             pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);
@@ -257,7 +292,8 @@ fuse_session_loop_mt(struct fuse_session *se_,
                      const int            raw_read_thread_count_,
                      const int            raw_process_thread_count_,
                      const int            raw_process_thread_queue_depth_,
-                     const std::string    pin_threads_type_)
+                     const std::string    pin_threads_type_,
+                     const u64            read_spin_us_)
 {
   sem_t finished;
   bool cloned_read_fds;
@@ -303,7 +339,8 @@ fuse_session_loop_mt(struct fuse_session *se_,
       for(auto i = 0; i < read_thread_count; i++)
         read_tp->enqueue_work(SyncWorker(se_,
                                          fuse_session_read_fd(se_,i),
-                                         &finished));
+                                         &finished,
+                                         read_spin_us_));
     }
 
   if(read_tp)
@@ -317,12 +354,14 @@ fuse_session_loop_mt(struct fuse_session *se_,
                "process-thread-count={}; "
                "process-thread-queue-depth={}; "
                "fuse-dev-ioc-clone={}; "
-               "pin-threads={};",
+               "pin-threads={}; "
+               "read-spin-us={};",
                read_thread_count,
                process_thread_count,
                process_thread_queue_depth,
                cloned_read_fds,
-               pin_threads_type_);
+               pin_threads_type_,
+               read_spin_us_);
 
   while(!fuse_session_exited(se_))
     sem_wait(&finished);
@@ -351,7 +390,8 @@ fuse_loop_mt(struct fuse *f_)
                              fuse_cfg.read_thread_count,
                              fuse_cfg.process_thread_count,
                              fuse_cfg.process_thread_queue_depth,
-                             fuse_cfg.pin_threads);
+                             fuse_cfg.pin_threads,
+                             fuse_cfg.read_spin_us);
 
   MaintenanceThread::stop();