RGW - Add POSIX Driver

This is the MVP for a driver for RGW that operates on top of a POSIX
filesystem.  It supports get, put, list, copy, multipart, external
access via the filesystem itself, and ordered bucket listings via an
LRU-based cache.

Note that this is currently a Filter, indended to run on top of dbstore.
This is because it currently doesn't have any User implementation, so it
depends on dbstore's User.  Everything else is implemented in
POSIXDriver.  Once there is a User implementation, this will become a
Store, instead of a Filter.

Commit messages from bucket listing cache:

  rgw/posixdriver: recycle lmdb database handles as required

    While LMDB workflows often do not close/return database handles,
    ours continually reuses them.  This requires us to close each
    handle (atomically) when a cache entry is recycled.

  rgw/posixdriver: don't instantiate bucket cache entries from notify events

  rgw/posixdriver: incorporate lmdb-safe for now

    The current inclusion is based on https://github.com/Martchus/lmdb-safe,
    which is actively maintained but currently has some packaging issues the
    author has agreed to accept fixes for.

    For now, skip the submodule to save time and remove an external dependency.

  rgw/posixdriver: fix listing of cached, empty bucket

    * check lmdb enumeration result in all cases and w/better style
    * add unit test for enumeration of an empty cached directory

  rgw/posixdriver: nest lmdbs in a directory under the dbroot path to avoid cleanup issues

  rgw/posixdriver: refactor for posix integration

    * Derive BucketCache types as templates on a SAL driver and SAL
      bucket pair.

    * Integrate cache fills as callbacks into SAL layer (or mock, for
      tests)

    * Renaming and cleanups

  rgw/posixdriver: add bucket cache implementation and tests

    Adds free-standing cache of buckets and object names, with
    bucket names (and listing attributes, upcoming) managed in
    a hashed set of lmdb databases, which provides ordering and
    a high-performance listing cache.

    An framework for notification on new object creation (e.g.,
    outside S3 workflow) is provided, and a Linux implementation
    using inotify.

    FindLMDB.cmake taken with attribution and license.

  rgw/posixdriver: add zpp_bits serialization (FAST)

Signed-off-by: Daniel Gryniewicz <dang@redhat.com>
Signed-off-by: Ali Maredia <amaredia@redhat.com>
Signed-off-by: Matt Benjamin <mbenjamin@redhat.com>

CMakeLists.txt

@@ -450,6 +450,7 @@ option(WITH_RADOSGW_DBSTORE "DBStore backend for RADOS Gateway" ON)
 option(WITH_RADOSGW_MOTR "CORTX-Motr backend for RADOS Gateway" OFF)
 option(WITH_RADOSGW_DAOS "DAOS backend for RADOS Gateway" OFF)
 option(WITH_RADOSGW_D4N "D4N wrapper for RADOS Gateway" ON)
+option(WITH_RADOSGW_POSIX "POSIX backend for Rados Gateway" ON)
 option(WITH_RADOSGW_SELECT_PARQUET "Support for s3 select on parquet objects" ON)
 option(WITH_RADOSGW_ARROW_FLIGHT "Build arrow flight when not using system-provided arrow" OFF)
 option(WITH_RADOSGW_BACKTRACE_LOGGING "Enable backtraces in rgw logs" OFF)

ceph.spec.in

@@ -270,6 +270,7 @@ BuildRequires:	xfsprogs-devel
 BuildRequires:	xmlstarlet
 BuildRequires:	nasm
 BuildRequires:	lua-devel
+BuildRequires:  lmdb-devel
 %if 0%{with seastar} || 0%{with jaeger}
 BuildRequires:  yaml-cpp-devel >= 0.6
 %endif

cmake/modules/FindLMDB.cmake

@@ -0,0 +1,61 @@
+# Copyright (c) 2014, The Monero Project
+# All rights reserved.
+# 
+# Redistribution and use in source and binary forms, with or without modification, are
+# permitted provided that the following conditions are met:
+# 
+# 1. Redistributions of source code must retain the above copyright notice, this list of
+#    conditions and the following disclaimer.
+# 
+# 2. Redistributions in binary form must reproduce the above copyright notice, this list
+#    of conditions and the following disclaimer in the documentation and/or other
+#    materials provided with the distribution.
+# 
+# 3. Neither the name of the copyright holder nor the names of its contributors may be
+#    used to endorse or promote products derived from this software without specific
+#    prior written permission.
+# 
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
+# MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
+# THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
+# STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
+# THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+MESSAGE(STATUS "Looking for liblmdb")
+
+FIND_PATH(LMDB_INCLUDE_DIR
+  NAMES lmdb.h
+  PATH_SUFFIXES include/ include/lmdb/
+  PATHS "${PROJECT_SOURCE_DIR}"
+  ${LMDB_ROOT}
+  $ENV{LMDB_ROOT}
+  /usr/local/
+  /usr/
+)
+
+if(STATIC)
+  if(MINGW)
+    find_library(LMDB_LIBRARIES liblmdb.dll.a)
+  else()
+    find_library(LMDB_LIBRARIES liblmdb.a)
+  endif()
+else()
+  find_library(LMDB_LIBRARIES lmdb)
+endif()
+
+IF(LMDB_INCLUDE_DIR)
+  MESSAGE(STATUS "Found liblmdb include (lmdb.h) in ${LMDB_INCLUDE_DIR}")
+  IF(LMDB_LIBRARIES)
+    MESSAGE(STATUS "Found liblmdb library")
+    set(LMDB_INCLUDE ${LMDB_INCLUDE_DIR})
+    set(LMDB_LIBRARY ${LMDB_LIBRARIES})
+  ELSE()
+    MESSAGE(FATAL_ERROR "${BoldRed}Could not find liblmdb library, please make sure you have installed liblmdb or liblmdb-dev or the equivalent${ColourReset}")
+  ENDIF()
+ELSE()
+  MESSAGE(FATAL_ERROR "${BoldRed}Could not find liblmdb library, please make sure you have installed liblmdb or liblmdb-dev or the equivalent${ColourReset}")
+ENDIF()

debian/control

@@ -65,6 +65,7 @@ Build-Depends: automake,
                libsqlite3-dev,
                libssl-dev,
                libtool,
+               liblmdb-dev,
                libudev-dev,
                libnl-genl-3-dev,
                libxml2-dev,

src/common/cohort_lru.h

@@ -16,6 +16,12 @@
 #include <boost/intrusive/list.hpp>
 #include <boost/intrusive/slist.hpp>
 
+#ifdef __CEPH__
+# include "include/ceph_assert.h"
+#else
+# include <assert.h>
+#endif
+
 #include "common/likely.h"
 
 #ifndef CACHE_LINE_SIZE

src/common/options/rgw.yaml.in

@@ -3618,6 +3618,7 @@ options:
   - none
   - base
   - d4n
+  - posix
 - name: dbstore_db_dir
   type: str
   level: advanced
@@ -3705,6 +3706,51 @@ options:
   default: false
   services:
   - rgw
+- name: rgw_posix_base_path
+  type: str
+  level: advanced
+  desc: experimental Option to set base path for POSIX Driver
+  long_desc: Base path for the POSIX driver.  All operations are relative to this path.
+    Defaults to /tmp/rgw_posix_driver
+  default: /tmp/rgw_posix_driver
+  services:
+  - rgw
+- name: rgw_posix_database_root
+  type: str
+  level: advanced
+  desc: experimental Path to parent of POSIX Driver LMDB bucket listing cache
+  long_desc: Parent directory of LMDB bucket listing cache databases.
+  default: /var/lib/ceph/radosgw
+  services:
+  - rgw
+- name: rgw_posix_cache_max_buckets
+  type: int
+  level: advanced
+  desc: experimental Number of buckets to maintain in the ordered listing cache
+  default: 100
+  services:
+  - rgw
+- name: rgw_posix_cache_lanes
+  type: int
+  level: advanced
+  desc: experimental Number of lanes in cache LRU
+  default: 3
+  services:
+  - rgw
+- name: rgw_posix_cache_partitions
+  type: int
+  level: advanced
+  desc: experimental Number of partitions in cache AVL
+  default: 3
+  services:
+  - rgw
+- name: rgw_posix_cache_lmdb_count
+  type: int
+  level: advanced
+  desc: experimental Number of lmdb partitions in the ordered listing cache
+  default: 3
+  services:
+  - rgw
 - name: rgw_luarocks_location
   type: str
   level: advanced

src/include/config-h.in.cmake

@@ -363,6 +363,9 @@
 /* Backend CORTX-DAOS for Rados Gateway */
 #cmakedefine WITH_RADOSGW_DAOS
 
+/* Backend POSIX for Rados Gateway */
+#cmakedefine WITH_RADOSGW_POSIX
+
 /* Defined if std::map::merge() is supported */
 #cmakedefine HAVE_STDLIB_MAP_SPLICING
 

src/rgw/CMakeLists.txt

@@ -227,6 +227,15 @@ if(WITH_RADOSGW_D4N)
   list(APPEND librgw_common_srcs driver/d4n/d4n_datacache.cc)
   list(APPEND librgw_common_srcs driver/d4n/rgw_sal_d4n.cc)
 endif()
+if(WITH_RADOSGW_POSIX)
+  #add_subdirectory(driver/posix)
+  find_package(LMDB REQUIRED)
+  add_compile_definitions(LMDB_SAFE_NO_CPP_UTILITIES)
+  list(APPEND librgw_common_srcs
+	      driver/posix/rgw_sal_posix.cc
+	      driver/posix/lmdb-safe.cc
+	      driver/posix/notify.cpp)
+endif()
 if(WITH_JAEGER)
   list(APPEND librgw_common_srcs rgw_tracer.cc)
 endif()
@@ -236,7 +245,6 @@ if(WITH_RADOSGW_ARROW_FLIGHT)
   list(APPEND librgw_common_srcs rgw_flight.cc rgw_flight_frontend.cc)
 endif(WITH_RADOSGW_ARROW_FLIGHT)
 
-
 add_library(rgw_common STATIC ${librgw_common_srcs})
 
 include(CheckCXXCompilerFlag)
