/*
  FUSE: Filesystem in Userspace
  Copyright (C) 2001-2007  Miklos Szeredi <miklos@szeredi.hu>

  This program can be distributed under the terms of the GNU LGPLv2.
  See the file COPYING.LIB
*/

#ifndef _GNU_SOURCE
#define _GNU_SOURCE
#endif

#include "mutex.hpp"

#include "debug.hpp"
#include "fatal.hpp"
#include "fmt/core.h"
#include "fuse_cfg.hpp"
#include "fuse_i.hpp"
#include "fuse_kernel.h"
#include "fuse_msgbuf.hpp"
#include "fuse_opt.h"
#include "fuse_pollhandle.h"
#include "stat_utils.h"
#include "syslog.hpp"

#include <atomic>

#if defined(__linux__)
#include <fcntl.h>
#include <sys/ioctl.h>
#endif

#include <stdio.h>
#include <stdlib.h>
#include <stddef.h>
#include <string.h>
#include <unistd.h>
#include <limits.h>
#include <errno.h>
#include <assert.h>
#include <sys/file.h>

#define PARAM(inarg) (((char*)(inarg)) + sizeof(*(inarg)))
#define OFFSET_MAX 0x7fffffffffffffffLL

#define container_of(ptr, type, member) ({                      \
      const decltype( ((type*)0)->member ) *__mptr = (ptr);     \
      (type *)( (char*)__mptr - offsetof(type,member) );})

static size_t pagesize;

struct fuse_ll
{
  struct fuse_lowlevel_ops op;
  uid_t owner;
  fuse_conn_info_t conn;
  mutex_t lock;
  int got_init;
  int got_destroy;
  uint64_t notify_ctr;
  struct fuse_notify_req notify_list;
};

static fuse_ll f = {};

static
__attribute__((constructor))
void
fuse_ll_constructor(void)
{
  pagesize = sysconf(_SC_PAGESIZE);
  if(pagesize <= 0)
    fatal::abort("pagesize query failed - {}",strerror(errno));
}

static
void
convert_stat(const struct stat *stbuf_,
             struct fuse_attr  *attr_)
{
  attr_->ino       = stbuf_->st_ino;
  attr_->mode      = stbuf_->st_mode;
  attr_->nlink     = stbuf_->st_nlink;
  attr_->uid       = stbuf_->st_uid;
  attr_->gid       = stbuf_->st_gid;
  attr_->rdev      = stbuf_->st_rdev;
  attr_->size      = stbuf_->st_size;
  attr_->blksize   = stbuf_->st_blksize;
  attr_->blocks    = stbuf_->st_blocks;
  attr_->atime     = stbuf_->st_atime;
  attr_->mtime     = stbuf_->st_mtime;
  attr_->ctime     = stbuf_->st_ctime;
  attr_->atimensec = ST_ATIM_NSEC(stbuf_);
  attr_->mtimensec = ST_MTIM_NSEC(stbuf_);
  attr_->ctimensec = ST_CTIM_NSEC(stbuf_);
}

static
size_t
iov_length(const struct iovec *iov,
           size_t              count)
{
  size_t seg;
  size_t ret = 0;

  for(seg = 0; seg < count; seg++)
    ret += iov[seg].iov_len;
  return ret;
}

static
int
fuse_send_msg(const int     fd_,
              struct iovec *iov,
              int           count)
{
  int rv;
  struct fuse_out_header *out = (fuse_out_header*)iov[0].iov_base;

  out->len = iov_length(iov, count);

  rv = writev(fd_,iov,count);
  if(rv == -1)
    return -errno;

  return 0;
}

#define MAX_ERRNO 4095

int
fuse_send_reply_iov_nofree(fuse_req_t   *req,
                           int           error,
                           struct iovec *iov,
                           int           count)
{
  struct fuse_out_header out;

  if(error > 0)
    error = -error;

  if(error <= -MAX_ERRNO)
    {
      fprintf(stderr,"fuse: bad error value: %i\n",error);
      error = -ERANGE;
    }

  out.unique = req->ctx.unique;
  out.error  = error;

  iov[0].iov_base = &out;
  iov[0].iov_len  = sizeof(struct fuse_out_header);

  return fuse_send_msg(req->fd, iov, count);
}

static
int
send_reply_iov(fuse_req_t   *req,
               int           error,
               struct iovec *iov,
               int           count)
{
  int res;

  res = fuse_send_reply_iov_nofree(req, error, iov, count);
  fuse_req_free(req);

  return res;
}

static
int
send_reply(fuse_req_t *req,
           int         error,
           const void *arg,
           size_t      argsize)
{
  struct iovec iov[2];
  int count = 1;
  if(argsize)
    {
      iov[1].iov_base = (void *) arg;
      iov[1].iov_len = argsize;
      count++;
    }

  return send_reply_iov(req, error, iov, count);
}

static
void
convert_statfs(const struct statvfs *stbuf,
               struct fuse_kstatfs  *kstatfs)
{
  kstatfs->bsize   = stbuf->f_bsize;
  kstatfs->frsize  = stbuf->f_frsize;
  kstatfs->blocks  = stbuf->f_blocks;
  kstatfs->bfree   = stbuf->f_bfree;
  kstatfs->bavail  = stbuf->f_bavail;
  kstatfs->files   = stbuf->f_files;
  kstatfs->ffree   = stbuf->f_ffree;
  kstatfs->namelen = stbuf->f_namemax;
}

static
int
send_reply_ok(fuse_req_t *req,
              const void *arg,
              size_t      argsize)
{
  return send_reply(req, 0, arg, argsize);
}

int
fuse_reply_err(fuse_req_t *req_,
               int         err_)
{
  if(fuse_cfg.debug)
    {
      struct fuse_out_header hdr = {};
      hdr.unique = req_->ctx.unique;
      hdr.error  = (err_ > 0) ? -err_ : err_;
      hdr.len    = sizeof(struct fuse_out_header);
      fuse_debug_out_header(&hdr);
    }

  return send_reply(req_,err_,NULL,0);
}

void
fuse_reply_none(fuse_req_t *req)
{
  fuse_req_free(req);
}

static
void
fill_entry(struct fuse_entry_out         *arg,
           const struct fuse_entry_param *e)
{
  arg->nodeid           = e->ino;
  arg->generation       = e->generation;
  arg->entry_valid      = e->timeout.entry;
  arg->entry_valid_nsec = 0;
  arg->attr_valid       = e->timeout.attr;
  arg->attr_valid_nsec  = 0;
  convert_stat(&e->attr,&arg->attr);
}

static
void
fill_open(struct fuse_open_out   *arg_,
          const fuse_file_info_t *ffi_)
{
  arg_->fh = ffi_->fh;
  if(ffi_->direct_io)
    arg_->open_flags |= FOPEN_DIRECT_IO;
  if(ffi_->keep_cache)
    arg_->open_flags |= FOPEN_KEEP_CACHE;
  if(ffi_->nonseekable)
    arg_->open_flags |= FOPEN_NONSEEKABLE;
  if(ffi_->cache_readdir)
    arg_->open_flags |= FOPEN_CACHE_DIR;
  if(ffi_->parallel_direct_writes)
    arg_->open_flags |= FOPEN_PARALLEL_DIRECT_WRITES;
  if(ffi_->noflush)
    arg_->open_flags |= FOPEN_NOFLUSH;
  if(ffi_->passthrough && fuse_backing_id_is_valid(ffi_->backing_id))
    {
      arg_->open_flags |= FOPEN_PASSTHROUGH;
      arg_->backing_id  = ffi_->backing_id;
    }
}

int
fuse_reply_entry(fuse_req_t                    *req,
                 const struct fuse_entry_param *e)
{
  struct fuse_entry_out arg = {};
  size_t size = req->conn.proto_minor < 9 ?
    FUSE_COMPAT_ENTRY_OUT_SIZE : sizeof(arg);

  /* before ABI 7.4 e->ino == 0 was invalid, only ENOENT meant
     negative entry */
  if(!e->ino && req->conn.proto_minor < 4)
    return fuse_reply_err(req, ENOENT);

  fill_entry(&arg, e);

#ifdef NDEBUG
  // TODO: Add checks for cases where a node could be marked bad by
  // the kernel.
#endif

  if(fuse_cfg.debug)
    fuse_debug_entry_out(req->ctx.unique,&arg,size);

  return send_reply_ok(req, &arg, size);
}

struct fuse_create_out
{
  struct fuse_entry_out e;
  struct fuse_open_out o;
};

int
fuse_reply_create(fuse_req_t                    *req,
                  const struct fuse_entry_param *e,
                  const fuse_file_info_t        *f)
{
  struct fuse_create_out buf = {};
  size_t entrysize = req->conn.proto_minor < 9 ?
    FUSE_COMPAT_ENTRY_OUT_SIZE : sizeof(struct fuse_entry_out);
  struct fuse_entry_out *earg = (struct fuse_entry_out*)&buf.e;
  struct fuse_open_out  *oarg = (struct fuse_open_out*)(((char*)&buf)+entrysize);

  fill_entry(earg, e);
  fill_open(oarg, f);

  if(fuse_cfg.debug)
    fuse_debug_entry_open_out(req->ctx.unique,earg,oarg);

  return send_reply_ok(req, &buf, entrysize + sizeof(struct fuse_open_out));
}

int
fuse_reply_attr(fuse_req_t        *req,
                const struct stat *attr,
                const uint64_t     timeout)
{
  struct fuse_attr_out arg = {};
  size_t size = req->conn.proto_minor < 9 ?
    FUSE_COMPAT_ATTR_OUT_SIZE : sizeof(arg);

  arg.attr_valid      = timeout;
  arg.attr_valid_nsec = 0;
  convert_stat(attr,&arg.attr);

  if(fuse_cfg.debug)
    fuse_debug_attr_out(req->ctx.unique,&arg,size);

  return send_reply_ok(req,&arg,size);
}

int
fuse_reply_statx(fuse_req_t        *req_,
                 int                flags_,
                 struct fuse_statx *st_,
                 uint64_t           timeout_)
{
  struct fuse_statx_out outarg{};

  outarg.flags           = flags_;
  outarg.attr_valid      = timeout_;
  outarg.attr_valid_nsec = 0;
  outarg.stat            = *(struct fuse_statx*)st_;

  if(fuse_cfg.debug)
    fuse_debug_statx_out(req_->ctx.unique,&outarg);

  return send_reply_ok(req_,&outarg,sizeof(outarg));
}

int
fuse_reply_readlink(fuse_req_t   *req,
                    const char   *linkname,
                    const size_t  linkname_len)
{
  if(not linkname)
    return fuse_reply_err(req,EIO);

  if(fuse_cfg.debug)
    fuse_debug_readlink(req->ctx.unique,linkname,linkname_len);

  return send_reply_ok(req,linkname,linkname_len);
}

int
fuse_reply_open(fuse_req_t             *req,
                const fuse_file_info_t *f)
{
  struct fuse_open_out arg = {};

  fill_open(&arg, f);

  if(fuse_cfg.debug)
    fuse_debug_open_out(req->ctx.unique,&arg,sizeof(arg));

  return send_reply_ok(req, &arg, sizeof(arg));
}

int
fuse_reply_write(fuse_req_t *req,
                 size_t      count)
{
  struct fuse_write_out arg = {};

  arg.size = count;

  if(fuse_cfg.debug)
    fuse_debug_write_out(req->ctx.unique,&arg);

  return send_reply_ok(req, &arg, sizeof(arg));
}

int
fuse_reply_copy_file_range_64(fuse_req_t *req,
                              size_t      count)
{
  struct fuse_copy_file_range_out arg = {};

  arg.bytes_copied = count;

  return send_reply_ok(req, &arg, sizeof(arg));
}

int
fuse_reply_buf(fuse_req_t *req,
               const char *buf,
               size_t      size)
{
  if(fuse_cfg.debug)
    fuse_debug_data_out(req->ctx.unique,size);

  return send_reply_ok(req, buf, size);
}

#if defined(__linux__)
/* Automatic splice of large READ replies (branch fd -> pipe -> /dev/fuse).
   The copy path costs read + pread + writev and copies the payload
   twice. The splice path costs read + write(header) + splice(fd ->
   pipe) + splice(pipe -> /dev/fuse) and never copies the payload into
   userspace; the kernel copies it once, into the request's pages. The
   caller gates on fuse_reply_splice_fits (fuse_lowlevel.h).

   Every staging failure is recoverable: nothing has reached the kernel,
   so the pipe is drained and the caller replies through the copy path.
   Only a kernel without splice (ENOSYS), or a seccomp policy refusing
   it, disables the path for the whole process. A pipe-max-size too
   small for a reply only caps the reply size that is spliced
   (g_reply_pipe_limit), and a branch that rejects splice is reported
   per open file (FuseReplyResult::unsupported). */
static std::atomic<bool> g_reply_splice_off{false};

/* Smallest reply pipe, in bytes, found to exceed the pipe-max-size
   sysctl, which caps unprivileged F_SETPIPE_SZ; a 1MiB reply plus its
   header page needs one page more than the default 1MiB. Replies that
   need at least this much stay on the copy path unless this thread's
   pipe already holds them; smaller replies keep splicing. A refusal
   from the per-user pipe quota is not recorded: it lifts as pipes are
   closed. 0 = no refusal seen. */
static std::atomic<size_t> g_reply_pipe_limit{0};

/* The pipe-max-size sysctl, read once at session setup. Read lazily, on
   the first refused resize, it could fail for the same shortage of
   descriptors that made that resize or pipe fail and then read as 0 for
   good. 0 when it cannot be read, in which case every EPERM is treated
   as that limit. */
static std::atomic<size_t> g_pipe_max_size{0};

static
void
_read_pipe_max_size(void)
{
  unsigned long n = 0;
  FILE *f = ::fopen("/proc/sys/fs/pipe-max-size","re");

  if(f == NULL)
    return;
  if(::fscanf(f,"%lu",&n) == 1)
    g_pipe_max_size.store((size_t)n,std::memory_order_relaxed);
  ::fclose(f);
}

static
void
_reply_splice_disable(const char *what_,
                      const int   err_)
{
  if(g_reply_splice_off.exchange(true))
    return;
  SysLog::notice("read reply splice disabled: {} failed: {};"
                 " using copy path",
                 what_,strerror(err_));
}

/* One latch per distinct failure reason: these are independent,
   recurring conditions (a pipe quota hiccup, a header-staging error, a
   branch-splice error, a submission error, ...), so silencing all of
   them after whichever happens to fire first would hide later,
   unrelated failures from the log. A submission the kernel refused
   before it looked at the request (the copy path then answers it) is
   a different incident from one that fails the request, so they do
   not share a latch. */
enum class SpliceWarnReason
{
  pipe_quota,
  pipe_create,
  header_stage,
  branch_splice,
  submit_fallback,
  submit_failed,
  _count
};

static std::atomic<bool> g_reply_splice_warned[(size_t)SpliceWarnReason::_count] = {};

static
void
_reply_splice_warn(const char           *what_,
                   const int             err_,
                   const SpliceWarnReason reason_,
                   const char           *outcome_ =
                   "falling back to copy path for this reply")
{
  if(g_reply_splice_warned[(size_t)reason_].exchange(true))
    return;
  SysLog::warning("read reply splice: {} failed: {}; {}",
                  what_,strerror(err_),outcome_);
}

/* pipe_quota (the per-user pipe count/memory limit) is the one reason
   that is genuinely transient -- it lifts as pipes are closed, unlike
   the others, which are properties of the kernel or a branch fd that
   do not change at runtime. Clear its latch once a resize succeeds, so
   a later recurrence (a distinct incident) is logged again instead of
   being silenced forever by whichever incident fired first. */
static
void
_reply_splice_warn_clear(const SpliceWarnReason reason_)
{
  g_reply_splice_warned[(size_t)reason_].store(false,std::memory_order_relaxed);
}