@@ -270,6 +278,7 @@ target_link_libraries(rgw_common
     ${EXPAT_LIBRARIES}
     ${ARROW_LIBRARIES}
     ${ARROW_FLIGHT_LIBRARIES}
+    ${LMDB_LIBRARIES}
     ${ALLOC_LIBS}
   PUBLIC
     ${LUA_LIBRARIES}

src/rgw/driver/posix/README.md

@@ -0,0 +1,37 @@
+# POSIX Driver
+Standalone Rados Gateway (RGW) on a local POSIX filesystem (Experimental)
+
+
+## CMake Option
+Add below cmake option (enabled by default)
+
+    -DWITH_RADOSGW_POSIX=ON 
+
+
+## Build
+
+    cd build
+    ninja [vstart]
+
+
+## Running Test cluster
+Currently, POSIXDriver depends on DBStore for user storage.  This will change, eventually, but for now, it's run as a filter on top of DBStore.  Not that only users are stored in DBStore, the rest is in the POSIX filesystem.
+Edit ceph.conf to add below option
+
+    [client]
+        rgw backend store = dbstore
+        rgw config store = dbstore
+        rgw filter = posix
+
+Start vstart cluster
+
+    MON=0 OSD=0 MDS=0 MGR=0 RGW=1 ../src/vstart.sh -o rgw_backend_store=dbstore -o rgw_config_store=dbstore -o rgw_filter=posix -n -d
+
+The above vstart command brings up RGW server on POSIXDriver. It creates default zonegroup, zone and few default users (eg., testid) to be used for s3 operations.
+
+`radosgw-admin` can be used to create and remove other users, zonegroups and zones.
+
+By default, the directory exported is *'/tmp/rgw_posix_driver'*.   This can be changed with the `rgw_posix_base_path` option, either in ceph.conf or on the vstart command line above.
+
+The POSIXDriver keeps a LMDB based cache of directories, so that it can provide ordered listings.  This directory lives in `rgw_posix_database_root`, which by default is in *'/var/lib/ceph/radosgw'*
+

src/rgw/driver/posix/bucket_cache.cpp

@@ -0,0 +1,4 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+#include "bucket_cache.h"
+

src/rgw/driver/posix/bucket_cache.h

@@ -0,0 +1,549 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#pragma once
+
+#include <iostream>
+#include <lmdb.h>
+#include <memory>
+#include <tuple>
+#include <vector>
+#include <string>
+#include <string_view>
+#include <mutex>
+#include <shared_mutex>
+#include <condition_variable>
+#include <filesystem>
+#include <boost/intrusive/avl_set.hpp>
+#include "include/function2.hpp"
+#include "common/cohort_lru.h"
+#include "lmdb-safe.hh"
+#include "zpp_bits.h"
+#include "notify.h"
+#include <stdint.h>
+#include <time.h> // struct timespec
+#include <xxhash.h>
+
+#include "rgw_common.h"
+#include "rgw_sal.h"
+
+#include "fmt/format.h"
+
+namespace file::listing {
+
+namespace bi = boost::intrusive;
+namespace sf = std::filesystem; 
+
+typedef bi::link_mode<bi::safe_link> link_mode; /* XXX normal */
+typedef bi::avl_set_member_hook<link_mode> member_hook_t;
+
+template <typename D, typename B>
+struct BucketCache;
+
+template <typename D, typename B>
+struct BucketCacheEntry : public cohort::lru::Object
+{
+  using lock_guard = std::lock_guard<std::mutex>;
+  using unique_lock = std::unique_lock<std::mutex>;
+
+  static constexpr uint32_t FLAG_NONE     = 0x0000;
+  static constexpr uint32_t FLAG_FILLED   = 0x0001;
+  static constexpr uint32_t FLAG_DELETED  = 0x0002;
+  
+  static constexpr uint64_t seed = 8675309;
+
+  BucketCache<D, B>* bc;
+  std::string name;
+  std::shared_ptr<LMDBSafe::MDBEnv> env;
+  LMDBSafe::MDBDbi dbi;
+  uint64_t hk;
+  member_hook_t name_hook;
+
+  // XXX clean this up
+  std::mutex mtx; // XXX Adam's preferred shared mtx?
+  std::condition_variable cv;
+  uint32_t flags;
+
+public:
+  BucketCacheEntry(BucketCache<D, B>* bc, const std::string& name, uint64_t hk)
+    : bc(bc), name(name), hk(hk), flags(FLAG_NONE) {}
+
+  void set_env(std::shared_ptr<LMDBSafe::MDBEnv>& _env, LMDBSafe::MDBDbi& _dbi) {
+    env = _env;
+    dbi = _dbi;
+  }
+
+  inline bool deleted() const {
+    return flags & FLAG_DELETED;
+  }
+
+  class Factory : public cohort::lru::ObjectFactory
+  {
+  public:
+    BucketCache<D, B>* bc;
+    const std::string& name;
+    uint64_t hk;
+    uint32_t flags;
+
+    Factory() = delete;
+    Factory(BucketCache<D, B> *bc, const std::string& name)
+      : bc(bc), name(name), flags(FLAG_NONE) {
+      hk = XXH64(name.c_str(), name.length(), BucketCacheEntry::seed);
+    }
+
+    void recycle (cohort::lru::Object* o) override {
+      /* re-use an existing object */
+      o->~Object(); // call lru::Object virtual dtor
+      // placement new!
+      new (o) BucketCacheEntry(bc, name, hk);
+    }
+
+    cohort::lru::Object* alloc() override {
+      return new BucketCacheEntry(bc, name, hk);
+    }
+  }; /* Factory */
+
+  struct BucketCacheEntryLT
+  {
+    // for internal ordering
+    bool operator()(const BucketCacheEntry& lhs, const BucketCacheEntry& rhs) const
+      { return (lhs.name < rhs.name); }
+
+    // for external search by name
+    bool operator()(const std::string& k, const BucketCacheEntry& rhs) const
+      { return k < rhs.name; }
+
+    bool operator()(const BucketCacheEntry& lhs, const std::string& k) const
+      { return lhs.name < k; }
+  };
+
+  struct BucketCacheEntryEQ
+  {
+    bool operator()(const BucketCacheEntry& lhs, const BucketCacheEntry& rhs) const
+      { return (lhs.name == rhs.name); }
+
+    bool operator()(const std::string& k, const BucketCacheEntry& rhs) const
+      { return k == rhs.name; }
+
+    bool operator()(const BucketCacheEntry& lhs, const std::string& k) const
+      { return lhs.name == k; }
+  };
+
+  typedef cohort::lru::LRU<std::mutex> bucket_lru;
+
+  typedef bi::member_hook<BucketCacheEntry, member_hook_t, &BucketCacheEntry::name_hook> name_hook_t;
+  typedef bi::avltree<BucketCacheEntry, bi::compare<BucketCacheEntryLT>, name_hook_t> bucket_avl_t;
+  typedef cohort::lru::TreeX<BucketCacheEntry, bucket_avl_t, BucketCacheEntryLT, BucketCacheEntryEQ, std::string,
+			     std::mutex> bucket_avl_cache;
+
+  bool reclaim(const cohort::lru::ObjectFactory* newobj_fac) {
+    auto factory = dynamic_cast<const BucketCacheEntry<D, B>::Factory*>(newobj_fac);
+    if (factory == nullptr) {
+        return false;
+    }
+    { /* anon block */
+      /* in this case, we are being called from a context which holds
+       * A partition lock, and this may be still in use */
+      lock_guard{mtx};
+      if (! deleted()) {
+	flags |= FLAG_DELETED;
+	bc->recycle_count++;
+
+	//std::cout << fmt::format("reclaim {}!", name) << std::endl;
+	bc->un->remove_watch(name);
+#if 1
+	// depends on safe_link
+	if (! name_hook.is_linked()) {
+	  // this should not happen!
+	  abort();
+	}
+#endif
+	bc->cache.remove(hk, this, bucket_avl_cache::FLAG_NONE);
+
+	/* discard lmdb data associated with this bucket */
+	auto txn = env->getRWTransaction();
+	mdb_drop(*txn, dbi, 0);
+	txn->commit();
+	/* LMDB applications don't "normally" close database handles,
+	 * but doing so (atomically) is supported, and we must as
+	 * we continually recycle them */
+	mdb_dbi_close(*env, dbi); // return db handle
+      } /* ! deleted */
+    }
+    return true;
+} /* reclaim */
+
+}; /* BucketCacheEntry */
+
+using fill_cache_cb_t =
+  const fu2::unique_function<int(const DoutPrefixProvider* dpp,
+    rgw_bucket_dir_entry&) const>;
+
+using list_bucket_each_t =
+  const fu2::unique_function<bool(const rgw_bucket_dir_entry&) const>;
+
+template <typename D, typename B>
+struct BucketCache : public Notifiable
+{
+  using lock_guard = std::lock_guard<std::mutex>;
+  using unique_lock = std::unique_lock<std::mutex>;
+
+  D* driver;
+  std::string bucket_root;
+  uint32_t max_buckets;
+  std::atomic<uint64_t> recycle_count;
+  std::mutex mtx;
+
+  /* the bucket lru cache keeps track of the buckets whose listings are
+   * being cached in lmdb databases and updated from notify */
+  typename BucketCacheEntry<D, B>::bucket_lru lru;
+  typename BucketCacheEntry<D, B>::bucket_avl_cache cache;
+  sf::path rp;
+
+  /* the lmdb handle cache maintains a vector of lmdb environments,
+   * each supports 1 rw and unlimited ro transactions;  the materialized
+   * listing for each bucket is stored as a database in one of these
+   * environments, selected by a hash of the bucket name; a bucket's database
+   * is dropped/cleared whenever its entry is reclaimed from cache; the entire
+   * complex is cleared on restart to preserve consistency */
+  class Lmdbs
+  {
+    std::string database_root;
+    uint8_t lmdb_count;
+    std::vector<std::shared_ptr<LMDBSafe::MDBEnv>> envs;
+    sf::path dbp;
+
+  public:
+    Lmdbs(std::string& database_root, uint8_t lmdb_count)
+      : database_root(database_root), lmdb_count(lmdb_count),
+        dbp(database_root) {
+
+      /* create a root for lmdb directory partitions (if it doesn't
+       * exist already) */
+      sf::path safe_root_path{dbp / fmt::format("rgw_posix_lmdbs")};
+      sf::create_directory(safe_root_path);
+
+      /* purge cache completely */
+      for (const auto& dir_entry : sf::directory_iterator{safe_root_path}) {
+	sf::remove_all(dir_entry);
+      }
+
+      /* repopulate cache basis */
+      for (int ix = 0; ix < lmdb_count; ++ix) {
+	sf::path env_path{safe_root_path / fmt::format("part_{}", ix)};
+	sf::create_directory(env_path);
+	auto env = LMDBSafe::getMDBEnv(env_path.string().c_str(), 0 /* flags? */, 0600);
+	envs.push_back(env);
+      }
+    }
+
+    inline std::shared_ptr<LMDBSafe::MDBEnv>& get_sp_env(BucketCacheEntry<D, B>* bucket)  {
+      return envs[(bucket->hk % lmdb_count)];
+    }
+
+    inline LMDBSafe::MDBEnv& get_env(BucketCacheEntry<D, B>* bucket) {
+      return *(get_sp_env(bucket));
+    }
+
+    const std::string& get_root() const { return database_root; }
+  } lmdbs;
+
+  std::unique_ptr<Notify> un;
+
+public:
+  BucketCache(D* driver, std::string bucket_root, std::string database_root,
+	      uint32_t max_buckets=100, uint8_t max_lanes=3,
+	      uint8_t max_partitions=3, uint8_t lmdb_count=3)
+    : driver(driver), bucket_root(bucket_root), max_buckets(max_buckets),
+      lru(max_lanes, max_buckets/max_lanes),
+      cache(max_lanes, max_buckets/max_partitions),
+      rp(bucket_root),
+      lmdbs(database_root, lmdb_count),
+      un(Notify::factory(this, bucket_root))
+    {
+      if (! (sf::exists(rp) && sf::is_directory(rp))) {
+	std::cerr << fmt::format("{} bucket root {} invalid", __func__,
+				 bucket_root) << std::endl;
+	exit(1);
+      }
+
+      sf::path dp{database_root};
+      if (! (sf::exists(dp) && sf::is_directory(dp))) {
+	std::cerr << fmt::format("{} database root {} invalid", __func__,
+				 database_root) << std::endl;
+	exit(1);
+      }
+    }
+
+  static constexpr uint32_t FLAG_NONE     = 0x0000;
+  static constexpr uint32_t FLAG_CREATE   = 0x0001;
+  static constexpr uint32_t FLAG_LOCK     = 0x0002;
+
+  typedef std::tuple<BucketCacheEntry<D, B>*, uint32_t> GetBucketResult;
+
+  GetBucketResult get_bucket(const std::string& name, uint32_t flags)
+    {
+      /* this fn returns a bucket locked appropriately, having atomically
+       * found or inserted the required BucketCacheEntry in_avl*/
+      BucketCacheEntry<D, B>* b{nullptr};
+      typename BucketCacheEntry<D, B>::Factory fac(this, name);
+      typename BucketCacheEntry<D, B>::bucket_avl_cache::Latch lat;
+      uint32_t iflags{cohort::lru::FLAG_INITIAL};
+      GetBucketResult result{nullptr, 0};
+
+    retry:
+      b = cache.find_latch(fac.hk /* partition selector */,
+			   name /* key */, lat /* serializer */, BucketCacheEntry<D, B>::bucket_avl_cache::FLAG_LOCK);
+      /* LATCHED */
+      if (b) {
+	b->mtx.lock();
+	if (b->deleted() ||
+	    ! lru.ref(b, cohort::lru::FLAG_INITIAL)) {
+	  // lru ref failed
+	  lat.lock->unlock();
+	  b->mtx.unlock();
+	  goto retry;
+	}
+	lat.lock->unlock();
+	/* LOCKED */
+      } else {
+	/* BucketCacheEntry not in cache */
+	if (! (flags & BucketCache<D, B>::FLAG_CREATE)) {
+	  /* the caller does not want to instantiate a new cache
+	   * entry (i.e., only wants to notify on an existing one) */
+	  return result;
+	}
+	/* we need to create it */
+	b = static_cast<BucketCacheEntry<D, B>*>(
+	  lru.insert(&fac, cohort::lru::Edge::MRU, iflags));
+	if (b) [[likely]] {
+	  b->mtx.lock();
+
+	  /* attach bucket to an lmdb partition and prepare it for i/o */
+	  auto& env = lmdbs.get_sp_env(b);
+	  auto dbi = env->openDB(b->name, MDB_CREATE);
+	  b->set_env(env, dbi);
+
+	  if (! (iflags & cohort::lru::FLAG_RECYCLE)) [[likely]] {
+	    /* inserts at cached insert iterator, releasing latch */
+	    cache.insert_latched(b, lat, BucketCacheEntry<D, B>::bucket_avl_cache::FLAG_UNLOCK);
+	  } else {
+	    /* recycle step invalidates Latch */
+	    lat.lock->unlock(); /* !LATCHED */
+	    cache.insert(fac.hk, b, BucketCacheEntry<D, B>::bucket_avl_cache::FLAG_NONE);
+	  }
+	  get<1>(result) |= BucketCache<D, B>::FLAG_CREATE;
+	} else {
+	  /* XXX lru allocate failed? seems impossible--that would mean that
+	   * fallback to the allocator also failed, and maybe we should abend */
+	  lat.lock->unlock();
+	  goto retry; /* !LATCHED */
+	}
+      } /* have BucketCacheEntry */
+
+      if (! (flags & BucketCache<D, B>::FLAG_LOCK)) {
+	b->mtx.unlock();
+      }
+      get<0>(result) = b;
+      return result;
+    } /* get_bucket */
+
+  static inline std::string concat_key(const rgw_obj_index_key& k) {
+    std::string k_str;
+    k_str.reserve(k.name.size() + k.instance.size());
+    k_str += k.name;
+    k_str += k.instance;
+    return k_str;
+  }
+
+  int fill(const DoutPrefixProvider* dpp, BucketCacheEntry<D, B>* bucket,
+	    B* sal_bucket, uint32_t flags, optional_yield y) /* assert: LOCKED */
+  {
+      auto txn = bucket->env->getRWTransaction();
+
+      /* instruct the bucket provider to enumerate all entries,
+       * in any order */
+      auto rc = sal_bucket->fill_cache(dpp, y,
+	[&](const DoutPrefixProvider* dpp, rgw_bucket_dir_entry& bde) -> int {
+	  auto concat_k = concat_key(bde.key);
+	  std::string ser_data;
+	  zpp::bits::out out(ser_data);
+	  struct timespec ts{ceph::real_clock::to_timespec(bde.meta.mtime)};
+	  auto errc =
+	    out(bde.key.name, bde.key.instance, /* XXX bde.key.ns, */
+		bde.ver.pool, bde.ver.epoch, bde.exists,
+		bde.meta.category, bde.meta.size, ts.tv_sec, ts.tv_nsec,
+		bde.meta.owner, bde.meta.owner_display_name, bde.meta.accounted_size,
+		bde.meta.storage_class, bde.meta.appendable, bde.meta.etag);
+	  /*std::cout << fmt::format("fill: bde.key.name: {}", bde.key.name)
+	    << std::endl;*/
+	  if (errc.code != std::errc{0}) {
+	    abort();
+	    return 0; // XXX non-zero return?
+	  }
+	  txn->put(bucket->dbi, concat_k, ser_data);
+	  //std::cout << fmt::format("{} {}", __func__, bde.key.name) << '\n';
+	  return 0;
+	});
+
+      txn->commit();
+      bucket->flags |= BucketCacheEntry<D, B>::FLAG_FILLED;
+      un->add_watch(bucket->name, bucket);
+      return rc;
+    } /* fill */
+
+  int list_bucket(const DoutPrefixProvider* dpp, optional_yield y, B* sal_bucket,
+		  std::string marker, list_bucket_each_t each_func) {
+
+    using namespace LMDBSafe;
+
+    int rc __attribute__((unused)) = 0;
+    GetBucketResult gbr =
+      get_bucket(sal_bucket->get_name(),
+		 BucketCache<D, B>::FLAG_LOCK | BucketCache<D, B>::FLAG_CREATE);
+    auto [b /* BucketCacheEntry */, flags] = gbr;
+    if (b /* XXX again, can this fail? */) {
+      if (! (b->flags & BucketCacheEntry<D, B>::FLAG_FILLED)) {
+	/* bulk load into lmdb cache */
+	rc = fill(dpp, b, sal_bucket, FLAG_NONE, y);
+      }
+      /* display them */
+      b->mtx.unlock();
+      /*! LOCKED */
+
+      auto txn = b->env->getROTransaction();
+      auto cursor=txn->getCursor(b->dbi);
+      MDBOutVal key, data;
+      bool again{true};
+
+      const auto proc_result = [&]() {
+	zpp::bits::errc errc{};
+	rgw_bucket_dir_entry bde{};
+	/* XXX we may not need to recover the cache key */
+	std::string_view svk __attribute__((unused)) =
+	  key.get<string_view>(); // {name, instance, [ns]}
+	std::string_view svv = data.get<string_view>();
+	std::string ser_v{svv};
+	zpp::bits::in in_v(ser_v);
+	struct timespec ts;
+	errc =
+	  in_v(bde.key.name, bde.key.instance, /* bde.key.ns, */
+	       bde.ver.pool, bde.ver.epoch, bde.exists,
+	       bde.meta.category, bde.meta.size, ts.tv_sec, ts.tv_nsec,
+	       bde.meta.owner, bde.meta.owner_display_name, bde.meta.accounted_size,
+	       bde.meta.storage_class, bde.meta.appendable, bde.meta.etag);
+	if (errc.code != std::errc{0}) {
+	  abort();
+	}
+	bde.meta.mtime = ceph::real_clock::from_timespec(ts);
+	again = each_func(bde);
+      };
+
+      if (! marker.empty()) {
+	MDBInVal k(marker);
+	auto rc = cursor.lower_bound(k, key, data);
+	if (rc == MDB_NOTFOUND) {
+	  /* no key sorts after k/marker, so there is nothing to do */
+	  return 0;
+	}
+	proc_result();
+      } else {
+	/* position at start of index */
+	auto rc = cursor.get(key, data, MDB_FIRST);
+	if (rc == MDB_SUCCESS) {
+	  proc_result();
+	}
+      }
+      while(cursor.get(key, data, MDB_NEXT) == MDB_SUCCESS) {
+	if (!again) {
+	  return 0;
+	}
+	proc_result();
+      }
+      lru.unref(b, cohort::lru::FLAG_NONE);
+    } /* b */
+
+    return 0;
+  } /* list_bucket */
+
+  int notify(const std::string& bname, void* opaque,
+	     const std::vector<Notifiable::Event>& evec) override {
+
+    using namespace LMDBSafe;
+
+    int rc{0};
+    GetBucketResult gbr = get_bucket(bname, BucketCache<D, B>::FLAG_LOCK);
+    auto [b /* BucketCacheEntry */, flags] = gbr;
+    if (b) {
+      unique_lock ulk{b->mtx, std::adopt_lock};
+      if ((b->name != bname) ||
+	  (b != opaque) ||
+	  (! (b->flags & BucketCacheEntry<D, B>::FLAG_FILLED))) {
+	/* do nothing */
+	return 0;
+      }
+      ulk.unlock();
+      auto txn = b->env->getRWTransaction();
+      for (const auto& ev : evec) {
+	using EventType = Notifiable::EventType;
+	/*
+	std::string_view nil{""};
+	std::cout << fmt::format("notify {} {}!",
+				 ev.name ? *ev.name : nil,
+				 uint32_t(ev.type))
+				 << std::endl; */
+	switch (ev.type)
+	{
+	case EventType::ADD:
+	{
+	  rgw_bucket_dir_entry bde{};
+	  bde.key.name = *ev.name;
+	  /* XXX will need work (if not straight up magic) to have
+	   * side loading support instance and ns */
+	  auto concat_k = concat_key(bde.key);
+	  rc = driver->mint_listing_entry(b->name, bde);
+	  std::string ser_data;
+	  zpp::bits::out out(ser_data);
+	  struct timespec ts{ceph::real_clock::to_timespec(bde.meta.mtime)};
+	  auto errc =
+	    out(bde.key.name, bde.key.instance, /* XXX bde.key.ns, */
+		bde.ver.pool, bde.ver.epoch, bde.exists,
+		bde.meta.category, bde.meta.size, ts.tv_sec, ts.tv_nsec,
+		bde.meta.owner, bde.meta.owner_display_name, bde.meta.accounted_size,
+		bde.meta.storage_class, bde.meta.appendable, bde.meta.etag);
+	  if (errc.code != std::errc{0}) {
+	    abort();
+	  }
+	  txn->put(b->dbi, concat_k, ser_data);
+	}
+	  break;
+	case EventType::REMOVE:
+	{
+	  auto& ev_name = *ev.name;
+	  txn->del(b->dbi, ev_name);
+	}
+	  break;
+	[[unlikely]] case EventType::INVALIDATE:
+	{
+	  /* yikes, cache blown */
+	  ulk.lock();
+	  mdb_drop(*txn, b->dbi, 0);
+	  txn->commit();
+	  b->flags &= ~BucketCacheEntry<D, B>::FLAG_FILLED;
+	  return 0; /* don't process any more events in this batch */
+	}
+	  break;
+	default:
+	  /* unknown event */
+	  break;
+	}
+      } /* all events */
+      txn->commit();
+      lru.unref(b, cohort::lru::FLAG_NONE);
+    } /* b */
+    return rc;
+  } /* notify */
+  
+}; /* BucketCache */
+
+} // namespace file::listing

src/rgw/driver/posix/lmdb-safe-global.h

@@ -0,0 +1,36 @@
+/*
+MIT License
+
+Copyright (c) 2018 bert hubert
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+*/
+#ifndef LMDB_SAFE_GLOBAL
+#define LMDB_SAFE_GLOBAL
+
+#ifndef LMDB_SAFE_NO_CPP_UTILITIES
+#include <c++utilities/application/global.h>
+#else
+#undef LMDB_SAFE_STATIC
+#define LMDB_SAFE_STATIC 1
+#endif
+
+#ifdef LMDB_SAFE_STATIC
+#define LMDB_SAFE_EXPORT
+#define LMDB_SAFE_IMPORT
+#else
+#define LMDB_SAFE_EXPORT CPP_UTILITIES_GENERIC_LIB_EXPORT
+#define LMDB_SAFE_IMPORT CPP_UTILITIES_GENERIC_LIB_IMPORT
+#endif
+
+/*!
+ * \def LMDB_SAFE_EXPORT
+ * \brief Marks the symbol to be exported by the lmdb-safe library.
+ */
+
+/*!
+ * \def LMDB_SAFE_IMPORT
+ * \brief Marks the symbol to be imported from the lmdb-safe library.
+ */
+
+#endif // LMDB_SAFE_GLOBAL

src/rgw/driver/posix/lmdb-safe.cc

@@ -0,0 +1,369 @@
+/*
+MIT License
+
+Copyright (c) 2018 bert hubert
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+*/
+#include "lmdb-safe.hh"
+
+#include <fcntl.h>
+#include <sys/stat.h>
+
+#include <cstring>
+#include <map>
+#include <memory>
+#include <mutex>
+
+using namespace std;
+
+namespace LMDBSafe {
+
+MDBDbi::MDBDbi(MDB_env *env, MDB_txn *txn, const string_view dbname, unsigned int flags)
+{
+    (void)env;
+    // A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
+
+    if (const auto rc = mdb_dbi_open(txn, dbname.empty() ? 0 : &dbname[0], flags, &d_dbi))
+        throw LMDBError("Unable to open named database: ", rc);
+
+    // Database names are keys in the unnamed database, and may be read but not written.
+}
+
+MDBEnv::MDBEnv(const char *fname, unsigned int flags, mdb_mode_t mode, MDB_dbi maxDBs)
+{
+    mdb_env_create(&d_env);
+    if (const auto rc = mdb_env_set_mapsize(d_env, 16ULL * 4096 * 244140ULL)) { // 4GB
+        throw LMDBError("Setting map size: ", rc);
+    }
+    // Various other options may also need to be set before opening the handle, e.g. mdb_env_set_mapsize(), mdb_env_set_maxreaders(), mdb_env_set_maxdbs(),
+    if (const auto rc = mdb_env_set_maxdbs(d_env, maxDBs)) {
+        throw LMDBError("Setting maxdbs: ", rc);
+    }
+
+    // we need MDB_NOTLS since we rely on its semantics
+    if (const auto rc = mdb_env_open(d_env, fname, flags | MDB_NOTLS, mode)) {
+        // If this function fails, mdb_env_close() must be called to discard the MDB_env handle.
+        mdb_env_close(d_env);
+        throw LMDBError("Unable to open database file " + std::string(fname) + ": ", rc);
+    }
+}
+
+void MDBEnv::incROTX()
+{
+    std::lock_guard<std::mutex> l(d_countmutex);
+    ++d_ROtransactionsOut[std::this_thread::get_id()];
+}
+
+void MDBEnv::decROTX()
+{
+    std::lock_guard<std::mutex> l(d_countmutex);
+    --d_ROtransactionsOut[std::this_thread::get_id()];
+}
+
+void MDBEnv::incRWTX()
+{
+    std::lock_guard<std::mutex> l(d_countmutex);
+    ++d_RWtransactionsOut[std::this_thread::get_id()];
+}
+
+void MDBEnv::decRWTX()
+{
+    std::lock_guard<std::mutex> l(d_countmutex);
+    --d_RWtransactionsOut[std::this_thread::get_id()];
+}
+
+int MDBEnv::getRWTX()
+{
+    std::lock_guard<std::mutex> l(d_countmutex);
+    return d_RWtransactionsOut[std::this_thread::get_id()];
+}
+int MDBEnv::getROTX()
+{
+    std::lock_guard<std::mutex> l(d_countmutex);
+    return d_ROtransactionsOut[std::this_thread::get_id()];
+}
+
+std::shared_ptr<MDBEnv> getMDBEnv(const char *fname, unsigned int flags, mdb_mode_t mode, MDB_dbi maxDBs)
+{
+    struct Value {
+        weak_ptr<MDBEnv> wp;
+        unsigned int flags;
+    };
+
+    static std::map<tuple<dev_t, ino_t>, Value> s_envs;
+    static std::mutex mut;
+
+    struct stat statbuf;
+    if (stat(fname, &statbuf)) {
+        if (errno != ENOENT)
+            throw LMDBError("Unable to stat prospective mdb database: " + string(strerror(errno)));
+        else {
+            std::lock_guard<std::mutex> l(mut);
+            auto fresh = std::make_shared<MDBEnv>(fname, flags, mode, maxDBs);
+            if (stat(fname, &statbuf))
+                throw LMDBError("Unable to stat prospective mdb database: " + string(strerror(errno)));
+            auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
+            s_envs[key] = { fresh, flags };
+            return fresh;
+        }
+    }
+
+    std::lock_guard<std::mutex> l(mut);
+    auto key = std::tie(statbuf.st_dev, statbuf.st_ino);
+    auto iter = s_envs.find(key);
+    if (iter != s_envs.end()) {
+        auto sp = iter->second.wp.lock();
+        if (sp) {
+            if (iter->second.flags != flags)
+                throw LMDBError("Can't open mdb with differing flags");
+
+            return sp;
+        } else {
+            s_envs.erase(iter); // useful if make_shared fails
+        }
+    }
+
+    auto fresh = std::make_shared<MDBEnv>(fname, flags, mode, maxDBs);
+    s_envs[key] = { fresh, flags };
+
+    return fresh;
+}
+
+MDBDbi MDBEnv::openDB(const string_view dbname, unsigned int flags)
+{
+    unsigned int envflags;
+    mdb_env_get_flags(d_env, &envflags);
+    /*
+    This function must not be called from multiple concurrent transactions in the same process. A transaction that uses this function must finish (either commit or abort) before any other transaction in the process may use this function.
+  */
+    std::lock_guard<std::mutex> l(d_openmut);
+
+    if (!(envflags & MDB_RDONLY)) {
+        auto rwt = getRWTransaction();
+        MDBDbi ret = rwt->openDB(dbname, flags);
+        rwt->commit();
+        return ret;
+    }
+
+    MDBDbi ret;
+    {
+        auto rwt = getROTransaction();
+        ret = rwt->openDB(dbname, flags);
+    }
+    return ret;
+}
+
+MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv *parent, MDB_txn *txn)
+    : MDBROTransactionImpl(parent, txn)
+
+{
+}
+
+MDB_txn *MDBRWTransactionImpl::openRWTransaction(MDBEnv *env, MDB_txn *parent, unsigned int flags)
+{
+    MDB_txn *result;
+    if (env->getRWTX())
+        throw LMDBError("Duplicate RW transaction");
+
+    for (int tries = 0; tries < 3; ++tries) { // it might happen twice, who knows
+        if (int rc = mdb_txn_begin(env->d_env, parent, flags, &result)) {
+            if (rc == MDB_MAP_RESIZED && tries < 2) {
+                // "If the mapsize is increased by another process (..) mdb_txn_begin() will return MDB_MAP_RESIZED.
+                // call mdb_env_set_mapsize with a size of zero to adopt the new size."
+                mdb_env_set_mapsize(env->d_env, 0);
+                continue;
+            }
+            throw LMDBError("Unable to start RW transaction: ", rc);
+        }
+        break;
+    }
+    env->incRWTX();
+    return result;
+}
+
+MDBRWTransactionImpl::MDBRWTransactionImpl(MDBEnv *parent, unsigned int flags)
+    : MDBRWTransactionImpl(parent, openRWTransaction(parent, nullptr, flags))
+{
+}
+
+MDBRWTransactionImpl::~MDBRWTransactionImpl()
+{
+    MDBRWTransactionImpl::abort();
+}
+
+void MDBRWTransactionImpl::commit()
+{
+    closeRORWCursors();
+    if (!d_txn) {
+        return;
+    }
+
+    if (const auto rc = mdb_txn_commit(d_txn)) {
+        throw LMDBError("Committing transaction: ", rc);
+    }
+    environment().decRWTX();
+    d_txn = nullptr;
+}
+
+void MDBRWTransactionImpl::abort()
+{
+    closeRORWCursors();
+    if (!d_txn) {
+        return;
+    }
+
+    mdb_txn_abort(d_txn);
+    // prevent the RO destructor from cleaning up the transaction itself
+    environment().decRWTX();
+    d_txn = nullptr;
+}
+
+MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, MDB_txn *txn)
+    : d_parent(parent)
+    , d_cursors()
+    , d_txn(txn)
+{
+}
+
+MDB_txn *MDBROTransactionImpl::openROTransaction(MDBEnv *env, MDB_txn *parent, unsigned int flags)
+{
+    if (env->getRWTX())
+        throw LMDBError("Duplicate RO transaction");
+
+    /*
+    A transaction and its cursors must only be used by a single thread, and a thread may only have a single transaction at a time. If MDB_NOTLS is in use, this does not apply to read-only transactions. */
+    MDB_txn *result = nullptr;
+    for (int tries = 0; tries < 3; ++tries) { // it might happen twice, who knows
+        if (const auto rc = mdb_txn_begin(env->d_env, parent, MDB_RDONLY | flags, &result)) {
+            if (rc == MDB_MAP_RESIZED && tries < 2) {
+                // "If the mapsize is increased by another process (..) mdb_txn_begin() will return MDB_MAP_RESIZED.
+                // call mdb_env_set_mapsize with a size of zero to adopt the new size."
+                mdb_env_set_mapsize(env->d_env, 0);
+                continue;
+            }
+            throw LMDBError("Unable to start RO transaction: ", rc);
+        }
+        break;
+    }
+    env->incROTX();
+
+    return result;
+}
+
+void MDBROTransactionImpl::closeROCursors()
+{
+    // we need to move the vector away to ensure that the cursors don’t mess with our iteration.
+    std::vector<MDBROCursor *> buf;
+    std::swap(d_cursors, buf);
+    for (auto &cursor : buf) {
+        cursor->close();
+    }
+}
+
+MDBROTransactionImpl::MDBROTransactionImpl(MDBEnv *parent, unsigned int flags)
+    : MDBROTransactionImpl(parent, openROTransaction(parent, nullptr, flags))
+{
+}
+
+MDBROTransactionImpl::~MDBROTransactionImpl()
+{
+    // this is safe because C++ will not call overrides of virtual methods in destructors.
+    MDBROTransactionImpl::commit();
+}
+
+void MDBROTransactionImpl::abort()
+{
+    closeROCursors();
+    // if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
+    if (d_txn) {
+        d_parent->decROTX();
+        mdb_txn_abort(d_txn); // this appears to work better than abort for r/o database opening
+        d_txn = nullptr;
+    }
+}
+
+void MDBROTransactionImpl::commit()
+{
+    closeROCursors();
+    // if d_txn is non-nullptr here, either the transaction object was invalidated earlier (e.g. by moving from it), or it is an RW transaction which has already cleaned up the d_txn pointer (with an abort).
+    if (d_txn) {
+        d_parent->decROTX();
+        if (const auto rc = mdb_txn_commit(d_txn)) { // this appears to work better than abort for r/o database opening
+            throw LMDBError("Error committing transaction: ", rc);
+        }
+        d_txn = nullptr;
+    }
+}
+
+void MDBRWTransactionImpl::clear(MDB_dbi dbi)
+{
+    if (const auto rc = mdb_drop(d_txn, dbi, 0)) {
+        throw LMDBError("Error clearing database: ", rc);
+    }
+}
+
+MDBRWCursor MDBRWTransactionImpl::getRWCursor(const MDBDbi &dbi)
+{
+    MDB_cursor *cursor;
+    ;
+    if (const auto rc = mdb_cursor_open(d_txn, dbi, &cursor)) {
+        throw LMDBError("Error creating RO cursor: ", rc);
+    }
+    return MDBRWCursor(d_rw_cursors, cursor);
+}
+
+MDBRWCursor MDBRWTransactionImpl::getCursor(const MDBDbi &dbi)
+{
+    return getRWCursor(dbi);
+}
+
+MDBRWTransaction MDBRWTransactionImpl::getRWTransaction()
+{
+    MDB_txn *txn;
+    if (const auto rc = mdb_txn_begin(environment(), *this, 0, &txn)) {
+        throw LMDBError("Failed to start child transaction: ", rc);
+    }
+    // we need to increase the counter here because commit/abort on the child transaction will decrease it
+    environment().incRWTX();
+    return MDBRWTransaction(new MDBRWTransactionImpl(&environment(), txn));
+}
+
+MDBROTransaction MDBRWTransactionImpl::getROTransaction()
+{
+    return getRWTransaction();
+}
+
+MDBROTransaction MDBEnv::getROTransaction()
+{
+    return MDBROTransaction(new MDBROTransactionImpl(this));
+}
+MDBRWTransaction MDBEnv::getRWTransaction()
+{
+    return MDBRWTransaction(new MDBRWTransactionImpl(this));
+}
+
+void MDBRWTransactionImpl::closeRWCursors()
+{
+    decltype(d_rw_cursors) buf;
+    std::swap(d_rw_cursors, buf);
+    for (auto &cursor : buf) {
+        cursor->close();
+    }
+}
+
+MDBROCursor MDBROTransactionImpl::getCursor(const MDBDbi &dbi)
+{
+    return getROCursor(dbi);
+}
+
+MDBROCursor MDBROTransactionImpl::getROCursor(const MDBDbi &dbi)
+{
+    MDB_cursor *cursor;
+    if (const auto rc = mdb_cursor_open(d_txn, dbi, &cursor)) {
+        throw LMDBError("Error creating RO cursor: ", rc);
+    }
+    return MDBROCursor(d_cursors, cursor);
+}
+
+} // namespace LMDBSafe