/* Whether splice itself is unusable in this process: missing from the
   kernel (ENOSYS) or refused by a seccomp policy, which commonly
   answers EPERM but may use any errno. A zero-length splice returns 0
   before looking at its descriptors, so only then does it fail. An
   error from splicing a branch file normally comes from that file's
   own read path instead (ENOSYS from a FUSE branch whose server does
   not implement READ, EIO, ...), which this tells apart. */
static
bool
_splice_syscall_unusable()
{
  return (::splice(-1,nullptr,-1,nullptr,0,0) == -1);
}

static
void
fuse_send_errno(const int      fd_,
                const int      errno_,
                const uint64_t unique_id_);

namespace
{
  /* One pipe per worker thread, created on first use and reused for
     every large reply that thread sends. Empty between replies: every
     path that leaves bytes behind drains or closes it. */
  struct ReplyPipe
  {
    int    fd[2] = {-1,-1};
    size_t cap   = 0;

    ~ReplyPipe() { reset(); }

    void
    reset()
    {
      if(fd[0] != -1) ::close(fd[0]);
      if(fd[1] != -1) ::close(fd[1]);
      fd[0] = fd[1] = -1;
      cap = 0;
    }

    /* 0 when a pipe of at least bytes_ capacity is held, else -errno. */
    int
    ready(const size_t bytes_)
    {
      if(fd[0] == -1)
        {
          /* The initial capacity is not queried: cap 0 makes the
             first use resize to exactly what it needs, which with 4KiB
             pages it would have to do anyway (a reply of 64KiB plus
             its header needs more than the default 16 slots). */
          if(::pipe2(fd,O_CLOEXEC|O_NONBLOCK) == -1)
            return -errno;
        }
      if(bytes_ <= cap)
        return 0;
      if(bytes_ > (size_t)INT_MAX)
        return -EINVAL;
      const int c = ::fcntl(fd[0],F_SETPIPE_SZ,(int)bytes_);
      if(c == -1)
        return -errno;
      cap = (size_t)c;
      return ((cap >= bytes_) ? 0 : -ENOSPC);
    }

    /* /dev/null, opened on first use and kept. A failed open (EMFILE at
       that moment, say) is retried by the next drain rather than
       cached, which would leave every later drain copying. */
    static
    int
    devnull()
    {
      static std::atomic<int> g_fd{-1};
      int fd = g_fd.load(std::memory_order_acquire);

      if(fd != -1)
        return fd;

      fd = ::open("/dev/null",O_WRONLY|O_CLOEXEC);
      if(fd == -1)
        return -1;

      int expected = -1;
      if(g_fd.compare_exchange_strong(expected,fd,std::memory_order_acq_rel))
        return fd;

      ::close(fd);

      return expected;
    }

    /* Discard anything staged. Splicing it into /dev/null drops the
       page references without copying the data into userspace; reading
       it out is the fallback when that is not available (no /dev/null,
       a buffer error), so the pipe/capacity survive to be reused. Both
       ends are non-blocking, so each stops at empty (EAGAIN); a pipe
       that cannot be emptied by either is closed instead, and ready()
       makes a new one on next use. */
    void
    drain()
    {
      /* true once the pipe reads empty (EAGAIN), false on any other
         outcome. */
      const auto pump = [](auto op_)
      {
        for(;;)
          {
            const ssize_t rv = op_();
            if((rv > 0) || ((rv == -1) && (errno == EINTR)))
              continue;
            return ((rv == -1) && (errno == EAGAIN));
          }
      };

      const int devnull = ReplyPipe::devnull();
      if((devnull != -1) &&
         pump([&]{ return ::splice(fd[0],nullptr,devnull,nullptr,
                                   (size_t)1 << 30,SPLICE_F_NONBLOCK); }))
        return;

      char buf[16384];
      if(pump([&]{ return ::read(fd[0],buf,sizeof(buf)); }))
        return;

      reset();
    }
  };
}

static thread_local ReplyPipe t_reply_pipe;

/* Pipe capacity a reply needs: one slot for the header plus the file
   range, which may start mid-page. The kernel fills pipes one page per
   slot, so capacity is counted in pages, not bytes. */
static
size_t
_reply_pipe_need(const off_t  offset_,
                 const size_t size_)
{
  const size_t slots =
    (((uintptr_t)offset_ % pagesize) + size_ + pagesize - 1) / pagesize;

  return (1 + slots) * pagesize;
}

/* Whether a pipe of need_ bytes may be used: always below any recorded
   pipe-max-size refusal, and above it only when this thread's pipe
   already has that capacity (so no resize is attempted). */
static
bool
_reply_pipe_allowed(const size_t need_)
{
  const size_t limit = g_reply_pipe_limit.load(std::memory_order_relaxed);
  return ((limit == 0) || (need_ < limit) || (need_ <= t_reply_pipe.cap));
}

/* The pipe capacity a reply of size_ bytes at offset_ needs, or 0 when
   it is not spliced: splice is off, the reply is below the threshold
   (cheaper through read + writev), or it needs a pipe over a recorded
   pipe-max-size refusal. The one predicate behind both
   fuse_reply_splice_fits and fuse_reply_data_splice_fd. */
static
size_t
_reply_splice_need(const off_t  offset_,
                   const size_t size_)
{
  if(g_reply_splice_off.load(std::memory_order_relaxed) ||
     (size_ < FUSE_SPLICE_REPLY_MIN))
    return 0;

  const size_t need = _reply_pipe_need(offset_,size_);

  return (_reply_pipe_allowed(need) ? need : 0);
}

bool
fuse_reply_splice_fits(const off_t  offset_,
                       const size_t size_)
{
  return (_reply_splice_need(offset_,size_) != 0);
}