src/rgw/driver/posix/lmdb-safe.hh

@@ -0,0 +1,638 @@
+/*
+MIT License
+
+Copyright (c) 2018 bert hubert
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+*/
+#pragma once
+
+#include "lmdb-safe-global.h"
+
+#include <lmdb.h>
+
+#include <algorithm>
+#include <cstring>
+#include <limits>
+#include <map>
+#include <memory>
+#include <mutex>
+#include <stdexcept>
+#include <string>
+#include <thread>
+#include <vector>
+
+/*!
+ * \brief The LMDBSafe namespace contains all classes/types contained by the lmdb-safe and
+ *        lmdb-typed libraries.
+ * \remarks
+ * - Error strategy: Anything that "should never happen" turns into an exception. But things
+ *   like "duplicate entry" or "no such key" are for you to deal with.
+ * - Thread safety: We are as safe as LMDB. You can talk to MDBEnv from as many threads as you
+ *   want.
+ */
+namespace LMDBSafe {
+
+// apple compiler somehow has string_view even in c++11!
+#ifdef __cpp_lib_string_view
+using std::string_view;
+#else
+#include <boost/version.hpp>
+#if BOOST_VERSION > 105400
+#include <boost/utility/string_view.hpp>
+using boost::string_view;
+#else
+#include <boost/utility/string_ref.hpp>
+using string_view = boost::string_ref;
+#endif
+#endif
+
+/*!
+ * \brief The LMDBError class is thrown when an error happens.
+ */
+class LMDB_SAFE_EXPORT LMDBError : public std::runtime_error {
+public:
+    explicit LMDBError(const std::string &error) noexcept
+        : std::runtime_error(error)
+        , ec(0)
+    {
+    }
+
+    explicit LMDBError(const std::string &context, int error) noexcept
+        : std::runtime_error(context + mdb_strerror(error))
+        , ec(error)
+    {
+    }
+
+    const int ec;
+};
+
+/*!
+ * \brief The MDBDbi class is our only 'value type' object as 1) a dbi is actually an integer
+ *        and 2) per LMDB documentation, we never close it.
+ */
+class LMDB_SAFE_EXPORT MDBDbi {
+public:
+    MDBDbi()
+    {
+        d_dbi = std::numeric_limits<decltype(d_dbi)>::max();
+    }
+    explicit MDBDbi(MDB_env *env, MDB_txn *txn, string_view dbname, unsigned int flags);
+
+    operator const MDB_dbi &() const
+    {
+        return d_dbi;
+    }
+
+    MDB_dbi d_dbi;
+};
+
+class MDBRWTransactionImpl;
+class MDBROTransactionImpl;
+using MDBROTransaction = std::unique_ptr<MDBROTransactionImpl>;
+using MDBRWTransaction = std::unique_ptr<MDBRWTransactionImpl>;
+
+/*!
+ * \brief The MDBEnv class is a handle to an MDB environment.
+ */
+class LMDB_SAFE_EXPORT MDBEnv {
+public:
+    MDBEnv(const char *fname, unsigned int flags, mdb_mode_t mode, MDB_dbi maxDBs = 10);
+
+    /*!
+     * \brief Closes the MDB environment.
+     * \remarks Only a single thread may call this function. All transactions, databases, and cursors must already be closed
+     *          before calling this function.
+     */
+    ~MDBEnv()
+    {
+        mdb_env_close(d_env);
+    }
+
+    MDBDbi openDB(const string_view dbname, unsigned int flags);
+
+    MDBRWTransaction getRWTransaction();
+    MDBROTransaction getROTransaction();
+
+    operator MDB_env *&()
+    {
+        return d_env;
+    }
+    MDB_env *d_env;
+
+    int getRWTX();
+    void incRWTX();
+    void decRWTX();
+    int getROTX();
+    void incROTX();
+    void decROTX();
+
+private:
+    std::mutex d_openmut;
+    std::mutex d_countmutex;
+    std::map<std::thread::id, int> d_RWtransactionsOut;
+    std::map<std::thread::id, int> d_ROtransactionsOut;
+};
+
+/*!
+ * \brief Opens an MDB environment for the specified database file.
+ */
+LMDB_SAFE_EXPORT std::shared_ptr<MDBEnv> getMDBEnv(const char *fname, unsigned int flags, mdb_mode_t mode, MDB_dbi maxDBs = 128);
+
+/*!
+ * \brief The MDBOutVal struct is the handle to an MDB value used as output.
+ */
+struct LMDB_SAFE_EXPORT MDBOutVal {
+    operator MDB_val &()
+    {
+        return d_mdbval;
+    }
+
+    template <class T, typename std::enable_if<std::is_arithmetic<T>::value, T>::type * = nullptr> const T get()
+    {
+        T ret;
+        if (d_mdbval.mv_size != sizeof(T))
+            throw LMDBError("MDB data has wrong length for type");
+
+        memcpy(&ret, d_mdbval.mv_data, sizeof(T));
+        return ret;
+    }
+
+    template <class T, typename std::enable_if<std::is_class<T>::value, T>::type * = nullptr> T get() const;
+
+    template <class T> T get_struct() const
+    {
+        T ret;
+        if (d_mdbval.mv_size != sizeof(T))
+            throw LMDBError("MDB data has wrong length for type");
+
+        memcpy(&ret, d_mdbval.mv_data, sizeof(T));
+        return ret;
+    }
+
+    template <class T> const T *get_struct_ptr() const
+    {
+        if (d_mdbval.mv_size != sizeof(T))
+            throw LMDBError("MDB data has wrong length for type");
+
+        return reinterpret_cast<const T *>(d_mdbval.mv_data);
+    }
+
+    MDB_val d_mdbval;
+};
+
+template <> inline std::string MDBOutVal::get<std::string>() const
+{
+    return std::string(static_cast<char *>(d_mdbval.mv_data), d_mdbval.mv_size);
+}
+
+template <> inline string_view MDBOutVal::get<string_view>() const
+{
+    return string_view(static_cast<char *>(d_mdbval.mv_data), d_mdbval.mv_size);
+}
+
+/*!
+ * \brief The MDBInVal struct is the handle to an MDB value used as input.
+ */
+class LMDB_SAFE_EXPORT MDBInVal {
+public:
+    MDBInVal(const MDBOutVal &rhs)
+    {
+        d_mdbval = rhs.d_mdbval;
+    }
+
+    template <class T, typename std::enable_if<std::is_arithmetic<T>::value, T>::type * = nullptr> MDBInVal(T i)
+    {
+        memcpy(&d_memory[0], &i, sizeof(i));
+        d_mdbval.mv_size = sizeof(T);
+        d_mdbval.mv_data = d_memory;
+        ;
+    }
+
+    MDBInVal(const char *s)
+    {
+        d_mdbval.mv_size = strlen(s);
+        d_mdbval.mv_data = static_cast<void *>(const_cast<char *>(s));
+    }
+
+    MDBInVal(string_view v)
+    {
+        d_mdbval.mv_size = v.size();
+        d_mdbval.mv_data = static_cast<void *>(const_cast<char *>(v.data()));
+    }
+
+    MDBInVal(const std::string &v)
+    {
+        d_mdbval.mv_size = v.size();
+        d_mdbval.mv_data = static_cast<void *>(const_cast<char *>(v.data()));
+    }
+
+    template <typename T> static MDBInVal fromStruct(const T &t)
+    {
+        MDBInVal ret;
+        ret.d_mdbval.mv_size = sizeof(T);
+        ret.d_mdbval.mv_data = static_cast<void *>(&const_cast<T &>(t));
+        return ret;
+    }
+
+    operator MDB_val &()
+    {
+        return d_mdbval;
+    }
+    MDB_val d_mdbval;
+
+private:
+    MDBInVal()
+    {
+    }
+    char d_memory[sizeof(double)];
+};
+
+class MDBROCursor;
+
+/*!
+ * \brief The MDBROTransactionImpl class wraps read operations.
+ */
+class LMDB_SAFE_EXPORT MDBROTransactionImpl {
+protected:
+    MDBROTransactionImpl(MDBEnv *parent, MDB_txn *txn);
+
+private:
+    static MDB_txn *openROTransaction(MDBEnv *env, MDB_txn *parent, unsigned int flags = 0);
+
+    MDBEnv *d_parent;
+    std::vector<MDBROCursor *> d_cursors;
+
+protected:
+    MDB_txn *d_txn;
+
+    void closeROCursors();
+
+public:
+    explicit MDBROTransactionImpl(MDBEnv *parent, unsigned int flags = 0);
+
+    MDBROTransactionImpl(const MDBROTransactionImpl &src) = delete;
+    MDBROTransactionImpl &operator=(const MDBROTransactionImpl &src) = delete;
+
+    // The move constructor/operator cannot be made safe due to Object Slicing with MDBRWTransaction.
+    MDBROTransactionImpl(MDBROTransactionImpl &&rhs) = delete;
+    MDBROTransactionImpl &operator=(MDBROTransactionImpl &&rhs) = delete;
+
+    virtual ~MDBROTransactionImpl();
+
+    virtual void abort();
+    virtual void commit();
+
+    int get(MDB_dbi dbi, const MDBInVal &key, MDBOutVal &val)
+    {
+        if (!d_txn)
+            throw LMDBError("Attempt to use a closed RO transaction for get");
+
+        const auto rc = mdb_get(d_txn, dbi, const_cast<MDB_val *>(&key.d_mdbval), &val.d_mdbval);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Getting data: ", rc);
+
+        return rc;
+    }
+
+    int get(MDB_dbi dbi, const MDBInVal &key, string_view &val)
+    {
+        MDBOutVal out;
+        int rc = get(dbi, key, out);
+        if (!rc)
+            val = out.get<string_view>();
+        return rc;
+    }
+
+    // this is something you can do, readonly
+    MDBDbi openDB(string_view dbname, unsigned int flags)
+    {
+        return MDBDbi(d_parent->d_env, d_txn, dbname, flags);
+    }
+
+    MDBROCursor getCursor(const MDBDbi &);
+    MDBROCursor getROCursor(const MDBDbi &);
+
+    operator MDB_txn *()
+    {
+        return d_txn;
+    }
+
+    inline operator bool() const
+    {
+        return d_txn;
+    }
+
+    inline MDBEnv &environment()
+    {
+        return *d_parent;
+    }
+};
+
+/*!
+ * \brief The MDBGenCursor class represents a MDB_cursor handle.
+ * \remarks
+ * - A cursor in a read-only transaction must be closed explicitly, before or after its transaction ends.
+ *   It can be reused with mdb_cursor_renew() before finally closing it.
+ * - "If the parent transaction commits, the cursor must not be used again."
+ */
+template <class Transaction, class T> class MDBGenCursor {
+private:
+    std::vector<T *> *d_registry;
+    MDB_cursor *d_cursor;
+
+public:
+    MDBGenCursor()
+        : d_registry(nullptr)
+        , d_cursor(nullptr)
+    {
+    }
+
+    MDBGenCursor(std::vector<T *> &registry, MDB_cursor *cursor)
+        : d_registry(&registry)
+        , d_cursor(cursor)
+    {
+        registry.emplace_back(static_cast<T *>(this));
+    }
+
+private:
+    void move_from(MDBGenCursor *src)
+    {
+        if (!d_registry) {
+            return;
+        }
+
+        auto iter = std::find(d_registry->begin(), d_registry->end(), src);
+        if (iter != d_registry->end()) {
+            *iter = static_cast<T *>(this);
+        } else {
+            d_registry->emplace_back(static_cast<T *>(this));
+        }
+    }
+
+public:
+    MDBGenCursor(const MDBGenCursor &src) = delete;
+
+    MDBGenCursor(MDBGenCursor &&src) noexcept
+        : d_registry(src.d_registry)
+        , d_cursor(src.d_cursor)
+    {
+        move_from(&src);
+        src.d_registry = nullptr;
+        src.d_cursor = nullptr;
+    }
+
+    MDBGenCursor &operator=(const MDBGenCursor &src) = delete;
+
+    MDBGenCursor &operator=(MDBGenCursor &&src) noexcept
+    {
+        d_registry = src.d_registry;
+        d_cursor = src.d_cursor;
+        move_from(&src);
+        src.d_registry = nullptr;
+        src.d_cursor = nullptr;
+        return *this;
+    }
+
+    ~MDBGenCursor()
+    {
+        close();
+    }
+
+public:
+    int get(MDBOutVal &key, MDBOutVal &data, MDB_cursor_op op)
+    {
+        const auto rc = mdb_cursor_get(d_cursor, &key.d_mdbval, &data.d_mdbval, op);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Unable to get from cursor: ", rc);
+        return rc;
+    }
+
+    int find(const MDBInVal &in, MDBOutVal &key, MDBOutVal &data)
+    {
+        key.d_mdbval = in.d_mdbval;
+        const auto rc = mdb_cursor_get(d_cursor, const_cast<MDB_val *>(&key.d_mdbval), &data.d_mdbval, MDB_SET);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Unable to find from cursor: ", rc);
+        return rc;
+    }
+
+    int lower_bound(const MDBInVal &in, MDBOutVal &key, MDBOutVal &data)
+    {
+        key.d_mdbval = in.d_mdbval;
+
+        const auto rc = mdb_cursor_get(d_cursor, const_cast<MDB_val *>(&key.d_mdbval), &data.d_mdbval, MDB_SET_RANGE);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Unable to lower_bound from cursor: ", rc);
+        return rc;
+    }
+
+    int nextprev(MDBOutVal &key, MDBOutVal &data, MDB_cursor_op op)
+    {
+        const auto rc = mdb_cursor_get(d_cursor, &key.d_mdbval, &data.d_mdbval, op);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Unable to prevnext from cursor: ", rc);
+        return rc;
+    }
+
+    int next(MDBOutVal &key, MDBOutVal &data)
+    {
+        return nextprev(key, data, MDB_NEXT);
+    }
+
+    int prev(MDBOutVal &key, MDBOutVal &data)
+    {
+        return nextprev(key, data, MDB_PREV);
+    }
+
+    int currentlast(MDBOutVal &key, MDBOutVal &data, MDB_cursor_op op)
+    {
+        const auto rc = mdb_cursor_get(d_cursor, &key.d_mdbval, &data.d_mdbval, op);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Unable to next from cursor: ", rc);
+        return rc;
+    }
+
+    int current(MDBOutVal &key, MDBOutVal &data)
+    {
+        return currentlast(key, data, MDB_GET_CURRENT);
+    }
+    int last(MDBOutVal &key, MDBOutVal &data)
+    {
+        return currentlast(key, data, MDB_LAST);
+    }
+    int first(MDBOutVal &key, MDBOutVal &data)
+    {
+        return currentlast(key, data, MDB_FIRST);
+    }
+
+    operator MDB_cursor *()
+    {
+        return d_cursor;
+    }
+
+    operator bool() const
+    {
+        return d_cursor;
+    }
+
+    void close()
+    {
+        if (d_registry) {
+            auto iter = std::find(d_registry->begin(), d_registry->end(), static_cast<T *>(this));
+            if (iter != d_registry->end()) {
+                d_registry->erase(iter);
+            }
+            d_registry = nullptr;
+        }
+        if (d_cursor) {
+            mdb_cursor_close(d_cursor);
+            d_cursor = nullptr;
+        }
+    }
+};
+
+/*!
+ * \brief The MDBROCursor class represents a read-only cursor.
+ */
+class LMDB_SAFE_EXPORT MDBROCursor : public MDBGenCursor<MDBROTransactionImpl, MDBROCursor> {
+public:
+    MDBROCursor() = default;
+    using MDBGenCursor<MDBROTransactionImpl, MDBROCursor>::MDBGenCursor;
+    MDBROCursor(const MDBROCursor &src) = delete;
+    MDBROCursor(MDBROCursor &&src) = default;
+    MDBROCursor &operator=(const MDBROCursor &src) = delete;
+    MDBROCursor &operator=(MDBROCursor &&src) = default;
+    ~MDBROCursor() = default;
+};
+
+class MDBRWCursor;
+
+/*!
+ * \brief The MDBRWTransactionImpl class wraps write operations.
+ */
+class LMDB_SAFE_EXPORT MDBRWTransactionImpl : public MDBROTransactionImpl {
+protected:
+    MDBRWTransactionImpl(MDBEnv *parent, MDB_txn *txn);
+
+private:
+    static MDB_txn *openRWTransaction(MDBEnv *env, MDB_txn *parent, unsigned int flags);
+
+private:
+    std::vector<MDBRWCursor *> d_rw_cursors;
+
+    void closeRWCursors();
+    inline void closeRORWCursors()
+    {
+        closeROCursors();
+        closeRWCursors();
+    }
+
+public:
+    explicit MDBRWTransactionImpl(MDBEnv *parent, unsigned int flags = 0);
+
+    MDBRWTransactionImpl(const MDBRWTransactionImpl &rhs) = delete;
+    MDBRWTransactionImpl(MDBRWTransactionImpl &&rhs) = delete;
+    MDBRWTransactionImpl &operator=(const MDBRWTransactionImpl &rhs) = delete;
+    MDBRWTransactionImpl &operator=(MDBRWTransactionImpl &&rhs) = delete;
+
+    ~MDBRWTransactionImpl() override;
+
+    void commit() override;
+    void abort() override;
+
+    void clear(MDB_dbi dbi);
+
+    void put(MDB_dbi dbi, const MDBInVal &key, const MDBInVal &val, unsigned int flags = 0)
+    {
+        if (!d_txn)
+            throw LMDBError("Attempt to use a closed RW transaction for put");
+        if (const auto rc = mdb_put(d_txn, dbi, const_cast<MDB_val *>(&key.d_mdbval), const_cast<MDB_val *>(&val.d_mdbval), flags))
+            throw LMDBError("Putting data: ", rc);
+    }
+
+    int del(MDBDbi &dbi, const MDBInVal &key, const MDBInVal &val)
+    {
+        const auto rc = mdb_del(d_txn, dbi, const_cast<MDB_val *>(&key.d_mdbval), const_cast<MDB_val *>(&val.d_mdbval));
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Deleting data: ", rc);
+        return rc;
+    }
+
+    int del(MDBDbi &dbi, const MDBInVal &key)
+    {
+        const auto rc = mdb_del(d_txn, dbi, const_cast<MDB_val *>(&key.d_mdbval), 0);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Deleting data: ", rc);
+        return rc;
+    }
+
+    int get(MDBDbi &dbi, const MDBInVal &key, MDBOutVal &val)
+    {
+        if (!d_txn)
+            throw LMDBError("Attempt to use a closed RW transaction for get");
+
+        const auto rc = mdb_get(d_txn, dbi, const_cast<MDB_val *>(&key.d_mdbval), &val.d_mdbval);
+        if (rc && rc != MDB_NOTFOUND)
+            throw LMDBError("Getting data: ", rc);
+        return rc;
+    }
+
+    int get(MDBDbi &dbi, const MDBInVal &key, string_view &val)
+    {
+        MDBOutVal out;
+        const auto rc = get(dbi, key, out);
+        if (!rc)
+            val = out.get<string_view>();
+        return rc;
+    }
+
+    MDBDbi openDB(string_view dbname, unsigned int flags)
+    {
+        return MDBDbi(environment().d_env, d_txn, dbname, flags);
+    }
+
+    MDBRWCursor getRWCursor(const MDBDbi &);
+    MDBRWCursor getCursor(const MDBDbi &);
+
+    MDBRWTransaction getRWTransaction();
+    MDBROTransaction getROTransaction();
+};
+
+/*!
+ * \brief The MDBRWCursor class implements RW operations based on MDBGenCursor.
+ * \remarks
+ * - "A cursor in a write-transaction can be closed before its transaction ends, and will otherwise
+ *   be closed when its transaction ends." This is a problem for us since it may means we are closing
+ *   the cursor twice, which is bad.
+ */
+class LMDB_SAFE_EXPORT MDBRWCursor : public MDBGenCursor<MDBRWTransactionImpl, MDBRWCursor> {
+public:
+    MDBRWCursor() = default;
+    using MDBGenCursor<MDBRWTransactionImpl, MDBRWCursor>::MDBGenCursor;
+    MDBRWCursor(const MDBRWCursor &src) = delete;
+    MDBRWCursor(MDBRWCursor &&src) = default;
+    MDBRWCursor &operator=(const MDBRWCursor &src) = delete;
+    MDBRWCursor &operator=(MDBRWCursor &&src) = default;
+    ~MDBRWCursor() = default;
+
+    void put(const MDBOutVal &key, const MDBInVal &data)
+    {
+        if (const auto rc = mdb_cursor_put(*this, const_cast<MDB_val *>(&key.d_mdbval), const_cast<MDB_val *>(&data.d_mdbval), MDB_CURRENT))
+            throw LMDBError("Putting data via mdb_cursor_put: ", rc);
+    }
+
+    void put(const MDBOutVal &key, const MDBOutVal &data, unsigned int flags = 0)
+    {
+        if (const auto rc = mdb_cursor_put(*this, const_cast<MDB_val *>(&key.d_mdbval), const_cast<MDB_val *>(&data.d_mdbval), flags))
+            throw LMDBError("Putting data via mdb_cursor_put: ", rc);
+    }
+
+    void del(unsigned int flags = 0)
+    {
+        if (const auto rc = mdb_cursor_del(*this, flags))
+            throw LMDBError("Deleting data via mdb_cursor_del: ", rc);
+    }
+};
+
+} // namespace LMDBSafe