FuseReplyResult
fuse_reply_data_splice_fd(fuse_req_t *req_,
                          FileRegion &region_)
{
  const size_t need = _reply_splice_need(region_.offset,region_.size);
  if(need == 0)
    return FuseReplyResult::fallback;

  ReplyPipe &p = t_reply_pipe;
  struct fuse_out_header out = {};
  out.unique = req_->ctx.unique;
  out.len    = sizeof(out) + region_.size;

  const size_t prev_cap = p.cap;
  int err = p.ready(need);
  if(err != 0)
    {
      /* EPERM means either larger than pipe-max-size without
         CAP_SYS_RESOURCE, a fixed bound worth remembering so replies
         this large skip straight to the copy path, or over the
         per-user pipe quota, which is transient: only this reply falls
         back. An unreadable sysctl is treated as the fixed bound. */
      const size_t max = g_pipe_max_size.load(std::memory_order_relaxed);
      if((err == -EPERM) && ((max == 0) || (need > max)))
        {
          size_t cur = g_reply_pipe_limit.load(std::memory_order_relaxed);
          while(((cur == 0) || (need < cur)) &&
                !g_reply_pipe_limit.compare_exchange_weak(cur,need))
            ;
          /* cur is still 0 only in the thread that set the first limit. */
          if(cur == 0)
            SysLog::notice("read reply splice: a {} byte pipe exceeds"
                           " pipe-max-size; reads this large use the"
                           " copy path, smaller reads are still spliced",
                           need);
        }
      else if(err == -EPERM)
        _reply_splice_warn("growing the reply pipe (per-user pipe quota)",-err,
                           SpliceWarnReason::pipe_quota);
      else if(err == -ENOSYS)
        _reply_splice_disable("creating the reply pipe",-err);
      else
        _reply_splice_warn("creating the reply pipe",-err,
                           SpliceWarnReason::pipe_create);
      return FuseReplyResult::fallback;
    }

  /* A resize just succeeded, so any earlier per-user pipe quota
     shortage (the one transient SpliceWarnReason) has lifted; let a
     later recurrence be logged again instead of staying silenced by an
     earlier one. A pipe that was already big enough says nothing about
     the quota: it must not re-arm the warning while other threads still
     fail to grow theirs, nor write the shared latch on every reply. */
  if(p.cap != prev_cap)
    _reply_splice_warn_clear(SpliceWarnReason::pipe_quota);

  /* write(), not vmsplice(): vmsplice without SPLICE_F_GIFT queues a
     reference to this stack page rather than a copy, and the pipe
     outlives the frame whenever it is consumed later (or not at all).
     A 16-byte write into an empty pipe copies and is atomic. */
  ssize_t n;
  do
    n = ::write(p.fd[1],&out,sizeof(out));
  while((n == -1) && (errno == EINTR));
  if(n != (ssize_t)sizeof(out))
    {
      err = ((n == -1) ? errno : EIO);
      p.drain();
      _reply_splice_warn("staging the reply header",err,
                         SpliceWarnReason::header_stage);
      return FuseReplyResult::fallback;
    }

  size_t done = 0;
  while(done < region_.size)
    {
      loff_t pos = region_.offset + (off_t)done;
      n = ::splice(region_.fd,&pos,p.fd[1],nullptr,
                   region_.size - done,SPLICE_F_NONBLOCK);
      if((n == -1) && (errno == EINTR))
        continue;
      if(n > 0)
        {
          done += (size_t)n;
          continue;
        }

      /* n == 0: the file ended before the region did (it shrank after
         the caller sized the region). n == -1: this descriptor rejects
         splice (EINVAL: its filesystem has no splice_read) or failed.
         EAGAIN is also end-of-file when the file ended inside the
         region's last page of an exactly full pipe (no room is left
         for the splice that would return 0). Nothing reached the
         kernel. A rejecting descriptor is a property of that branch
         file, not of the transport, so the path stays on and the
         caller is told to stop offering it; only a first splice can
         show that, since one that already moved data proves the
         descriptor splices. An error disables the path only when
         splice itself is unusable (missing, or refused by seccomp),
         not when it is the branch file's read error; otherwise every
         large read would pay for a staging attempt that cannot
         succeed. */
      err = ((n == -1) ? errno : 0);
      p.drain();
      if((err != 0) && (err != EAGAIN))
        {
          if(_splice_syscall_unusable())
            _reply_splice_disable("splicing from the branch file",err);
          else if(err != EINVAL)
            _reply_splice_warn("splicing from the branch file",err,
                               SpliceWarnReason::branch_splice);
        }
      return (((err == EINVAL) && (done == 0)) ?
              FuseReplyResult::unsupported :
              FuseReplyResult::fallback);
    }

  /* fd is not read again on any outcome below (see FileRegion::lock). */
  if(region_.lock.owns_lock())
    region_.lock.unlock();

  /* The whole message is staged now, so a log write that throws (fmt on
     a full or broken log file) must not leave it in this thread's pipe:
     the worker survives the exception and would splice it in front of
     its next reply. */
  if(fuse_cfg.debug)
    {
      try
        {
          fuse_debug_data_out(req_->ctx.unique,region_.size);
        }
      catch(...)
        {
          p.drain();
          throw;
        }
    }

  /* Pages spliced in from the branch page cache are never offered for
     SPLICE_F_MOVE: stealing them would evict the branch's cache pages.
     The kernel copies them into the request's pages instead. */
  const size_t total = out.len;
  do
    n = ::splice(p.fd[0],nullptr,req_->fd,nullptr,total,0);
  while((n == -1) && (errno == EINTR));
  if(n == (ssize_t)total)
    return FuseReplyResult::sent;

  err = ((n == -1) ? errno : EIO);

  /* The kernel takes the whole message or none of it. If the pipe still
     holds all of it, the submission failed before the kernel looked at
     the request (e.g. ENOMEM allocating its buffer list), which is
     still pending, so the copy path can answer it. */
  int queued = 0;
  if((::ioctl(p.fd[0],FIONREAD,&queued) == 0) && ((size_t)queued == total))
    {
      p.drain();
      _reply_splice_warn("submitting the reply",err,
                         SpliceWarnReason::submit_fallback);
      return FuseReplyResult::fallback;
    }

  /* Otherwise the kernel may have completed the request before failing
     (it validates after copying; a copy error fails it with EIO), so
     the payload is never replayed. ENOENT means the request was already
     retired or the connection aborted: nothing is left to answer, as
     when the copy path's writev fails the same way, so it is neither
     logged nor answered. Any other error-only reply completes a request
     that is still pending. */
  p.drain();
  if(err != ENOENT)
    {
      _reply_splice_warn("submitting the reply",err,
                         SpliceWarnReason::submit_failed,
                         "failing the request");

      fuse_send_errno(req_->fd,err,req_->ctx.unique);
    }

  return FuseReplyResult::failed;
}
#else
bool
fuse_reply_splice_fits(const off_t,
                       const size_t)
{
  return false;
}

FuseReplyResult
fuse_reply_data_splice_fd(fuse_req_t *,
                          FileRegion &)
{
  return FuseReplyResult::fallback;
}
#endif

int
fuse_reply_data(fuse_req_t   *req,
                char         *buf_,
                const size_t  bufsize_)
{
  int res;
  struct iovec iov[2];
  struct fuse_out_header out;

  iov[0].iov_base = &out;
  iov[0].iov_len  = sizeof(struct fuse_out_header);
  iov[1].iov_base = buf_;
  iov[1].iov_len  = bufsize_;

  out.unique = req->ctx.unique;
  out.error = 0;

  if(fuse_cfg.debug)
    fuse_debug_data_out(req->ctx.unique,bufsize_);

  res = fuse_send_msg(req->fd,iov,2);
  fuse_req_free(req);

  return res;
}

int
fuse_reply_statfs(fuse_req_t           *req,
                  const struct statvfs *stbuf)
{
  struct fuse_statfs_out arg = {};
  size_t size = req->conn.proto_minor < 4 ?
    FUSE_COMPAT_STATFS_SIZE : sizeof(arg);

  convert_statfs(stbuf, &arg.st);

  if(fuse_cfg.debug)
    fuse_debug_statfs_out(req->ctx.unique,&arg);

  return send_reply_ok(req, &arg, size);
}

int
fuse_reply_xattr(fuse_req_t *req,
                 size_t      count)
{
  struct fuse_getxattr_out arg = {};

  arg.size = count;

  if(fuse_cfg.debug)
    fuse_debug_getxattr_out(req->ctx.unique,&arg);

  return send_reply_ok(req, &arg, sizeof(arg));
}

int
fuse_reply_bmap(fuse_req_t *req,
                uint64_t    idx)
{
  struct fuse_bmap_out arg = {};

  arg.block = idx;

  if(fuse_cfg.debug)
    fuse_debug_bmap_out(req->ctx.unique,&arg);

  return send_reply_ok(req, &arg, sizeof(arg));
}

static
struct fuse_ioctl_iovec*
fuse_ioctl_iovec_copy(const struct iovec *iov,
                      size_t              count)
{
  struct fuse_ioctl_iovec *fiov;
  size_t i;

  fiov = (fuse_ioctl_iovec*)malloc(sizeof(fiov[0]) * count);
  if(!fiov)
    return NULL;

  for (i = 0; i < count; i++)
    {
      fiov[i].base = (uintptr_t) iov[i].iov_base;
      fiov[i].len = iov[i].iov_len;
    }

  return fiov;
}

int
fuse_reply_ioctl_retry(fuse_req_t         *req,
                       const struct iovec *in_iov,
                       size_t              in_count,
                       const struct iovec *out_iov,
                       size_t              out_count)
{
  int res;
  struct fuse_ioctl_out arg = {};
  struct fuse_ioctl_iovec *in_fiov = NULL;
  struct fuse_ioctl_iovec *out_fiov = NULL;
  struct iovec iov[4];
  size_t count = 1;

  arg.flags |= FUSE_IOCTL_RETRY;
  arg.in_iovs = in_count;
  arg.out_iovs = out_count;

  if(fuse_cfg.debug)
    fuse_debug_ioctl_out(req->ctx.unique,&arg);
  iov[count].iov_base = &arg;
  iov[count].iov_len = sizeof(arg);
  count++;

  if(req->conn.proto_minor < 16)
    {
      if(in_count)
        {
          iov[count].iov_base = (void *)in_iov;
          iov[count].iov_len = sizeof(in_iov[0]) * in_count;
          count++;
        }

      if(out_count)
        {
          iov[count].iov_base = (void *)out_iov;
          iov[count].iov_len = sizeof(out_iov[0]) * out_count;
          count++;
        }
    }
  else
    {
      /* Can't handle non-compat 64bit ioctls on 32bit */
      if((sizeof(void *) == 4) && (req->ioctl_64bit))
        {
          res = fuse_reply_err(req, EINVAL);
          goto out;
        }

      if(in_count)
        {
          in_fiov = fuse_ioctl_iovec_copy(in_iov, in_count);
          if(!in_fiov)
            goto enomem;

          iov[count].iov_base = (void *)in_fiov;
          iov[count].iov_len = sizeof(in_fiov[0]) * in_count;
          count++;
        }
      if(out_count)
        {
          out_fiov = fuse_ioctl_iovec_copy(out_iov, out_count);
          if(!out_fiov)
            goto enomem;

          iov[count].iov_base = (void *)out_fiov;
          iov[count].iov_len = sizeof(out_fiov[0]) * out_count;
          count++;
        }
    }

  res = send_reply_iov(req, 0, iov, count);

 out:
  free(in_fiov);
  free(out_fiov);

  return res;

 enomem:
  res = fuse_reply_err(req, ENOMEM);
  goto out;
}

int
fuse_reply_ioctl(fuse_req_t *req,
                 int         result,
                 const void *buf,
                 uint32_t    size)
{
  int count;
  struct iovec iov[3];
  struct fuse_ioctl_out arg;

  arg.result   = result;
  arg.flags    = 0;
  arg.in_iovs  = 0;
  arg.out_iovs = 0;

  if(fuse_cfg.debug)
    fuse_debug_ioctl_out(req->ctx.unique,&arg);

  count = 1;
  iov[count].iov_base = &arg;
  iov[count].iov_len  = sizeof(arg);
  count++;

  if(size)
    {
      iov[count].iov_base = (char*)buf;
      iov[count].iov_len  = size;
      count++;
    }

  return send_reply_iov(req, 0, iov, count);
}

int
fuse_reply_poll(fuse_req_t *req,
                unsigned    revents)
{
  struct fuse_poll_out arg = {};

  arg.revents = revents;

  if(fuse_cfg.debug)
    fuse_debug_poll_out(req->ctx.unique,&arg);

  return send_reply_ok(req, &arg, sizeof(arg));
}

static
void
do_lookup(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.lookup(req,hdr_);
}

static
void
do_forget(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.forget(req,hdr_);
}

static
void
do_batch_forget(fuse_req_t            *req,
                struct fuse_in_header *hdr_)
{
  f.op.forget_multi(req,hdr_);
}

static
void
do_getattr(fuse_req_t            *req,
           struct fuse_in_header *hdr_)
{
  f.op.getattr(req, hdr_);
}

static
void
do_setattr(fuse_req_t            *req_,
           struct fuse_in_header *hdr_)
{
  f.op.setattr(req_,hdr_);
}

static
void
do_access(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.access(req,hdr_);
}

static
void
do_readlink(fuse_req_t            *req,
            struct fuse_in_header *hdr_)
{
  f.op.readlink(req,hdr_);
}

static
void
do_mknod(fuse_req_t            *req,
         struct fuse_in_header *hdr_)
{
  f.op.mknod(req,hdr_);
}

static
void
do_mkdir(fuse_req_t            *req,
         struct fuse_in_header *hdr_)
{
  f.op.mkdir(req,hdr_);
}

static
void
do_unlink(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.unlink(req,hdr_);
}

static
void
do_rmdir(fuse_req_t            *req,
         struct fuse_in_header *hdr_)
{
  f.op.rmdir(req,hdr_);
}

static
void
do_symlink(fuse_req_t            *req,
           struct fuse_in_header *hdr_)
{
  f.op.symlink(req,hdr_);
}

static
void
do_rename(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.rename(req,hdr_);
}

static
void
do_link(fuse_req_t            *req,
        struct fuse_in_header *hdr_)
{
  f.op.link(req,hdr_);
}

static
void
do_create(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.create(req,hdr_);
}

static
void
do_open(fuse_req_t            *req,
        struct fuse_in_header *hdr_)
{
  f.op.open(req,hdr_);
}

static
void
do_read(fuse_req_t            *req,
        struct fuse_in_header *hdr_)
{
  f.op.read(req,hdr_);
}

static
void
do_write(fuse_req_t            *req,
         struct fuse_in_header *hdr_)
{
  f.op.write(req,hdr_);
}

static
void
do_flush(fuse_req_t            *req,
         struct fuse_in_header *hdr_)
{
  f.op.flush(req,hdr_);
}

static
void
do_release(fuse_req_t            *req,
           struct fuse_in_header *hdr_)
{
  f.op.release(req,hdr_);
}

static
void
do_fsync(fuse_req_t            *req,
         struct fuse_in_header *hdr_)
{
  f.op.fsync(req,hdr_);
}

static
void
do_opendir(fuse_req_t            *req,
           struct fuse_in_header *hdr_)
{
  f.op.opendir(req,hdr_);
}

static
void
do_readdir(fuse_req_t            *req,
           struct fuse_in_header *hdr_)
{
  f.op.readdir(req,hdr_);
}

static
void
do_readdirplus(fuse_req_t            *req_,
               struct fuse_in_header *hdr_)
{
  f.op.readdir_plus(req_,hdr_);
}

static
void
do_releasedir(fuse_req_t            *req,
              struct fuse_in_header *hdr_)
{
  f.op.releasedir(req,hdr_);
}

static
void
do_fsyncdir(fuse_req_t            *req,
            struct fuse_in_header *hdr_)
{
  f.op.fsyncdir(req,hdr_);
}

static
void
do_statfs(fuse_req_t            *req,
          struct fuse_in_header *hdr_)
{
  f.op.statfs(req,hdr_);
}

static
void
do_setxattr(fuse_req_t            *req,
            struct fuse_in_header *hdr_)
{
  f.op.setxattr(req,hdr_);
}

static
void
do_getxattr(fuse_req_t            *req,
            struct fuse_in_header *hdr_)
{
  f.op.getxattr(req,hdr_);
}

static
void
do_listxattr(fuse_req_t            *req,
             struct fuse_in_header *hdr_)
{
  f.op.listxattr(req,hdr_);
}

static
void
do_removexattr(fuse_req_t            *req,
               struct fuse_in_header *hdr_)
{
  f.op.removexattr(req,hdr_);
}

void
do_interrupt(fuse_req_t *req,
             struct fuse_in_header *hdr_)
{
  fuse_req_free(req);
}

static
void
do_bmap(fuse_req_t            *req,
        struct fuse_in_header *hdr_)
{
  f.op.bmap(req,hdr_);
}

static
void
do_ioctl(fuse_req_t *req,
         struct fuse_in_header *hdr_)
{
  f.op.ioctl(req, hdr_);
}

void
fuse_pollhandle_destroy(fuse_pollhandle_t *ph)
{
  free(ph);
}

static
void
do_poll(fuse_req_t            *req,
        struct fuse_in_header *hdr_)
{
  f.op.poll(req,hdr_);
}

static
void
do_fallocate(fuse_req_t            *req,
             struct fuse_in_header *hdr_)
{
  f.op.fallocate(req,hdr_);
}

static
void
do_init(fuse_req_t            *req,
        struct fuse_in_header *hdr_)
{
  struct fuse_init_out outarg = {};
  struct fuse_init_in *arg = (struct fuse_init_in *)&hdr_[1];
  size_t bufsize;
  uint64_t inargflags;
  uint64_t outargflags;
  u32 max_write;
  u32 max_readahead;

  max_write = UINT_MAX;
  max_readahead = UINT_MAX;
  bufsize = req->se->bufsize;

  inargflags = 0;
  outargflags = 0;

  fuse_syslog_fuse_init_in(arg);

  f.conn.proto_major = arg->major;
  f.conn.proto_minor = arg->minor;
  f.conn.capable = 0;
  f.conn.want = 0;