src/rgw/driver/posix/notify.cpp

@@ -0,0 +1,21 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#include "notify.h"
+#ifdef linux
+#include <sys/inotify.h>
+#endif
+
+namespace file::listing {
+
+  std::unique_ptr<Notify> Notify::factory(Notifiable* n, const std::string& bucket_root)
+  {
+#ifdef __linux__
+    return std::unique_ptr<Notify>(new Inotify(n, bucket_root));
+#else
+#error currently, rgw posix driver requires inotify
+#endif /* linux */
+    return nullptr;
+  } /* Notify::factory */
+
+} // namespace file::listing

src/rgw/driver/posix/notify.h

@@ -0,0 +1,255 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#pragma once
+
+#include <iostream>
+#include <string>
+#include <memory>
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <chrono>
+#include <optional>
+#include <filesystem>
+#include <limits>
+#include <cstdlib>
+#include "unordered_dense.h"
+#include <unistd.h>
+#include <poll.h>
+#ifdef __linux__
+#include <sys/inotify.h>
+#include <sys/eventfd.h>
+#endif
+#include <fmt/format.h>
+
+namespace file::listing {
+
+  namespace sf = std::filesystem;
+
+  class Notifiable
+  {
+  public:
+    enum class EventType : uint8_t
+    {
+      ADD = 0,
+      REMOVE,
+      INVALIDATE
+    };
+
+    struct Event
+    {
+      EventType type;
+      std::optional<std::string_view> name;
+
+      Event(EventType type, std::optional<std::string_view> name) noexcept
+	: type(type), name(name)
+	{}
+
+      Event(Event&& rhs) noexcept
+	: type(rhs.type), name(rhs.name)
+	{}
+    };
+
+    virtual ~Notifiable() {};
+
+    virtual int notify(const std::string&, void*, const std::vector<Event>&) = 0;
+  };
+
+  class Notify
+  {
+    Notifiable* n;
+    sf::path rp;
+
+    Notify(Notifiable* n, const std::string& bucket_root)
+      : n(n), rp(bucket_root)
+      {}
+
+    friend class Inotify;
+  public:
+    static std::unique_ptr<Notify> factory(Notifiable* n, const std::string& bucket_root);
+    
+    virtual int add_watch(const std::string& dname, void* opaque) = 0;
+    virtual int remove_watch(const std::string& dname) = 0;
+    virtual ~Notify()
+      {}
+  }; /* Notify */
+
+#ifdef __linux__
+  class Inotify : public Notify
+  {
+    static constexpr uint32_t rd_size = 8192;
+    static constexpr uint32_t aw_mask = IN_ALL_EVENTS &
+      ~(IN_MOVE_SELF|IN_OPEN|IN_ACCESS|IN_ATTRIB|IN_CLOSE_WRITE|IN_CLOSE_NOWRITE|IN_MODIFY|IN_DELETE_SELF);
+
+    static constexpr uint64_t sig_shutdown = std::numeric_limits<uint64_t>::max() - 0xdeadbeef;
+
+    class WatchRecord
+    {
+    public:
+      int wd;
+      std::string name;
+      void* opaque;
+    public:
+      WatchRecord(int wd, const std::string& name, void* opaque) noexcept
+	: wd(wd), name(name), opaque(opaque)
+	{}
+
+      WatchRecord(WatchRecord&& wr) noexcept
+	: wd(wr.wd), name(wr.name), opaque(wr.opaque)
+	{}
+
+      WatchRecord& operator=(WatchRecord&& wr) {
+	wd = wr.wd;
+	name = std::move(wr.name);
+	opaque = wr.opaque;
+	return *this;
+      }
+    }; /* WatchRecord */
+
+    using wd_callback_map_t = ankerl::unordered_dense::map<int, WatchRecord>;
+    using wd_remove_map_t = ankerl::unordered_dense::map<std::string, int>;
+
+    int wfd, efd;
+    std::thread thrd;
+    wd_callback_map_t wd_callback_map;
+    wd_remove_map_t wd_remove_map;
+    bool shutdown{false};
+
+    class AlignedBuf
+    {
+      char* m;
+    public:
+      AlignedBuf() {
+	m = static_cast<char*>(aligned_alloc(__alignof__(struct inotify_event), rd_size));
+	if (! m) [[unlikely]] {
+	  std::cerr << fmt::format("{} buffer allocation failure", __func__) << std::endl;
+	  abort();
+	}
+      }
+      ~AlignedBuf() {
+	std::free(m);
+      }
+      char* get() {
+	return m;
+      }
+    }; /* AlignedBuf */
+    
+    void ev_loop() {
+      std::unique_ptr<AlignedBuf> up_buf = std::make_unique<AlignedBuf>();
+      struct inotify_event* event;
+      char* buf = up_buf.get()->get();
+      ssize_t len;
+      int npoll;
+
+      nfds_t nfds{2};
+      struct pollfd fds[2] = {{wfd, POLLIN}, {efd, POLLIN}};
+
+    restart:
+      while(! shutdown) {
+	npoll = poll(fds, nfds, -1); /* for up to 10 fds, poll is fast as epoll */
+	if (shutdown) {
+	  return;
+	}
+	if (npoll == -1) {
+	  if (errno == EINTR) {
+	    continue;
+	  }
+	  // XXX
+	}
+	if (npoll > 0) {
+	  len = read(wfd, buf, rd_size);
+	  if (len == -1) {
+	    continue; // hopefully, was EAGAIN
+	  }
+	  std::vector<Notifiable::Event> evec;
+	  for (char* ptr = buf; ptr < buf + len;
+	       ptr += sizeof(struct inotify_event) + event->len) {
+	    event = reinterpret_cast<struct inotify_event*>(ptr);
+	    const auto& it = wd_callback_map.find(event->wd);
+	    //std::cout << fmt::format("event! {}", event->name) << std::endl;
+	    if (it == wd_callback_map.end()) [[unlikely]] {
+	      /* non-destructive race, it happens */
+	      continue;
+	    }
+	    const auto& wr = it->second;
+	    if (event->mask & IN_Q_OVERFLOW) [[unlikely]] {
+	      /* cache blown, invalidate */
+	      evec.clear();
+	      evec.emplace_back(Notifiable::Event(Notifiable::EventType::INVALIDATE, std::nullopt));
+	      n->notify(wr.name, wr.opaque, evec);
+	      goto restart;
+	    } else {
+	      if ((event->mask & IN_CREATE) ||
+		  (event->mask & IN_MOVED_TO)) {
+		/* new object in dir */
+		evec.emplace_back(Notifiable::Event(Notifiable::EventType::ADD, event->name));
+	      } else if ((event->mask & IN_DELETE) ||
+			 (event->mask & IN_MOVED_FROM)) {
+		/* object removed from dir */
+		evec.emplace_back(Notifiable::Event(Notifiable::EventType::REMOVE, event->name));
+	      }
+	    } /* !overflow */
+	    if (evec.size() > 0) {
+	      n->notify(wr.name, wr.opaque, evec);
+	    }
+	  } /* events */
+	} /* n > 0 */
+      }
+    } /* ev_loop */
+
+    Inotify(Notifiable* n, const std::string& bucket_root)
+      : Notify(n, bucket_root),
+	thrd(&Inotify::ev_loop, this)
+      {
+	wfd = inotify_init1(IN_NONBLOCK);
+	if (wfd == -1) {
+	  std::cerr << fmt::format("{} inotify_init1 failed with {}", __func__, wfd) << std::endl;
+	  exit(1);
+	}
+	efd = eventfd(0, EFD_NONBLOCK);
+      }
+
+    void signal_shutdown() {
+      uint64_t msg{sig_shutdown};
+      (void) write(efd, &msg, sizeof(uint64_t));
+    }
+
+    friend class Notify;
+  public:
+    virtual int add_watch(const std::string& dname, void* opaque) override {
+      sf::path wp{rp / dname};
+      int wd = inotify_add_watch(wfd, wp.c_str(), aw_mask);
+      if (wd == -1) {
+	std::cerr << fmt::format("{} inotify_add_watch {} failed with {}", __func__, dname, wd) << std::endl;
+      } else {
+	wd_callback_map.insert(wd_callback_map_t::value_type(wd, WatchRecord(wd, dname, opaque)));
+	wd_remove_map.insert(wd_remove_map_t::value_type(dname, wd));
+      }
+      return wd;
+    }
+
+    virtual int remove_watch(const std::string& dname) override {
+      int r{0};
+      const auto& elt = wd_remove_map.find(dname);
+      if (elt != wd_remove_map.end()) {
+	auto& wd = elt->second;
+	r = inotify_rm_watch(wfd, wd);
+	if (r == -1) {
+	  std::cerr << fmt::format("{} inotify_rm_watch {} failed with {}", __func__, dname, wd) << std::endl;
+	}
+	wd_callback_map.erase(wd);
+	wd_remove_map.erase(std::string(dname));
+      }
+      return r;
+    }
+
+    virtual ~Inotify() {
+      shutdown = true;
+      signal_shutdown();
+      thrd.join();
+    }
+  };
+#endif /* linux */
+
+} // namespace file::listing