  outarg.major     = FUSE_KERNEL_VERSION;
  outarg.minor     = FUSE_KERNEL_MINOR_VERSION;
  outarg.max_pages = FUSE_DEFAULT_MAX_PAGES_PER_REQ;

  if(arg->major < 7)
    {
      fprintf(stderr, "fuse: unsupported protocol version: %u.%u\n",
              arg->major, arg->minor);
      fuse_reply_err(req, EPROTO);
      return;
    }

  if(arg->major > 7)
    {
      /* Wait for a second INIT request with a 7.X version */
      send_reply_ok(req, &outarg, sizeof(outarg));
      return;
    }

  if(arg->minor >= 6)
    {
      inargflags = arg->flags;
      if(inargflags & FUSE_INIT_EXT)
        inargflags |= (((uint64_t)arg->flags2) << 32);

      if(arg->max_readahead < max_readahead)
        max_readahead = arg->max_readahead;

      if(inargflags & FUSE_ASYNC_READ)
        f.conn.capable |= FUSE_CAP_ASYNC_READ;
      if(inargflags & FUSE_ATOMIC_O_TRUNC)
        f.conn.capable |= FUSE_CAP_ATOMIC_O_TRUNC;
      if(inargflags & FUSE_EXPORT_SUPPORT)
        f.conn.capable |= FUSE_CAP_EXPORT_SUPPORT;
      if(inargflags & FUSE_BIG_WRITES)
        f.conn.capable |= FUSE_CAP_BIG_WRITES;
      if(inargflags & FUSE_DONT_MASK)
        f.conn.capable |= FUSE_CAP_DONT_MASK;
      if(inargflags & FUSE_POSIX_ACL)
        f.conn.capable |= FUSE_CAP_POSIX_ACL;
      if(inargflags & FUSE_CACHE_SYMLINKS)
        f.conn.capable |= FUSE_CAP_CACHE_SYMLINKS;
      if(inargflags & FUSE_ASYNC_DIO)
        f.conn.capable |= FUSE_CAP_ASYNC_DIO;
      if(inargflags & FUSE_PARALLEL_DIROPS)
        f.conn.capable |= FUSE_CAP_PARALLEL_DIROPS;
      if(inargflags & FUSE_MAX_PAGES)
        f.conn.capable |= FUSE_CAP_MAX_PAGES;
      if(inargflags & FUSE_WRITEBACK_CACHE)
        f.conn.capable |= FUSE_CAP_WRITEBACK_CACHE;
      if(inargflags & FUSE_DO_READDIRPLUS)
        f.conn.capable |= FUSE_CAP_READDIR_PLUS;
      if(inargflags & FUSE_READDIRPLUS_AUTO)
        f.conn.capable |= FUSE_CAP_READDIR_PLUS_AUTO;
      if(inargflags & FUSE_SETXATTR_EXT)
        f.conn.capable |= FUSE_CAP_SETXATTR_EXT;
      if(inargflags & FUSE_DIRECT_IO_ALLOW_MMAP)
        f.conn.capable |= FUSE_CAP_DIRECT_IO_ALLOW_MMAP;
      if(inargflags & FUSE_CREATE_SUPP_GROUP)
        f.conn.capable |= FUSE_CAP_CREATE_SUPP_GROUP;
      if(inargflags & FUSE_PASSTHROUGH)
        f.conn.capable |= FUSE_CAP_PASSTHROUGH;
      if(inargflags & FUSE_HANDLE_KILLPRIV)
        f.conn.capable |= FUSE_CAP_HANDLE_KILLPRIV;
      if(inargflags & FUSE_HANDLE_KILLPRIV_V2)
        f.conn.capable |= FUSE_CAP_HANDLE_KILLPRIV_V2;
      if(inargflags & FUSE_ALLOW_IDMAP)
        f.conn.capable |= FUSE_CAP_ALLOW_IDMAP;
    }
  else
    {
      f.conn.want &= ~FUSE_CAP_ASYNC_READ;
      max_readahead = 0;
    }

  if(f.conn.proto_minor >= 18)
    f.conn.capable |= FUSE_CAP_IOCTL_DIR;

  if(bufsize < FUSE_MIN_READ_BUFFER)
    {
      fprintf(stderr, "fuse: warning: buffer size too small: %zu\n",
              bufsize);
      bufsize = FUSE_MIN_READ_BUFFER;
    }

  bufsize -= pagesize;
  if(bufsize < max_write)
    max_write = bufsize;

  f.got_init = 1;
  f.op.init(&f.conn);

  outargflags = outarg.flags;
  if((inargflags & FUSE_MAX_PAGES) && (f.conn.want & FUSE_CAP_MAX_PAGES))
    {
      outargflags      |= FUSE_MAX_PAGES;
      outarg.max_pages  = fuse_cfg.max_pages;

      msgbuf_set_bufsize(outarg.max_pages);
      max_write = (msgbuf_get_pagesize() * outarg.max_pages);
    }

  if(f.conn.want & FUSE_CAP_ASYNC_READ)
    outargflags |= FUSE_ASYNC_READ;
  if(f.conn.want & FUSE_CAP_ATOMIC_O_TRUNC)
    outargflags |= FUSE_ATOMIC_O_TRUNC;
  if(f.conn.want & FUSE_CAP_EXPORT_SUPPORT)
    outargflags |= FUSE_EXPORT_SUPPORT;
  if(f.conn.want & FUSE_CAP_BIG_WRITES)
    outargflags |= FUSE_BIG_WRITES;
  if(f.conn.want & FUSE_CAP_DONT_MASK)
    outargflags |= FUSE_DONT_MASK;
  if(f.conn.want & FUSE_CAP_POSIX_ACL)
    outargflags |= FUSE_POSIX_ACL;
  if(f.conn.want & FUSE_CAP_CACHE_SYMLINKS)
    outargflags |= FUSE_CACHE_SYMLINKS;
  if(f.conn.want & FUSE_CAP_ASYNC_DIO)
    outargflags |= FUSE_ASYNC_DIO;
  if(f.conn.want & FUSE_CAP_PARALLEL_DIROPS)
    outargflags |= FUSE_PARALLEL_DIROPS;
  if(f.conn.want & FUSE_CAP_WRITEBACK_CACHE)
    outargflags |= FUSE_WRITEBACK_CACHE;
  if(f.conn.want & FUSE_CAP_READDIR_PLUS)
    outargflags |= FUSE_DO_READDIRPLUS;
  if(f.conn.want & FUSE_CAP_READDIR_PLUS_AUTO)
    outargflags |= FUSE_READDIRPLUS_AUTO;
  if(f.conn.want & FUSE_CAP_SETXATTR_EXT)
    outargflags |= FUSE_SETXATTR_EXT;
  if(f.conn.want & FUSE_CAP_CREATE_SUPP_GROUP)
    outargflags |= FUSE_CREATE_SUPP_GROUP;
  if(f.conn.want & FUSE_CAP_DIRECT_IO_ALLOW_MMAP)
    outargflags |= FUSE_DIRECT_IO_ALLOW_MMAP;
  if(f.conn.want & FUSE_CAP_HANDLE_KILLPRIV)
    outargflags |= FUSE_HANDLE_KILLPRIV;
  if(f.conn.want & FUSE_CAP_HANDLE_KILLPRIV_V2)
    outargflags |= FUSE_HANDLE_KILLPRIV_V2;
  if(f.conn.want & FUSE_CAP_ALLOW_IDMAP)
    outargflags |= FUSE_ALLOW_IDMAP;

  if(f.conn.want & FUSE_CAP_PASSTHROUGH)
    {
      outargflags |= FUSE_PASSTHROUGH;
      outarg.max_stack_depth = fuse_cfg.passthrough_max_stack_depth;
    }

  if((inargflags & FUSE_REQUEST_TIMEOUT) && fuse_cfg.request_timeout)
    {
      outargflags |= FUSE_REQUEST_TIMEOUT;
      outarg.request_timeout = fuse_cfg.request_timeout;
    }

  if(inargflags & FUSE_INIT_EXT)
    {
      outargflags |= FUSE_INIT_EXT;
      outarg.flags2 = (outargflags >> 32);
    }

  outarg.flags = outargflags;

  outarg.max_readahead = max_readahead;
  outarg.max_write = max_write;
  if(f.conn.proto_minor >= 13)
    {
      if(fuse_cfg.max_background >= (1 << 16))
        fuse_cfg.max_background = ((1 << 16) - 1);
      if(fuse_cfg.congestion_threshold > fuse_cfg.max_background)
        fuse_cfg.congestion_threshold = fuse_cfg.max_background;
      if(!fuse_cfg.congestion_threshold)
        fuse_cfg.congestion_threshold = (fuse_cfg.max_background * 3 / 4);

      outarg.max_background = fuse_cfg.max_background;
      outarg.congestion_threshold = fuse_cfg.congestion_threshold;
    }

  if(f.conn.proto_minor >= 23)
    outarg.time_gran = 1;

  size_t outargsize;
  if(arg->minor < 5)
    outargsize = FUSE_COMPAT_INIT_OUT_SIZE;
  else if(arg->minor < 23)
    outargsize = FUSE_COMPAT_22_INIT_OUT_SIZE;
  else
    outargsize = sizeof(outarg);

  fuse_syslog_fuse_init_out(&outarg);

  if(fuse_cfg.debug)
    fuse_debug_init_out(req->ctx.unique,&outarg,outargsize);

  send_reply_ok(req, &outarg, outargsize);
}

static
void
do_destroy(fuse_req_t            *req,
           struct fuse_in_header *hdr_)
{
  f.got_destroy = 1;

  f.op.destroy();

  send_reply_ok(req,NULL,0);
}

static
void
list_del_nreq(struct fuse_notify_req *nreq)
{
  struct fuse_notify_req *prev = nreq->prev;
  struct fuse_notify_req *next = nreq->next;
  prev->next = next;
  next->prev = prev;
}

static
void
list_add_nreq(struct fuse_notify_req *nreq,
              struct fuse_notify_req *next)
{
  struct fuse_notify_req *prev = next->prev;
  nreq->next = next;
  nreq->prev = prev;
  prev->next = nreq;
  next->prev = nreq;
}

static
void
list_init_nreq(struct fuse_notify_req *nreq)
{
  nreq->next = nreq;
  nreq->prev = nreq;
}

static
void
do_notify_reply(fuse_req_t            *req,
                struct fuse_in_header *hdr_)
{
  struct fuse_notify_req *nreq;
  struct fuse_notify_req *head;

  mutex_lock(f.lock);
  head = &f.notify_list;
  for(nreq = head->next; nreq != head; nreq = nreq->next)
    {
      if(nreq->unique == req->ctx.unique)
        {
          list_del_nreq(nreq);
          break;
        }
    }
  mutex_unlock(f.lock);

  if(nreq != head)
    nreq->reply(nreq, req, hdr_->nodeid, &hdr_[1]);
}

static
void
do_copy_file_range(fuse_req_t            *req_,
                   struct fuse_in_header *hdr_)
{
  f.op.copy_file_range(req_,hdr_);
}

static
void
do_copy_file_range_64(fuse_req_t            *req_,
                      struct fuse_in_header *hdr_)
{
  f.op.copy_file_range(req_,hdr_);
}

static
void
do_setupmapping(fuse_req_t            *req_,
                struct fuse_in_header *hdr_)
{
  f.op.setupmapping(req_,hdr_);
}

static
void
do_removemapping(fuse_req_t            *req_,
                 struct fuse_in_header *hdr_)
{
  f.op.removemapping(req_,hdr_);
}

static
void
do_syncfs(fuse_req_t            *req_,
          struct fuse_in_header *hdr_)
{
  f.op.syncfs(req_,hdr_);
}

static
void
do_tmpfile(fuse_req_t            *req_,
           struct fuse_in_header *hdr_)
{
  f.op.tmpfile(req_,hdr_);
}

static
void
do_statx(fuse_req_t            *req_,
         struct fuse_in_header *hdr_)
{
  f.op.statx(req_,hdr_);
}

static
void
do_rename2(fuse_req_t            *req_,
           struct fuse_in_header *hdr_)
{
  f.op.rename2(req_,hdr_);
}

static
void
do_lseek(fuse_req_t            *req_,
         struct fuse_in_header *hdr_)
{
  f.op.lseek(req_,hdr_);
}

static
int
send_notify_iov(struct fuse_session *se,
                int                  notify_code,
                struct iovec        *iov,
                int                  count)
{
  struct fuse_out_header out;

  if(!f.got_init)
    return -ENOTCONN;

  out.unique = 0;
  out.error = notify_code;
  iov[0].iov_base = &out;
  iov[0].iov_len = sizeof(struct fuse_out_header);

  return fuse_send_msg(se->fd, iov, count);
}

int
fuse_lowlevel_notify_poll(fuse_pollhandle_t *ph)
{
  if(ph != NULL)
    {
      struct fuse_notify_poll_wakeup_out outarg;
      struct iovec iov[2];

      outarg.kh = ph->kh;

      iov[1].iov_base = &outarg;
      iov[1].iov_len = sizeof(outarg);

      return send_notify_iov(ph->se, FUSE_NOTIFY_POLL, iov, 2);
    }
  else
    {
      return 0;
    }
}

int
fuse_lowlevel_notify_inval_inode(struct fuse_session *se,
                                 uint64_t             ino,
                                 off_t                off,
                                 off_t                len)
{
  struct fuse_notify_inval_inode_out outarg;
  struct iovec iov[2];

  if(!se)
    return -EINVAL;

  outarg.ino = ino;
  outarg.off = off;
  outarg.len = len;

  iov[1].iov_base = &outarg;
  iov[1].iov_len  = sizeof(outarg);

  return send_notify_iov(se, FUSE_NOTIFY_INVAL_INODE, iov, 2);
}

int
fuse_lowlevel_notify_inval_entry(struct fuse_session *se,
                                 uint64_t             parent,
                                 const char          *name,
                                 size_t               namelen)
{
  struct fuse_notify_inval_entry_out outarg;
  struct iovec iov[3];

  if(!se)
    return -EINVAL;

  outarg.parent  = parent;
  outarg.namelen = namelen;
  // TODO: Add ability to set `flags`
  outarg.flags   = 0;

  iov[1].iov_base = &outarg;
  iov[1].iov_len = sizeof(outarg);
  iov[2].iov_base = (void *)name;
  iov[2].iov_len = namelen + 1;

  return send_notify_iov(se, FUSE_NOTIFY_INVAL_ENTRY, iov, 3);
}

int
fuse_lowlevel_notify_delete(struct fuse_session *se,
                            uint64_t             parent,
                            uint64_t             child,
                            const char          *name,
                            size_t               namelen)
{
  struct fuse_notify_delete_out outarg;
  struct iovec iov[3];

  if(!se)
    return -EINVAL;

  if(f.conn.proto_minor < 18)
    return -ENOSYS;

  outarg.parent = parent;
  outarg.child = child;
  outarg.namelen = namelen;
  outarg.padding = 0;

  iov[1].iov_base = &outarg;
  iov[1].iov_len = sizeof(outarg);
  iov[2].iov_base = (void *)name;
  iov[2].iov_len = namelen + 1;

  return send_notify_iov(se, FUSE_NOTIFY_DELETE, iov, 3);
}

struct fuse_retrieve_req
{
  struct fuse_notify_req nreq;
  void *cookie;
};

static
void
fuse_ll_retrieve_reply(struct fuse_notify_req *nreq,
                       fuse_req_t             *req,
                       uint64_t                ino,
                       const void             *inarg)
{
  struct fuse_retrieve_req *rreq =
    container_of(nreq, struct fuse_retrieve_req, nreq);

  fuse_reply_none(req);

  free(rreq);
}