src/rgw/driver/posix/rgw_sal_posix.h

@@ -0,0 +1,688 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright contributors to the Ceph project
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#pragma once
+
+#include "rgw_sal_filter.h"
+#include "rgw_sal_store.h"
+#include <memory>
+#include "common/dout.h"
+#include "bucket_cache.h"
+
+namespace rgw { namespace sal {
+
+class POSIXDriver;
+class POSIXBucket;
+class POSIXObject;
+
+using BucketCache = file::listing::BucketCache<POSIXDriver, POSIXBucket>;
+
+class POSIXDriver : public FilterDriver {
+private:
+
+  std::unique_ptr<BucketCache> bucket_cache;
+  std::string base_path;
+  int root_fd;
+
+public:
+  POSIXDriver(Driver* _next) : FilterDriver(_next)
+  { }
+  virtual ~POSIXDriver() { close(); }
+  virtual int initialize(CephContext *cct, const DoutPrefixProvider *dpp) override;
+  virtual std::unique_ptr<User> get_user(const rgw_user& u) override;
+  virtual int get_user_by_access_key(const DoutPrefixProvider* dpp, const
+				     std::string& key, optional_yield y,
+				     std::unique_ptr<User>* user) override;
+  virtual int get_user_by_email(const DoutPrefixProvider* dpp, const
+				std::string& email, optional_yield y,
+				std::unique_ptr<User>* user) override;
+  virtual int get_user_by_swift(const DoutPrefixProvider* dpp, const
+				std::string& user_str, optional_yield y,
+				std::unique_ptr<User>* user) override;
+  virtual std::unique_ptr<Object> get_object(const rgw_obj_key& k) override;
+  virtual int get_bucket(User* u, const RGWBucketInfo& i,
+			 std::unique_ptr<Bucket>* bucket) override;
+  virtual int get_bucket(const DoutPrefixProvider* dpp, User* u, const
+			 rgw_bucket& b, std::unique_ptr<Bucket>* bucket,
+			 optional_yield y) override;
+  virtual int get_bucket(const DoutPrefixProvider* dpp, User* u, const
+			 std::string& tenant, const std::string& name,
+			 std::unique_ptr<Bucket>* bucket, optional_yield y) override;
+  virtual std::string zone_unique_trans_id(const uint64_t unique_num) override;
+
+  virtual std::unique_ptr<Writer> get_append_writer(const DoutPrefixProvider *dpp,
+				  optional_yield y,
+				  rgw::sal::Object* _head_obj,
+				  const rgw_user& owner,
+				  const rgw_placement_rule *ptail_placement_rule,
+				  const std::string& unique_tag,
+				  uint64_t position,
+				  uint64_t *cur_accounted_size) override;
+  virtual std::unique_ptr<Writer> get_atomic_writer(const DoutPrefixProvider *dpp,
+				  optional_yield y,
+				  rgw::sal::Object* _head_obj,
+				  const rgw_user& owner,
+				  const rgw_placement_rule *ptail_placement_rule,
+				  uint64_t olh_epoch,
+				  const std::string& unique_tag) override;
+
+  virtual void finalize(void) override;
+  virtual void register_admin_apis(RGWRESTMgr* mgr) override;
+
+  virtual std::unique_ptr<Notification> get_notification(rgw::sal::Object* obj,
+				 rgw::sal::Object* src_obj, struct req_state* s,
+				 rgw::notify::EventType event_type, optional_yield y,
+				 const std::string* object_name=nullptr) override;
+
+  virtual std::unique_ptr<Notification> get_notification(
+                                  const DoutPrefixProvider* dpp,
+                                  rgw::sal::Object* obj,
+                                  rgw::sal::Object* src_obj,
+                                  rgw::notify::EventType event_type,
+                                  rgw::sal::Bucket* _bucket,
+                                  std::string& _user_id,
+                                  std::string& _user_tenant,
+                                  std::string& _req_id,
+                                  optional_yield y) override;
+
+  /* Internal APIs */
+  int get_root_fd() { return root_fd; }
+  const std::string& get_base_path() const { return base_path; }
+  BucketCache* get_bucket_cache() { return bucket_cache.get(); }
+
+  int close();
+
+  /* called by BucketCache layer when a new object is discovered
+   * by inotify or similar */
+  int mint_listing_entry(
+    const std::string& bucket, rgw_bucket_dir_entry& bde /* OUT */);
+
+};
+
+class POSIXUser : public FilterUser {
+private:
+  POSIXDriver* driver;
+
+public:
+  POSIXUser(std::unique_ptr<User> _next, POSIXDriver* _driver) :
+    FilterUser(std::move(_next)),
+    driver(_driver) {}
+  virtual ~POSIXUser() = default;
+
+  virtual int list_buckets(const DoutPrefixProvider* dpp,
+			   const std::string& marker, const std::string& end_marker,
+			   uint64_t max, bool need_stats, BucketList& buckets,
+			   optional_yield y) override;
+  virtual int create_bucket(const DoutPrefixProvider* dpp,
+                            const rgw_bucket& b,
+                            const std::string& zonegroup_id,
+                            rgw_placement_rule& placement_rule,
+                            std::string& swift_ver_location,
+                            const RGWQuotaInfo* pquota_info,
+                            const RGWAccessControlPolicy& policy,
+                            Attrs& attrs,
+                            RGWBucketInfo& info,
+                            obj_version& ep_objv,
+                            bool exclusive,
+                            bool obj_lock_enabled,
+                            bool* existed,
+                            req_info& req_info,
+                            std::unique_ptr<Bucket>* bucket,
+                            optional_yield y) override;
+  virtual Attrs& get_attrs() override { return next->get_attrs(); }
+  virtual void set_attrs(Attrs& _attrs) override { next->set_attrs(_attrs); }
+  virtual int read_attrs(const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual int merge_and_store_attrs(const DoutPrefixProvider* dpp, Attrs&
+				    new_attrs, optional_yield y) override;
+  virtual int load_user(const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual int store_user(const DoutPrefixProvider* dpp, optional_yield y, bool
+			 exclusive, RGWUserInfo* old_info = nullptr) override;
+  virtual int remove_user(const DoutPrefixProvider* dpp, optional_yield y) override;
+};
+
+class POSIXBucket : public StoreBucket {
+private:
+  POSIXDriver* driver;
+  int parent_fd{-1};
+  int dir_fd{-1};
+  struct statx stx;
+  bool stat_done{false};
+  RGWAccessControlPolicy acls;
+  std::optional<std::string> ns;
+
+public:
+  POSIXBucket(POSIXDriver *_dr, int _p_fd, const rgw_bucket& _b, User* _u, std::optional<std::string> _ns = std::nullopt)
+    : StoreBucket(_b, _u),
+    driver(_dr),
+    parent_fd(_p_fd),
+    acls(),
+    ns(_ns)
+    { }
+
+  POSIXBucket(POSIXDriver *_dr, int _p_fd, const RGWBucketEnt& _e, User* _u)
+    : StoreBucket(_e, _u),
+    driver(_dr),
+    parent_fd(_p_fd),
+    acls()
+    { }
+
+  POSIXBucket(POSIXDriver *_dr, int _p_fd, const RGWBucketInfo& _i, User* _u)
+    : StoreBucket(_i, _u),
+    driver(_dr),
+    parent_fd(_p_fd),
+    acls()
+    { }
+
+  POSIXBucket(const POSIXBucket& _b) :
+    StoreBucket(_b),
+    driver(_b.driver),
+    parent_fd(_b.parent_fd),
+    /* Don't want to copy dir_fd */
+    stx(_b.stx),
+    stat_done(_b.stat_done),
+    acls(_b.acls),
+    ns(_b.ns) {}
+
+  virtual ~POSIXBucket() { close(); }
+
+  virtual std::unique_ptr<Object> get_object(const rgw_obj_key& key) override;
+  virtual int list(const DoutPrefixProvider* dpp, ListParams&, int,
+		   ListResults&, optional_yield y) override;
+  virtual int merge_and_store_attrs(const DoutPrefixProvider* dpp,
+				    Attrs& new_attrs, optional_yield y) override;
+  virtual int remove_bucket(const DoutPrefixProvider* dpp, bool delete_children,
+			    bool forward_to_master, req_info* req_info,
+			    optional_yield y) override;
+  virtual int remove_bucket_bypass_gc(int concurrent_max,
+				      bool keep_index_consistent,
+				      optional_yield y,
+				      const DoutPrefixProvider *dpp) override;
+  virtual int load_bucket(const DoutPrefixProvider* dpp, optional_yield y,
+			  bool get_stats = false) override;
+  virtual RGWAccessControlPolicy& get_acl(void) override { return acls; }
+  virtual int set_acl(const DoutPrefixProvider* dpp, RGWAccessControlPolicy& acl,
+		      optional_yield y) override;
+  virtual int read_stats(const DoutPrefixProvider *dpp,
+			 const bucket_index_layout_generation& idx_layout,
+			 int shard_id, std::string* bucket_ver, std::string* master_ver,
+			 std::map<RGWObjCategory, RGWStorageStats>& stats,
+			 std::string* max_marker = nullptr,
+			 bool* syncstopped = nullptr) override;
+  virtual int read_stats_async(const DoutPrefixProvider *dpp,
+			       const bucket_index_layout_generation& idx_layout,
+			       int shard_id, RGWGetBucketStats_CB* ctx) override;
+  virtual int sync_user_stats(const DoutPrefixProvider *dpp, optional_yield y) override;
+  virtual int update_container_stats(const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual int check_bucket_shards(const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual int chown(const DoutPrefixProvider* dpp, User& new_user, optional_yield y) override;
+  virtual int put_info(const DoutPrefixProvider* dpp, bool exclusive,
+		       ceph::real_time mtime, optional_yield y) override;
+  virtual int check_empty(const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual int check_quota(const DoutPrefixProvider *dpp, RGWQuota& quota, uint64_t obj_size, optional_yield y, bool check_size_only = false) override;
+  virtual int try_refresh_info(const DoutPrefixProvider* dpp, ceph::real_time* pmtime, optional_yield y) override;
+  virtual int read_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch,
+			 uint64_t end_epoch, uint32_t max_entries, bool* is_truncated,
+			 RGWUsageIter& usage_iter, std::map<rgw_user_bucket, rgw_usage_log_entry>& usage) override;
+  virtual int trim_usage(const DoutPrefixProvider *dpp, uint64_t start_epoch, uint64_t end_epoch, optional_yield y) override;
+  virtual int remove_objs_from_index(const DoutPrefixProvider *dpp, std::list<rgw_obj_index_key>& objs_to_unlink) override;
+  virtual int check_index(const DoutPrefixProvider *dpp, std::map<RGWObjCategory, RGWStorageStats>& existing_stats, std::map<RGWObjCategory, RGWStorageStats>& calculated_stats) override;
+  virtual int rebuild_index(const DoutPrefixProvider *dpp) override;
+  virtual int set_tag_timeout(const DoutPrefixProvider *dpp, uint64_t timeout) override;
+  virtual int purge_instance(const DoutPrefixProvider* dpp, optional_yield y) override;
+
+  virtual std::unique_ptr<Bucket> clone() override {
+    return std::make_unique<POSIXBucket>(*this);
+  }
+
+  virtual std::unique_ptr<MultipartUpload> get_multipart_upload(
+				const std::string& oid,
+				std::optional<std::string> upload_id=std::nullopt,
+				ACLOwner owner={}, ceph::real_time mtime=real_clock::now()) override;
+  virtual int list_multiparts(const DoutPrefixProvider *dpp,
+			      const std::string& prefix,
+			      std::string& marker,
+			      const std::string& delim,
+			      const int& max_uploads,
+			      std::vector<std::unique_ptr<MultipartUpload>>& uploads,
+			      std::map<std::string, bool> *common_prefixes,
+			      bool *is_truncated, optional_yield y) override;
+  virtual int abort_multiparts(const DoutPrefixProvider* dpp,
+			       CephContext* cct, optional_yield y) override;
+
+  /* Internal APIs */
+  int create(const DoutPrefixProvider *dpp, optional_yield y, bool* existed);
+  void set_stat(struct statx _stx) { stx = _stx; stat_done = true; }
+  int get_dir_fd(const DoutPrefixProvider *dpp) { open(dpp); return dir_fd; }
+  /* TODO dang Escape the bucket name for file use */
+  std::string get_fname();
+  int get_shadow_bucket(const DoutPrefixProvider* dpp, optional_yield y,
+			const std::string& ns, const std::string& tenant,
+			const std::string& name, bool create,
+			std::unique_ptr<POSIXBucket>* shadow);
+  template <typename F>
+    int for_each(const DoutPrefixProvider* dpp, const F& func);
+  int open(const DoutPrefixProvider *dpp);
+  int close();
+  int rename(const DoutPrefixProvider* dpp, optional_yield y, Object* target_obj);
+  int copy(const DoutPrefixProvider *dpp, optional_yield y, POSIXBucket* db, POSIXObject* dobj);
+
+  /* integration w/bucket listing cache */
+  using fill_cache_cb_t = file::listing::fill_cache_cb_t;
+
+  /* enumerate all entries by callback, in any order */
+  int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t cb);
+  
+private:
+  int stat(const DoutPrefixProvider *dpp);
+  int write_attrs(const DoutPrefixProvider *dpp, optional_yield y);
+}; /* POSIXBucket */
+
+class POSIXObject : public StoreObject {
+private:
+  POSIXDriver* driver;
+  RGWAccessControlPolicy acls;
+  int obj_fd{-1};
+  struct statx stx;
+  bool stat_done{false};
+  std::unique_ptr<rgw::sal::POSIXBucket> shadow;
+  std::string temp_fname;
+  std::map<std::string, int64_t> parts;
+
+public:
+  struct POSIXReadOp : ReadOp {
+    POSIXObject* source;
+
+    POSIXReadOp(POSIXObject* _source) :
+      source(_source) {}
+    virtual ~POSIXReadOp() = default;
+
+    virtual int prepare(optional_yield y, const DoutPrefixProvider* dpp) override;
+    virtual int read(int64_t ofs, int64_t left, bufferlist& bl, optional_yield y,
+		     const DoutPrefixProvider* dpp) override;
+    virtual int iterate(const DoutPrefixProvider* dpp, int64_t ofs, int64_t end,
+			RGWGetDataCB* cb, optional_yield y) override;
+    virtual int get_attr(const DoutPrefixProvider* dpp, const char* name,
+			 bufferlist& dest, optional_yield y) override;
+  };
+
+  struct POSIXDeleteOp : DeleteOp {
+    POSIXObject* source;
+
+    POSIXDeleteOp(POSIXObject* _source) :
+      source(_source) {}
+    virtual ~POSIXDeleteOp() = default;
+
+    virtual int delete_obj(const DoutPrefixProvider* dpp, optional_yield y) override;
+  };
+
+  POSIXObject(POSIXDriver *_dr, const rgw_obj_key& _k)
+    : StoreObject(_k),
+    driver(_dr),
+    acls() {}
+
+  POSIXObject(POSIXDriver* _driver, const rgw_obj_key& _k, Bucket* _b) :
+    StoreObject(_k, _b),
+    driver(_driver),
+    acls() {}
+
+  POSIXObject(const POSIXObject& _o) :
+    StoreObject(_o),
+    driver(_o.driver) {}
+
+  virtual ~POSIXObject() { close(); }
+
+  virtual int delete_object(const DoutPrefixProvider* dpp,
+			    optional_yield y,
+			    bool prevent_versioning = false) override;
+  virtual int copy_object(User* user,
+               req_info* info, const rgw_zone_id& source_zone,
+               rgw::sal::Object* dest_object, rgw::sal::Bucket* dest_bucket,
+               rgw::sal::Bucket* src_bucket,
+               const rgw_placement_rule& dest_placement,
+               ceph::real_time* src_mtime, ceph::real_time* mtime,
+               const ceph::real_time* mod_ptr, const ceph::real_time* unmod_ptr,
+               bool high_precision_time,
+               const char* if_match, const char* if_nomatch,
+               AttrsMod attrs_mod, bool copy_if_newer, Attrs& attrs,
+               RGWObjCategory category, uint64_t olh_epoch,
+               boost::optional<ceph::real_time> delete_at,
+               std::string* version_id, std::string* tag, std::string* etag,
+               void (*progress_cb)(off_t, void *), void* progress_data,
+               const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual RGWAccessControlPolicy& get_acl(void) override { return acls; }
+  virtual int set_acl(const RGWAccessControlPolicy& acl) override { acls = acl; return 0; }
+  virtual int get_obj_state(const DoutPrefixProvider* dpp, RGWObjState **state, optional_yield y, bool follow_olh = true) override;
+  virtual int set_obj_attrs(const DoutPrefixProvider* dpp, Attrs* setattrs,
+			    Attrs* delattrs, optional_yield y) override;
+  virtual int get_obj_attrs(optional_yield y, const DoutPrefixProvider* dpp,
+			    rgw_obj* target_obj = NULL) override;
+  virtual int modify_obj_attrs(const char* attr_name, bufferlist& attr_val,
+			       optional_yield y, const DoutPrefixProvider* dpp) override;
+  virtual int delete_obj_attrs(const DoutPrefixProvider* dpp, const char* attr_name,
+			       optional_yield y) override;
+  virtual bool is_expired() override;
+  virtual void gen_rand_obj_instance_name() override;
+  virtual std::unique_ptr<MPSerializer> get_serializer(const DoutPrefixProvider *dpp,
+						       const std::string& lock_name) override;
+  virtual int transition(Bucket* bucket,
+			 const rgw_placement_rule& placement_rule,
+			 const real_time& mtime,
+			 uint64_t olh_epoch,
+			 const DoutPrefixProvider* dpp,
+			 optional_yield y) override;
+  virtual int transition_to_cloud(Bucket* bucket,
+			 rgw::sal::PlacementTier* tier,
+			 rgw_bucket_dir_entry& o,
+			 std::set<std::string>& cloud_targets,
+			 CephContext* cct,
+			 bool update_object,
+			 const DoutPrefixProvider* dpp,
+			 optional_yield y) override;
+  virtual bool placement_rules_match(rgw_placement_rule& r1, rgw_placement_rule& r2) override;
+  virtual int dump_obj_layout(const DoutPrefixProvider *dpp, optional_yield y, Formatter* f) override;
+  virtual int swift_versioning_restore(bool& restored,
+				       const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual int swift_versioning_copy(const DoutPrefixProvider* dpp,
+				    optional_yield y) override;
+  virtual std::unique_ptr<ReadOp> get_read_op() override;
+  virtual std::unique_ptr<DeleteOp> get_delete_op() override;
+  virtual int omap_get_vals_by_keys(const DoutPrefixProvider *dpp, const std::string& oid,
+				    const std::set<std::string>& keys,
+				    Attrs* vals) override;
+  virtual int omap_set_val_by_key(const DoutPrefixProvider *dpp, const std::string& key,
+				  bufferlist& val, bool must_exist, optional_yield y) override;
+  virtual int chown(User& new_user, const DoutPrefixProvider* dpp, optional_yield y) override;
+  virtual std::unique_ptr<Object> clone() override {
+    return std::unique_ptr<Object>(new POSIXObject(*this));
+  }
+
+  int open(const DoutPrefixProvider *dpp, bool create, bool temp_file = false);
+  int close();
+  int write(int64_t ofs, bufferlist& bl, const DoutPrefixProvider* dpp, optional_yield y);
+  int write_attr(const DoutPrefixProvider* dpp, optional_yield y, const std::string& key, bufferlist& value);
+  int link_temp_file(const DoutPrefixProvider* dpp, optional_yield y);
+  void gen_temp_fname();
+  /* TODO dang Escape the object name for file use */
+  const std::string get_fname();
+  bool exists(const DoutPrefixProvider* dpp) { stat(dpp); return state.exists; }
+  int get_owner(const DoutPrefixProvider *dpp, optional_yield y, std::unique_ptr<User> *owner);
+  int copy(const DoutPrefixProvider *dpp, optional_yield y, POSIXBucket *sb,
+           POSIXBucket *db, POSIXObject *dobj);
+  int fill_bde(const DoutPrefixProvider *dpp, optional_yield y, rgw_bucket_dir_entry &bde);
+
+protected:
+  int read(int64_t ofs, int64_t end, bufferlist& bl, const DoutPrefixProvider* dpp, optional_yield y);
+  int generate_attrs(const DoutPrefixProvider* dpp, optional_yield y);
+  int get_fd() { return obj_fd; };
+private:
+  const std::string get_temp_fname();
+  int stat(const DoutPrefixProvider *dpp);
+  int generate_mp_etag(const DoutPrefixProvider* dpp, optional_yield y);
+  int generate_etag(const DoutPrefixProvider* dpp, optional_yield y);
+};
+
+struct POSIXMPObj {
+  std::string oid;
+  std::string upload_id;
+  ACLOwner owner;
+  multipart_upload_info upload_info;
+  std::string meta;
+
+  POSIXMPObj(POSIXDriver* driver, const std::string& _oid,
+	     std::optional<std::string> _upload_id, ACLOwner& _owner) {
+    if (_upload_id && !_upload_id->empty()) {
+      init(_oid, *_upload_id, _owner);
+    } else if (!from_meta(_oid, _owner)) {
+      init_gen(driver, _oid, _owner);
+    }
+  }
+  void init(const std::string& _oid, const std::string& _upload_id, ACLOwner& _owner) {
+    if (_oid.empty()) {
+      clear();
+      return;
+    }
+    oid = _oid;
+    upload_id = _upload_id;
+    owner = _owner;
+    meta = oid;
+    if (!upload_id.empty())
+      meta += "." + upload_id;
+  }
+  void init_gen(POSIXDriver* driver, const std::string& _oid, ACLOwner& _owner);
+  bool from_meta(const std::string& meta, ACLOwner& _owner) {
+    int end_pos = meta.length();
+    int mid_pos = meta.rfind('.', end_pos - 1); // <key>.<upload_id>
+    if (mid_pos < 0)
+      return false;
+    oid = meta.substr(0, mid_pos);
+    upload_id = meta.substr(mid_pos + 1, end_pos - mid_pos - 1);
+    init(oid, upload_id, _owner);
+    return true;
+  }
+  void clear() {
+    oid = "";
+    meta = "";
+    upload_id = "";
+  }
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(oid, bl);
+    encode(upload_id, bl);
+    encode(owner, bl);
+    encode(upload_info, bl);
+    encode(meta, bl);
+    ENCODE_FINISH(bl);
+  }
+
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(oid, bl);
+    decode(upload_id, bl);
+    decode(owner, bl);
+    decode(upload_info, bl);
+    decode(meta, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(POSIXMPObj)
+
+struct POSIXUploadPartInfo {
+  uint32_t num{0};
+  std::string etag;
+  ceph::real_time mtime;
+
+  void encode(bufferlist& bl) const {
+    ENCODE_START(1, 1, bl);
+    encode(num, bl);
+    encode(etag, bl);
+    encode(mtime, bl);
+    ENCODE_FINISH(bl);
+  }
+  void decode(bufferlist::const_iterator& bl) {
+    DECODE_START(1, bl);
+    decode(num, bl);
+    decode(etag, bl);
+    decode(mtime, bl);
+    DECODE_FINISH(bl);
+  }
+};
+WRITE_CLASS_ENCODER(POSIXUploadPartInfo)
+
+class POSIXMultipartUpload;
+
+class POSIXMultipartPart : public StoreMultipartPart {
+protected:
+  POSIXUploadPartInfo info;
+  POSIXMultipartUpload* upload;
+  std::unique_ptr<rgw::sal::POSIXObject> shadow;
+
+public:
+  POSIXMultipartPart(POSIXMultipartUpload* _upload) :
+    upload(_upload) {}
+  virtual ~POSIXMultipartPart() = default;
+
+  virtual uint32_t get_num() { return info.num; }
+  virtual uint64_t get_size() { return shadow->get_obj_size(); }
+  virtual const std::string& get_etag() { return info.etag; }
+  virtual ceph::real_time& get_mtime() { return info.mtime; }
+
+  int load(const DoutPrefixProvider* dpp, optional_yield y, POSIXDriver* driver, rgw_obj_key& key);
+
+  friend class POSIXMultipartUpload;
+};
+
+class POSIXMultipartUpload : public StoreMultipartUpload {
+protected:
+  POSIXDriver* driver;
+  POSIXMPObj mp_obj;
+  ceph::real_time mtime;
+  std::unique_ptr<rgw::sal::POSIXBucket> shadow;
+
+public:
+  POSIXMultipartUpload(POSIXDriver* _driver, Bucket* _bucket, const std::string& _oid,
+		       std::optional<std::string> _upload_id, ACLOwner _owner,
+		       ceph::real_time _mtime) :
+    StoreMultipartUpload(_bucket), driver(_driver),
+    mp_obj(driver, _oid, _upload_id, _owner), mtime(_mtime) {}
+  virtual ~POSIXMultipartUpload() = default;
+
+  virtual const std::string& get_meta() const override { return mp_obj.meta; }
+  virtual const std::string& get_key() const override { return mp_obj.oid; }
+  virtual const std::string& get_upload_id() const override { return mp_obj.upload_id; }
+  virtual const ACLOwner& get_owner() const override { return mp_obj.owner; }
+  virtual ceph::real_time& get_mtime() override { return mtime; }
+  virtual std::unique_ptr<rgw::sal::Object> get_meta_obj() override;
+
+  virtual int init(const DoutPrefixProvider* dpp, optional_yield y, ACLOwner& owner, rgw_placement_rule& dest_placement, rgw::sal::Attrs& attrs) override;
+  virtual int list_parts(const DoutPrefixProvider* dpp, CephContext* cct,
+			 int num_parts, int marker,
+			 int* next_marker, bool* truncated, optional_yield y,
+			 bool assume_unsorted = false) override;
+  virtual int abort(const DoutPrefixProvider* dpp, CephContext* cct, optional_yield y) override;
+  virtual int complete(const DoutPrefixProvider* dpp,
+		       optional_yield y, CephContext* cct,
+		       std::map<int, std::string>& part_etags,
+		       std::list<rgw_obj_index_key>& remove_objs,
+		       uint64_t& accounted_size, bool& compressed,
+		       RGWCompressionInfo& cs_info, off_t& ofs,
+		       std::string& tag, ACLOwner& owner,
+		       uint64_t olh_epoch,
+		       rgw::sal::Object* target_obj) override;
+  virtual int get_info(const DoutPrefixProvider *dpp, optional_yield y,
+		       rgw_placement_rule** rule, rgw::sal::Attrs* attrs) override;
+
+  virtual std::unique_ptr<Writer> get_writer(const DoutPrefixProvider *dpp,
+			  optional_yield y,
+			  rgw::sal::Object* _head_obj,
+			  const rgw_user& owner,
+			  const rgw_placement_rule *ptail_placement_rule,
+			  uint64_t part_num,
+			  const std::string& part_num_str) override;
+
+  POSIXBucket* get_shadow() { return shadow.get(); }
+private:
+  int load(bool create=false);
+};
+
+class POSIXAtomicWriter : public StoreWriter {
+private:
+  POSIXDriver* driver;
+  const rgw_user& owner;
+  const rgw_placement_rule *ptail_placement_rule;
+  uint64_t olh_epoch;
+  const std::string& unique_tag;
+  POSIXObject obj;
+
+public:
+  POSIXAtomicWriter(const DoutPrefixProvider *dpp,
+                    optional_yield y,
+		    rgw::sal::Object* _head_obj,
+                    POSIXDriver* _driver,
+                    const rgw_user& _owner,
+                    const rgw_placement_rule *_ptail_placement_rule,
+                    uint64_t _olh_epoch,
+                    const std::string& _unique_tag) :
+    StoreWriter(dpp, y),
+    driver(_driver),
+    owner(_owner),
+    ptail_placement_rule(_ptail_placement_rule),
+    olh_epoch(_olh_epoch),
+    unique_tag(_unique_tag),
+    obj(_driver, _head_obj->get_key(), _head_obj->get_bucket()) {}
+  virtual ~POSIXAtomicWriter() = default;
+
+  virtual int prepare(optional_yield y) override;
+  virtual int process(bufferlist&& data, uint64_t offset) override;
+  virtual int complete(size_t accounted_size, const std::string& etag,
+                       ceph::real_time *mtime, ceph::real_time set_mtime,
+		       std::map<std::string, bufferlist>& attrs,
+		       ceph::real_time delete_at,
+		       const char *if_match, const char *if_nomatch,
+		       const std::string *user_data,
+		       rgw_zone_set *zones_trace, bool *canceled,
+		       optional_yield y) override;
+};
+
+class POSIXMultipartWriter : public StoreWriter {
+private:
+  POSIXDriver* driver;
+  const rgw_user& owner;
+  const rgw_placement_rule *ptail_placement_rule;
+  uint64_t part_num;
+  std::unique_ptr<Bucket> shadow_bucket;
+  std::unique_ptr<POSIXObject> obj;
+
+public:
+  POSIXMultipartWriter(const DoutPrefixProvider *dpp,
+                    optional_yield y,
+		    std::unique_ptr<Bucket> _shadow_bucket,
+                    rgw_obj_key& _key,
+                    POSIXDriver* _driver,
+                    const rgw_user& _owner,
+                    const rgw_placement_rule *_ptail_placement_rule,
+                    uint64_t _part_num) :
+    StoreWriter(dpp, y),
+    driver(_driver),
+    owner(_owner),
+    ptail_placement_rule(_ptail_placement_rule),
+    part_num(_part_num),
+    shadow_bucket(std::move(_shadow_bucket)),
+    obj(std::make_unique<POSIXObject>(_driver, _key, shadow_bucket.get())) {}
+  virtual ~POSIXMultipartWriter() = default;
+
+  virtual int prepare(optional_yield y) override;
+  virtual int process(bufferlist&& data, uint64_t offset) override;
+  virtual int complete(size_t accounted_size, const std::string& etag,
+                       ceph::real_time *mtime, ceph::real_time set_mtime,
+		       std::map<std::string, bufferlist>& attrs,
+		       ceph::real_time delete_at,
+		       const char *if_match, const char *if_nomatch,
+		       const std::string *user_data,
+		       rgw_zone_set *zones_trace, bool *canceled,
+		       optional_yield y) override;
+
+};
+
+class MPPOSIXSerializer : public StoreMPSerializer {
+  POSIXObject* obj;
+
+public:
+  MPPOSIXSerializer(const DoutPrefixProvider *dpp, POSIXDriver* driver, POSIXObject* _obj, const std::string& lock_name) : obj(_obj) {}
+
+  virtual int try_lock(const DoutPrefixProvider *dpp, utime_t dur, optional_yield y) override;
+  virtual int unlock() override { return 0; }
+};
+
+} } // namespace rgw::sal

src/rgw/driver/rados/rgw_sal_rados.cc

@@ -774,12 +774,6 @@ int RadosBucket::put_info(const DoutPrefixProvider* dpp, bool exclusive, ceph::r
   return store->getRados()->put_bucket_instance_info(info, exclusive, mtime, &attrs, dpp, y);
 }
 
-/* Make sure to call get_bucket_info() if you need it first */
-bool RadosBucket::is_owner(User* user)
-{
-  return (info.owner.compare(user->get_id()) == 0);
-}
-
 int RadosBucket::check_empty(const DoutPrefixProvider* dpp, optional_yield y)
 {
   return store->getRados()->check_bucket_empty(dpp, info, y);

src/rgw/driver/rados/rgw_sal_rados.h

@@ -571,7 +571,6 @@ class RadosBucket : public StoreBucket {
     virtual int check_bucket_shards(const DoutPrefixProvider* dpp, optional_yield y) override;
     virtual int chown(const DoutPrefixProvider* dpp, User& new_user, optional_yield y) override;
     virtual int put_info(const DoutPrefixProvider* dpp, bool exclusive, ceph::real_time mtime, optional_yield y) override;
-    virtual bool is_owner(User* user) override;
     virtual int check_empty(const DoutPrefixProvider* dpp, optional_yield y) override;
     virtual int check_quota(const DoutPrefixProvider *dpp, RGWQuota& quota, uint64_t obj_size, optional_yield y, bool check_size_only = false) override;
     virtual int merge_and_store_attrs(const DoutPrefixProvider* dpp, Attrs& attrs, optional_yield y) override;

src/rgw/rgw_op.cc

@@ -3355,7 +3355,7 @@ void RGWCreateBucket::execute(optional_yield y)
     }
   }
 
-  s->bucket_owner.set_id(s->user->get_id()); /* XXX dang use s->bucket->owner */
+  s->bucket_owner.set_id(s->user->get_id());
   s->bucket_owner.set_name(s->user->get_display_name());
 
   string zonegroup_id;
@@ -4639,7 +4639,7 @@ void RGWPostObj::execute(optional_yield y)
       cs_info.compression_type = plugin->get_type_name();
       cs_info.orig_size = s->obj_size;
       cs_info.compressor_message = compressor->get_compressor_message();
-      cs_info.blocks = move(compressor->get_compression_blocks());
+      cs_info.blocks = std::move(compressor->get_compression_blocks());
       encode(cs_info, tmp);
       emplace_attr(RGW_ATTR_COMPRESSION, std::move(tmp));
     }
@@ -4896,7 +4896,6 @@ void RGWPutMetadataObject::pre_exec()
 
 void RGWPutMetadataObject::execute(optional_yield y)
 {
-  rgw_obj target_obj;
   rgw::sal::Attrs attrs, rmattrs;
 
   s->object->set_atomic();
@@ -4912,7 +4911,7 @@ void RGWPutMetadataObject::execute(optional_yield y)
   }
 
   /* check if obj exists, read orig attrs */
-  op_ret = s->object->get_obj_attrs(s->yield, s, &target_obj);
+  op_ret = s->object->get_obj_attrs(s->yield, s);
   if (op_ret < 0) {
     return;
   }

src/rgw/rgw_rest_s3.cc

@@ -589,7 +589,7 @@ send_data:
 
 int RGWGetObj_ObjStore_S3::get_decrypt_filter(std::unique_ptr<RGWGetObj_Filter> *filter, RGWGetObj_Filter* cb, bufferlist* manifest_bl)
 {
-  if (skip_decrypt) { // bypass decryption for multisite sync requests
+  if (skip_decrypt || !manifest_bl) { // bypass decryption for multisite sync requests
     return 0;
   }
 

src/rgw/rgw_sal.cc

@@ -56,6 +56,9 @@ extern rgw::sal::Driver* newMotrStore(CephContext *cct);
 #ifdef WITH_RADOSGW_DAOS
 extern rgw::sal::Driver* newDaosStore(CephContext *cct);
 #endif
+#ifdef WITH_RADOSGW_POSIX
+extern rgw::sal::Driver* newPOSIXDriver(rgw::sal::Driver* next);
+#endif
 extern rgw::sal::Driver* newBaseFilter(rgw::sal::Driver* next);
 #ifdef WITH_RADOSGW_D4N
 extern rgw::sal::Driver* newD4NFilter(rgw::sal::Driver* next);
@@ -217,6 +220,7 @@ rgw::sal::Driver* DriverManager::init_storage_provider(const DoutPrefixProvider*
     }
   }
 #endif
+  ldpp_dout(dpp, 20) << "Filter name: " << cfg.filter_name << dendl;
 
   if (cfg.filter_name.compare("base") == 0) {
     rgw::sal::Driver* next = driver;
@@ -240,6 +244,19 @@ rgw::sal::Driver* DriverManager::init_storage_provider(const DoutPrefixProvider*
     }
   }
 #endif
+#ifdef WITH_RADOSGW_POSIX
+  else if (cfg.filter_name.compare("posix") == 0) {
+    rgw::sal::Driver* next = driver;
+    ldpp_dout(dpp, 20) << "Creating POSIX driver" << dendl;
+    driver = newPOSIXDriver(next);
+
+    if (driver->initialize(cct, dpp) < 0) {
+      delete driver;
+      delete next;
+      return nullptr;
+    }
+  }
+#endif
 
   return driver;
 }
@@ -365,6 +382,8 @@ DriverManager::Config DriverManager::get_config(bool admin, CephContext* cct)
   const auto& config_filter = g_conf().get_val<std::string>("rgw_filter");
   if (config_filter == "base") {
     cfg.filter_name = "base";
+  } else if (config_filter == "posix") {
+    cfg.filter_name = "posix";
   }
 #ifdef WITH_RADOSGW_D4N
   else if (config_filter == "d4n") {

src/rgw/rgw_sal_dbstore.cc

@@ -339,12 +339,6 @@ namespace rgw::sal {
 
   }
 
-  /* Make sure to call get_bucket_info() if you need it first */
-  bool DBBucket::is_owner(User* user)
-  {
-    return (info.owner.compare(user->get_id()) == 0);
-  }
-
   int DBBucket::check_empty(const DoutPrefixProvider *dpp, optional_yield y)
   {
     /* XXX: Check if bucket contains any objects */
@@ -572,6 +566,9 @@ namespace rgw::sal {
 
   bool DBZone::has_zonegroup_api(const std::string& api) const
   {
+    if (api == "default")
+      return true;
+
     return false;
   }
 
@@ -1721,7 +1718,11 @@ namespace rgw::sal {
   int DBStore::get_zonegroup(const std::string& id, std::unique_ptr<ZoneGroup>* zg)
   {
     /* XXX: for now only one zonegroup supported */
-    ZoneGroup* group = new DBZoneGroup(this, std::make_unique<RGWZoneGroup>());
+    std::unique_ptr<RGWZoneGroup> rzg =
+        std::make_unique<RGWZoneGroup>("default", "default");
+    rzg->api_name = "default";
+    rzg->is_master = true;
+    ZoneGroup* group = new DBZoneGroup(this, std::move(rzg));
     if (!group)
       return -ENOMEM;
 

src/rgw/rgw_sal_dbstore.h

@@ -204,7 +204,6 @@ protected:
       virtual int check_bucket_shards(const DoutPrefixProvider *dpp, optional_yield y) override;
       virtual int chown(const DoutPrefixProvider *dpp, User& new_user, optional_yield y) override;
       virtual int put_info(const DoutPrefixProvider *dpp, bool exclusive, ceph::real_time mtime, optional_yield y) override;
-      virtual bool is_owner(User* user) override;
       virtual int check_empty(const DoutPrefixProvider *dpp, optional_yield y) override;
       virtual int check_quota(const DoutPrefixProvider *dpp, RGWQuota& quota, uint64_t obj_size, optional_yield y, bool check_size_only = false) override;
       virtual int merge_and_store_attrs(const DoutPrefixProvider *dpp, Attrs& attrs, optional_yield y) override;
@@ -318,7 +317,10 @@ protected:
     public:
       DBZone(DBStore* _store) : store(_store) {
 	realm = new RGWRealm();
-        zonegroup = new DBZoneGroup(store, std::make_unique<RGWZoneGroup>());
+	std::unique_ptr<RGWZoneGroup> rzg = std::make_unique<RGWZoneGroup>("default", "default");
+	rzg->api_name = "default";
+	rzg->is_master = true;
+        zonegroup = new DBZoneGroup(store, std::move(rzg));
         zone_public_config = new RGWZone();
         zone_params = new RGWZoneParams();
         current_period = new RGWPeriod();

src/rgw/rgw_sal_store.h

@@ -120,6 +120,7 @@ class StoreBucket : public Bucket {
     virtual int set_attrs(Attrs a) override { attrs = a; return 0; }
     virtual void set_owner(rgw::sal::User* _owner) override {
       owner = _owner;
+      info.owner = owner->get_id();
     }
     virtual void set_count(uint64_t _count) override {
       ent.count = _count;
@@ -128,6 +129,8 @@ class StoreBucket : public Bucket {
       ent.size = _size;
     }
     virtual User* get_owner(void) override { return owner; };
+    /* Make sure to call get_bucket_info() if you need it first */
+    virtual bool is_owner(User* user) override { return (info.owner.compare(user->get_id()) == 0); }
     virtual ACLOwner get_acl_owner(void) override { return ACLOwner(info.owner); };
     virtual bool empty() const override { return info.bucket.name.empty(); }
     virtual const std::string& get_name() const override { return info.bucket.name; }
@@ -176,6 +179,7 @@ class StoreBucket : public Bucket {
         optional_yield y, const DoutPrefixProvider *dpp) override {return 0;}
 
     friend class BucketList;
+
   protected:
     virtual void set_ent(RGWBucketEnt& _ent) { ent = _ent; info.bucket = ent.bucket; info.placement_rule = ent.placement_rule; }
 };

src/test/rgw/CMakeLists.txt

@@ -315,3 +315,14 @@ target_link_libraries(radosgw-cr-test ${rgw_libs} librados
   OATH::OATH
   ${CURL_LIBRARIES} ${EXPAT_LIBRARIES} ${BLKID_LIBRARIES}
   GTest::GTest)
+
+# unittest_posix_bucket_cache
+add_executable(unittest_posix_bucket_cache
+	test_posix_bucket_cache.cc)
+add_ceph_unittest(unittest_posix_bucket_cache)
+target_compile_definitions(unittest_posix_bucket_cache PUBLIC LMDB_SAFE_NO_CPP_UTILITIES)
+target_include_directories(unittest_posix_bucket_cache
+  SYSTEM PRIVATE "${CMAKE_SOURCE_DIR}/src/rgw"
+  SYSTEM PRIVATE "${CMAKE_SOURCE_DIR}/src/rgw/driver/posix")
+target_link_libraries(unittest_posix_bucket_cache  ${UNITTEST_LIBS}
+  ${rgw_libs} ${LMDB_LIBRARIES})

src/test/rgw/test_posix_bucket_cache.cc

@@ -0,0 +1,428 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab ft=cpp
+
+#include "bucket_cache.h"
+#include <iostream>
+#include <fstream>
+#include <filesystem>
+#include <string>
+#include <string_view>
+#include <random>
+#include <ranges>
+#include <thread>
+#include <stdint.h>
+
+#undef FMT_HEADER_ONLY
+#define FMT_HEADER_ONLY 1
+#include <fmt/format.h>
+
+#include <gtest/gtest.h>
+
+using namespace std::chrono_literals;
+
+namespace {
+
+  namespace sf = std::filesystem; 
+
+  std::string bucket_root = "bucket_root";
+  std::string database_root = "lmdb_root";
+  std::string bucket1_name = "stanley";
+  std::string bucket1_marker = ""; // start at the beginning
+
+  std::random_device rd;
+  std::mt19937 mt(rd());
+
+  DoutPrefixProvider* dpp{nullptr};
+  std::string tdir1{"tdir1"};
+  std::string tdir2{"tdir2"};
+  std::uniform_int_distribution<> dist_1m(1, 1000000);
+  std::vector<std::string> bvec;
+
+  class MockSalDriver
+  {
+  public:
+    /* called by BucketCache layer when a new object is discovered
+     * by inotify or similar */
+    int mint_listing_entry(
+      const std::string& bucket, rgw_bucket_dir_entry& bde /* OUT */) {
+
+      return 0;
+    }
+  }; /* MockSalDriver */
+
+  class MockSalBucket
+  {
+    std::string name;
+  public:
+    MockSalBucket(const std::string& name)
+      : name(name)
+      {}
+    const std::string& get_name() {
+      return name;
+    }
+
+    using fill_cache_cb_t = file::listing::fill_cache_cb_t;
+
+    int fill_cache(const DoutPrefixProvider* dpp, optional_yield y, fill_cache_cb_t cb) {
+      sf::path rp{bucket_root};
+      sf::path bp{rp / name};
+      if (! (sf::exists(rp) && sf::is_directory(rp))) {
+	std::cerr << fmt::format("{} bucket {} invalid", __func__, name)
+		  << std::endl;
+	exit(1);
+      }
+      for (const auto& dir_entry : sf::directory_iterator{bp}) {
+	rgw_bucket_dir_entry bde{};
+	auto fname = dir_entry.path().filename().string();
+	bde.key.name = fname;
+	cb(dpp, bde);
+      }
+      return 0;
+    } /* fill_cache */
+
+  }; /* MockSalBucket */
+
+  using BucketCache = file::listing::BucketCache<MockSalDriver, MockSalBucket>;
+  MockSalDriver sal_driver;
+  BucketCache* bucket_cache{nullptr};
+
+} // anonymous ns
+
+namespace sf = std::filesystem;
+
+TEST(BucketCache, SetupTDir1)
+{
+  sf::path tp{sf::path{bucket_root} / tdir1};
+  sf::remove_all(tp);
+  sf::create_directory(tp);
+
+  /* generate 100K unique files in random order */
+  std::string fbase{"file_"};
+  for (int ix = 0; ix < 100000; ++ix) {
+  retry:
+    auto n = dist_1m(mt);
+    sf::path ttp{tp / fmt::format("{}{}", fbase, n)};
+    if (sf::exists(ttp)) {
+      goto retry;
+    } else {
+      std::ofstream ofs(ttp);
+      ofs << "data for " << ttp << std::endl;
+      ofs.close();
+    }
+  } /* for 100K */
+} /* SetupTDir1 */
+
+TEST(BucketCache, InitBucketCache)
+{
+  bucket_cache = new BucketCache{&sal_driver, bucket_root, database_root}; // default tuning
+}
+
+auto func = [](const rgw_bucket_dir_entry& bde) -> bool
+  {
+    //std::cout << fmt::format("called back with {}", bde.key.name) << std::endl;
+    return true;
+  };
+
+TEST(BucketCache, ListTDir1)
+{
+  MockSalBucket sb{tdir1};
+  std::string marker = bucket1_marker;
+  (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, func);
+}
+
+
+TEST(BucketCache, SetupEmpty)
+{
+  sf::path tp{sf::path{bucket_root} / tdir2};
+  sf::remove_all(tp);
+  sf::create_directory(tp);
+
+  /* generate no objects in tdir2 */
+
+} /* SetupEmpty */
+
+TEST(BucketCache, ListEmpty)
+{
+  MockSalBucket sb{tdir2};
+  std::string marker = bucket1_marker;
+  (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, func);
+}
+
+TEST(BucketCache, ListThreads) /* clocked at 21ms on lemon, and yes,
+				* it did list 100K entries per thread */
+{
+  auto nthreads = 15;
+  std::vector<std::thread> threads;
+
+  auto func = [](const rgw_bucket_dir_entry& bde) -> int
+    {
+      //std::cout << fmt::format("called back with {}", bde.key.name) << std::endl;
+      return 0;
+    };
+
+  MockSalBucket sb{tdir1};
+  std::string marker = bucket1_marker;
+
+  for (int ix = 0; ix < nthreads; ++ix) {
+    threads.push_back(std::thread([&]() {
+      (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, func);
+    }));
+  }
+  for (auto& t : threads) {
+    t.join();
+  }
+}
+
+TEST(BucketCache, SetupRecycle1)
+{
+  int nbuckets = 5;
+  int nfiles = 10;
+
+  bvec = [&]() {
+    std::vector<std::string> v;
+    for (int ix = 0; ix < nbuckets; ++ix) {
+      v.push_back(fmt::format("recyle_{}", ix));
+    }
+    return v;
+  }();
+
+  for (auto& bucket : bvec) {
+    sf::path tp{sf::path{bucket_root} / bucket};
+    sf::remove_all(tp);
+    sf::create_directory(tp);
+
+    std::string fbase{"file_"};
+    for (int ix = 0; ix < nfiles; ++ix) {
+    retry:
+      auto n = dist_1m(mt);
+      sf::path ttp{tp / fmt::format("{}{}", fbase, n)};
+      if (sf::exists(ttp)) {
+	goto retry;
+      } else {
+	std::ofstream ofs(ttp);
+	ofs << "data for " << ttp << std::endl;
+	ofs.close();
+      }
+    } /* for buckets */
+  }
+} /* SetupRecycle1 */
+
+TEST(BucketCache, InitBucketCacheRecycle1)
+{
+  bucket_cache = new BucketCache{&sal_driver, bucket_root, database_root, 1, 1, 1, 1};
+}
+
+TEST(BucketCache, ListNRecycle1)
+{
+  /* the effect is to allocate a Bucket cache entry once, then recycle n-1 times */
+  for (auto& bucket : bvec) {
+    MockSalBucket sb{bucket};
+    std::string marker = bucket1_marker;
+    (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, func);
+  }
+  ASSERT_EQ(bucket_cache->recycle_count, 4);
+}
+
+TEST(BucketCache, TearDownBucketCacheRecycle1)
+{
+  delete bucket_cache;
+  bucket_cache = nullptr;
+}
+
+TEST(BucketCache, InitBucketCacheRecyclePartitions1)
+{
+  bucket_cache = new BucketCache{&sal_driver, bucket_root, database_root, 1, 1, 5 /* max partitions */, 1};
+}
+
+TEST(BucketCache, ListNRecyclePartitions1)
+{
+  /* the effect is to allocate a Bucket cache entry once, then recycle
+   * n-1 times--in addition, 5 cache partitions are mapped to 1 lru
+   * lane--verifying independence */
+  for (auto& bucket : bvec) {
+    MockSalBucket sb{bucket};
+    std::string marker = bucket1_marker;
+    (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, func);
+  }
+  ASSERT_EQ(bucket_cache->recycle_count, 4);
+}
+
+TEST(BucketCache, TearDownBucketCacheRecyclePartitions1)
+{
+  delete bucket_cache;
+  bucket_cache = nullptr;
+}
+
+TEST(BucketCache, SetupMarker1)
+{
+  int nfiles = 20;
+  std::string bucket{"marker1"};
+
+  sf::path tp{sf::path{bucket_root} / bucket};
+  sf::remove_all(tp);
+  sf::create_directory(tp);
+
+  std::string fbase{"file_"};
+  for (int ix = 0; ix < nfiles; ++ix) {
+    sf::path ttp{tp / fmt::format("{}{}", fbase, ix)};
+    std::ofstream ofs(ttp);
+    ofs << "data for " << ttp << std::endl;
+    ofs.close();
+  }
+} /* SetupMarker1 */
+
+TEST(BucketCache, InitBucketCacheMarker1)
+{
+  bucket_cache = new BucketCache{&sal_driver, bucket_root, database_root};
+}
+
+TEST(BucketCache, ListMarker1)
+{
+  std::string bucket{"marker1"};
+  std::string marker{"file_18"}; // midpoint+1
+  std::vector<std::string> names;
+
+  auto f = [&](const rgw_bucket_dir_entry& bde) -> int {
+    //std::cout << fmt::format("called back with {}", bde.key.name) << std::endl;
+    names.push_back(bde.key.name);
+    return true;
+  };
+
+  MockSalBucket sb{bucket};
+  (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, f);
+
+  ASSERT_EQ(names.size(), 10);
+  ASSERT_EQ(*names.begin(), "file_18");
+  ASSERT_EQ(*names.rbegin(), "file_9");
+}
+
+TEST(BucketCache, TearDownBucketCacheMarker1)
+{
+  delete bucket_cache;
+  bucket_cache = nullptr;
+}
+
+TEST(BucketCache, SetupInotify1)
+{
+  int nfiles = 20;
+  std::string bucket{"inotify1"};
+
+  sf::path tp{sf::path{bucket_root} / bucket};
+  sf::remove_all(tp);
+  sf::create_directory(tp);
+
+  std::string fbase{"file_"};
+  for (int ix = 0; ix < nfiles; ++ix) {
+    sf::path ttp{tp / fmt::format("{}{}", fbase, ix)};
+    std::ofstream ofs(ttp);
+    ofs << "data for " << ttp << std::endl;
+    ofs.close();
+  }
+} /* SetupInotify1 */
+
+TEST(BucketCache, InitBucketCacheInotify1)
+{
+  bucket_cache = new BucketCache{&sal_driver, bucket_root, database_root};
+}
+
+TEST(BucketCache, ListInotify1)
+{
+  std::string bucket{"inotify1"};
+  std::string marker{""};
+  std::vector<std::string> names;
+
+  auto f = [&](const rgw_bucket_dir_entry& bde) -> int {
+    //std::cout << fmt::format("called back with {}", bde.key.name) << std::endl;
+    names.push_back(bde.key.name);
+    return true;
+  };
+
+  MockSalBucket sb{bucket};
+
+  (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, f);
+  ASSERT_EQ(names.size(), 20);
+} /* ListInotify1 */
+
+TEST(BucketCache, UpdateInotify1)
+{
+  int nfiles = 10;
+  std::string bucket{"inotify1"};
+
+  sf::path tp{sf::path{bucket_root} / bucket};
+
+  /* add some */
+  std::string fbase{"upfile_"};
+  for (int ix = 0; ix < nfiles; ++ix) {
+    sf::path ttp{tp / fmt::format("{}{}", fbase, ix)};
+    std::ofstream ofs(ttp);
+    ofs << "data for " << ttp << std::endl;
+    ofs.close();
+  }
+
+  /* remove some */
+  fbase = "file_";
+  for (int ix = 5; ix < 10; ++ix) {
+    sf::path ttp{tp / fmt::format("{}{}", fbase, ix)};
+    sf::remove(ttp);
+  }
+
+  /* this step is async, temporally consistent */
+  std::cout << "waiting 1000ms for cache sync" << std::endl;
+  std::this_thread::sleep_for(1000ms);
+} /* SetupInotify1 */
+
+TEST(BucketCache, List2Inotify1)
+{
+  std::string bucket{"inotify1"};
+  std::string marker{""};
+  std::vector<std::string> names;
+
+  auto f = [&](const rgw_bucket_dir_entry& bde) -> int {
+    //std::cout << fmt::format("called back with {}", bde.key.name) << std::endl;
+    names.push_back(bde.key.name);
+    return true;
+  };
+
+  MockSalBucket sb{bucket};
+  (void) bucket_cache->list_bucket(dpp, null_yield, &sb, marker, f);
+  ASSERT_EQ(names.size(), 25);
+
+  /* check these */
+  sf::path tp{sf::path{bucket_root} / bucket};
+
+  std::string fbase{"upfile_"};
+  for (int ix = 0; ix < 10; ++ix) {
+    sf::path ttp{tp / fmt::format("{}{}", fbase, ix)};
+    ASSERT_TRUE(sf::exists(ttp));
+  }
+
+  /* remove some */
+  fbase = "file_";
+  for (int ix = 5; ix < 10; ++ix) {
+    sf::path ttp{tp / fmt::format("{}{}", fbase, ix)};
+    ASSERT_FALSE(sf::exists(ttp));
+  }
+} /* List2Inotify1 */
+
+TEST(BucketCache, TearDownInotify1)
+{
+  delete bucket_cache;
+  bucket_cache = nullptr;
+}
+
+int main (int argc, char *argv[])
+{
+
+  sf::path br{sf::path{bucket_root}};
+  sf::create_directory(br);
+  sf::remove_all(br);
+  sf::create_directory(br);
+
+  sf::path lr{sf::path{database_root}};
+  sf::create_directory(lr);
+  sf::remove_all(lr);
+  sf::create_directory(lr);
+
+  ::testing::InitGoogleTest(&argc, argv);
+  return RUN_ALL_TESTS();
+}