int
fuse_lowlevel_notify_retrieve(struct fuse_session *se,
                                uint64_t             ino,
                                size_t               size,
                                off_t                offset,
                                void                *cookie)
{
  struct fuse_notify_retrieve_out outarg;
  struct iovec iov[2];
  struct fuse_retrieve_req *rreq;
  int err;

  if(!se)
    return -EINVAL;

  if(f.conn.proto_minor < 15)
    return -ENOSYS;

  rreq = (fuse_retrieve_req*)malloc(sizeof(*rreq));
  if(rreq == NULL)
    return -ENOMEM;

  mutex_lock(f.lock);
  rreq->cookie = cookie;
  rreq->nreq.unique = f.notify_ctr++;
  rreq->nreq.reply = fuse_ll_retrieve_reply;
  list_add_nreq(&rreq->nreq, &f.notify_list);
  mutex_unlock(f.lock);

  outarg.notify_unique = rreq->nreq.unique;
  outarg.nodeid = ino;
  outarg.offset = offset;
  outarg.size = size;

  iov[1].iov_base = &outarg;
  iov[1].iov_len = sizeof(outarg);

  err = send_notify_iov(se, FUSE_NOTIFY_RETRIEVE, iov, 2);
  if(err)
    {
      mutex_lock(f.lock);
      list_del_nreq(&rreq->nreq);
      mutex_unlock(f.lock);
      free(rreq);
    }

  return err;
}

#define FUSE_OPCODE_LEN (FUSE_COPY_FILE_RANGE_64 + 1)

typedef void (*fuse_ll_func)(fuse_req_t*, struct fuse_in_header *);
const
fuse_ll_func
fuse_ll_funcs[FUSE_OPCODE_LEN] =
  {
    NULL,
    do_lookup,
    do_forget,
    do_getattr,
    do_setattr,
    do_readlink,
    do_symlink,
    NULL,
    do_mknod,
    do_mkdir,
    do_unlink,
    do_rmdir,
    do_rename,
    do_link,
    do_open,
    do_read,
    do_write,
    do_statfs,
    do_release,
    NULL,
    do_fsync,
    do_setxattr,
    do_getxattr,
    do_listxattr,
    do_removexattr,
    do_flush,
    do_init,
    do_opendir,
    do_readdir,
    do_releasedir,
    do_fsyncdir,
    NULL,
    NULL,
    NULL,
    do_access,
    do_create,
    do_interrupt,
    do_bmap,
    do_destroy,
    do_ioctl,
    do_poll,
    do_notify_reply,
    do_batch_forget,
    do_fallocate,
    do_readdirplus,
    do_rename2,
    do_lseek,
    do_copy_file_range,
    do_setupmapping,
    do_removemapping,
    do_syncfs,
    do_tmpfile,
    do_statx,
    do_copy_file_range_64
  };

#define FUSE_MAXOPS (sizeof(fuse_ll_funcs) / sizeof(fuse_ll_funcs[0]))

static
void
fuse_ll_destroy(void *data)
{
  if(f.got_init && !f.got_destroy)
    f.op.destroy();

  mutex_destroy(f.lock);
  msgbuf_gc();
}

static
void
fuse_send_errno(const int            fd_,
                const int            errno_,
                const uint64_t       unique_id_)
{
  struct fuse_out_header out = {};
  struct iovec           iov = {};

  out.unique   = unique_id_;
  out.error    = -errno_;
  out.len      = sizeof(struct fuse_out_header);
  iov.iov_base = &out;
  iov.iov_len  = sizeof(struct fuse_out_header);

  if(fuse_cfg.debug)
    fuse_debug_out_header(&out);

  fuse_send_msg(fd_,&iov,1);
}

static
void
fuse_send_enomem(const int            fd_,
                 const uint64_t       unique_id_)
{
  fuse_send_errno(fd_,ENOMEM,unique_id_);
}

static
int
fuse_ll_buf_receive_read(struct fuse_session *se_,
                         int                  fd_,
                         fuse_msgbuf_t       *msgbuf_)
{
  int rv;

  rv = read(fd_,msgbuf_->mem,msgbuf_->size);
  if(rv == -1)
    return -errno;

  if(rv < (int)sizeof(struct fuse_in_header))
    {
      fprintf(stderr, "short read from fuse device\n");
      return -EIO;
    }

  return rv;
}

static
void
fuse_ll_buf_process_read(struct fuse_session *se_,
                         int                  fd_,
                         const fuse_msgbuf_t *msgbuf_)
{
  int err;
  struct fuse_req_t *req;
  struct fuse_in_header *in;

  in = (struct fuse_in_header*)msgbuf_->mem;

  if(fuse_cfg.debug)
    fuse_debug_in_header(in);

  req = fuse_req_alloc();
  if(req == NULL)
    return fuse_send_enomem(fd_,in->unique);

  req->ctx.len    = in->len;
  req->ctx.opcode = in->opcode;
  req->ctx.unique = in->unique;
  req->ctx.nodeid = in->nodeid;
  req->ctx.uid    = in->uid;
  req->ctx.gid    = in->gid;
  req->ctx.pid    = in->pid;
  req->ctx.umask  = 0;
  req->conn       = f.conn;
  req->se         = se_;
  req->fd         = fd_;
  req->ioctl_64bit = 0;

  err = ENOSYS;
  if(in->opcode >= FUSE_MAXOPS)
    goto reply_err;
  if(fuse_ll_funcs[in->opcode] == NULL)
    goto reply_err;

  fuse_ll_funcs[in->opcode](req, in);

  return;

 reply_err:
  fuse_reply_err(req, err);
  return;
}

static
void
fuse_ll_buf_process_read_init(struct fuse_session *se_,
                              int                  fd_,
                              const fuse_msgbuf_t *msgbuf_)
{
  int err;
  fuse_req_t *req;
  struct fuse_in_header *in;

  in = (struct fuse_in_header*)msgbuf_->mem;

  if(fuse_cfg.debug)
    fuse_debug_in_header(in);

  req = fuse_req_alloc();
  if(req == NULL)
    return fuse_send_enomem(fd_,in->unique);

  req->ctx.len    = in->len;
  req->ctx.opcode = in->opcode;
  req->ctx.unique = in->unique;
  req->ctx.nodeid = in->nodeid;
  req->ioctl_64bit = 0;
  req->ctx.uid    = in->uid;
  req->ctx.gid    = in->gid;
  req->ctx.pid    = in->pid;
  req->ctx.umask  = 0;
  req->se         = se_;
  req->fd         = fd_;

  err = EIO;
  if(in->opcode != FUSE_INIT)
    goto reply_err;
  if(fuse_ll_funcs[FUSE_INIT] == NULL)
    goto reply_err;

  se_->process_buf = fuse_ll_buf_process_read;

  fuse_ll_funcs[in->opcode](req,in);

  return;

 reply_err:
  fuse_reply_err(req, err);
  return;
}

/*
 * always call fuse_lowlevel_new_common() internally, to work around a
 * misfeature in the FreeBSD runtime linker, which links the old
 * version of a symbol to internal references.
 */
struct fuse_session *
fuse_lowlevel_new_common(struct fuse_args               *args,
                         const struct fuse_lowlevel_ops *op,
                         size_t                          op_size)
{
  struct fuse_session *se;

  if(sizeof(struct fuse_lowlevel_ops) < op_size)
    {
      fprintf(stderr, "fuse: warning: library too old, some operations may not work\n");
      op_size = sizeof(struct fuse_lowlevel_ops);
    }

  list_init_nreq(&f.notify_list);
  f.notify_ctr = 1;
  mutex_init(f.lock);

  memcpy(&f.op,op,op_size);
  f.owner = getuid();

#if defined(__linux__)
  _read_pipe_max_size();
#endif

  se = fuse_session_new(&f,
                        (void*)fuse_ll_buf_receive_read,
                        (void*)fuse_ll_buf_process_read_init,
                        (void*)fuse_ll_destroy);

  if(!se)
    goto out_free;

  return se;

 out_free:
  mutex_destroy(f.lock);

  return nullptr;
}

struct fuse_session*
fuse_lowlevel_new(struct fuse_args               *args,
                  const struct fuse_lowlevel_ops *op,
                  size_t                          op_size)
{
  return fuse_lowlevel_new_common(args, op, op_size);
}