From: Abutalib Aghayev <agayev@cs.cmu.edu>
Date: Mon, 18 May 2020 21:08:25 +0000 (-0400)
Subject: os/bluestore: Enable writing data to HM-SMR hard drives.
X-Git-Tag: v17.0.0~2260^2
X-Git-Url: http://git.apps.os.sepia.ceph.com/?a=commitdiff_plain;h=ab865ca002c9ff9b2f2b968b4bfa8acf406f7fa5;p=ceph.git

os/bluestore: Enable writing data to HM-SMR hard drives.

Signed-off-by: Abutalib Aghayev <agayev@cs.cmu.edu>
---

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 064d3b94f1c3e..aef81c3e428c1 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -180,6 +180,8 @@ if(WITH_BLUESTORE)
   if(LINUX)
     find_package(aio)
     set(HAVE_LIBAIO ${AIO_FOUND})
+    find_package(zbc)
+    set(HAVE_LIBZBC ${ZBC_FOUND})
   elseif(FREEBSD)
     # POSIX AIO is integrated into FreeBSD kernel, and exposed by libc.
     set(HAVE_POSIXAIO ON)
diff --git a/cmake/modules/Findzbc.cmake b/cmake/modules/Findzbc.cmake
new file mode 100644
index 0000000000000..604981630810e
--- /dev/null
+++ b/cmake/modules/Findzbc.cmake
@@ -0,0 +1,19 @@
+# - Find ZBC
+#
+# ZBC_INCLUDE - Where to find zbc.h
+# ZBC_LIBRARIES - List of libraries when using zbc.
+# ZBC_FOUND - True if zbc found.
+
+find_path(ZBC_INCLUDE_DIR
+  zbc.h
+  HINTS $ENV{ZBC_ROOT}/libzbc
+  PATH_SUFFIXES libzbc)
+
+find_library(ZBC_LIBRARIES
+  zbc
+  HINTS $ENV{ZBC_ROOT}/lib)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(zbc DEFAULT_MSG ZBC_LIBRARIES ZBC_INCLUDE_DIR)
+
+mark_as_advanced(ZBC_INCLUDE_DIR ZBC_LIBRARIES)
diff --git a/doc/dev/zoned-storage.rst b/doc/dev/zoned-storage.rst
new file mode 100644
index 0000000000000..8fb2d24d297b6
--- /dev/null
+++ b/doc/dev/zoned-storage.rst
@@ -0,0 +1,104 @@
+=======================
+ Zoned Storage Support
+=======================
+
+http://zonedstorage.io
+
+Zoned Storage is a class of storage devices that enables host and storage
+devices to cooperate to achieve higher storage capacities, increased throughput,
+and lower latencies. The zoned storage interface is available through the SCSI
+Zoned Block Commands (ZBC) and Zoned Device ATA Command Set (ZAC) standards on
+Shingled Magnetic Recording (SMR) hard disks today and is also being adopted for
+NVMe Solid State Disks with the upcoming NVMe Zoned Namespaces (ZNS) standard.
+
+This project aims to enable Ceph to work on zoned storage drives and at the same
+time explore research problems related to adopting this new interface.  The
+first target is to enable non-ovewrite workloads (e.g. RGW) on host-managed SMR
+(HM-SMR) drives and explore cleaning (garbage collection) policies.  HM-SMR
+drives are high capacity hard drives with the ZBC/ZAC interface.  The longer
+term goal is to support ZNS SSDs, as they become available, as well as overwrite
+workloads.
+
+The first patch in these series enables writing data to HM-SMR drives.  Upcoming
+patches will introduce a rudimentary cleaner to form a baseline for further
+research.  Currently we can perform basic RADOS benchmarks on HM-SMR drives, as
+can be seen below.
+
+Please contact Abutalib Aghayev <agayev@cs.cmu.edu> for questions.
+
+::
+   
+  $ zbc_info /dev/sdc
+  Device /dev/sdc:
+      Vendor ID: ATA HGST HSH721414AL T240
+      Zoned block device interface, Host-managed zone model
+      27344764928 512-bytes sectors
+      3418095616 logical blocks of 4096 B
+      3418095616 physical blocks of 4096 B
+      14000.520 GB capacity
+      Read commands are unrestricted
+      672 KiB max R/W size
+      Maximum number of open sequential write required zones: 128
+  $ MON=1 OSD=1 MDS=0 sudo ../src/vstart.sh --new --localhost --bluestore --bluestore-devs /dev/sdc --bluestore-zoned
+  <snipped verbose output>
+  $ sudo ./bin/ceph osd pool create bench 32 32
+  pool 'bench' created
+  $ sudo ./bin/rados bench -p bench 10 write --no-cleanup
+  hints = 1
+  Maintaining 16 concurrent writes of 4194304 bytes to objects of size 4194304 for up to 10 seconds or 0 objects
+  Object prefix: benchmark_data_h0.cc.journaling712.narwhal.p_29846
+    sec Cur ops   started  finished  avg MB/s  cur MB/s last lat(s)  avg lat(s)
+      0       0         0         0         0         0           -           0
+      1      16        45        29   115.943       116    0.384175    0.407806
+      2      16        86        70   139.949       164    0.259845    0.391488
+      3      16       125       109   145.286       156     0.31727    0.404727
+      4      16       162       146   145.953       148    0.826671    0.409003
+      5      16       203       187   149.553       164     0.44815    0.404303
+      6      16       242       226   150.621       156    0.227488    0.409872
+      7      16       281       265   151.384       156    0.411896    0.408686
+      8      16       320       304   151.956       156    0.435135    0.411473
+      9      16       359       343   152.401       156    0.463699    0.408658
+     10      15       396       381   152.356       152    0.409554    0.410851
+  Total time run:         10.3305
+  Total writes made:      396
+  Write size:             4194304
+  Object size:            4194304
+  Bandwidth (MB/sec):     153.333
+  Stddev Bandwidth:       13.6561
+  Max bandwidth (MB/sec): 164
+  Min bandwidth (MB/sec): 116
+  Average IOPS:           38
+  Stddev IOPS:            3.41402
+  Max IOPS:               41
+  Min IOPS:               29
+  Average Latency(s):     0.411226
+  Stddev Latency(s):      0.180238
+  Max latency(s):         1.00844
+  Min latency(s):         0.108616
+  $ sudo ./bin/rados bench -p bench 10 rand
+  hints = 1
+    sec Cur ops   started  finished  avg MB/s  cur MB/s last lat(s)  avg lat(s)
+      0       0         0         0         0         0           -           0
+      1      16        61        45   179.903       180    0.117329    0.244067
+      2      16       116       100   199.918       220    0.144162    0.292305
+      3      16       174       158   210.589       232    0.170941    0.285481
+      4      16       251       235   234.918       308    0.241175    0.256543
+      5      16       316       300   239.914       260    0.206044    0.255882
+      6      15       392       377   251.206       308    0.137972    0.247426
+      7      15       458       443   252.984       264   0.0800146    0.245138
+      8      16       529       513   256.346       280    0.103529    0.239888
+      9      16       587       571   253.634       232    0.145535      0.2453
+     10      15       646       631   252.254       240    0.837727    0.246019
+  Total time run:       10.272
+  Total reads made:     646
+  Read size:            4194304
+  Object size:          4194304
+  Bandwidth (MB/sec):   251.558
+  Average IOPS:         62
+  Stddev IOPS:          10.005
+  Max IOPS:             77
+  Min IOPS:             45
+  Average Latency(s):   0.249385
+  Max latency(s):       0.888654
+  Min latency(s):       0.0103208
+  
diff --git a/src/common/options.cc b/src/common/options.cc
index 9d7c97af532e6..285c67fbb0d01 100644
--- a/src/common/options.cc
+++ b/src/common/options.cc
@@ -4388,7 +4388,7 @@ std::vector<Option> get_global_options() {
 
     Option("bluestore_allocator", Option::TYPE_STR, Option::LEVEL_ADVANCED)
     .set_default("hybrid")
-    .set_enum_allowed({"bitmap", "stupid", "avl", "hybrid"})
+    .set_enum_allowed({"bitmap", "stupid", "avl", "hybrid", "zoned"})
     .set_description("Allocator policy")
     .set_long_description("Allocator to use for bluestore.  Stupid should only be used for testing."),
 
diff --git a/src/include/config-h.in.cmake b/src/include/config-h.in.cmake
index e26671d3a005d..9f663396e4d0c 100644
--- a/src/include/config-h.in.cmake
+++ b/src/include/config-h.in.cmake
@@ -75,6 +75,9 @@
 /* Defined if you have libaio */
 #cmakedefine HAVE_LIBAIO
 
+/* Defined if you have libzbc */
+#cmakedefine HAVE_LIBZBC
+
 /* Defined if you have liburing */
 #cmakedefine HAVE_LIBURING
 
diff --git a/src/os/CMakeLists.txt b/src/os/CMakeLists.txt
index 740f5225d9122..1cc98a7aac9a8 100644
--- a/src/os/CMakeLists.txt
+++ b/src/os/CMakeLists.txt
@@ -45,6 +45,12 @@ if(HAVE_LIBAIO OR HAVE_POSIXAIO)
     bluestore/aio.cc)
 endif()
 
+if(HAVE_LIBZBC)
+  list(APPEND libos_srcs
+    bluestore/HMSMRDevice.cc
+    bluestore/ZonedAllocator.cc)
+endif()
+
 if(WITH_FUSE)
   list(APPEND libos_srcs
     FuseStore.cc)
@@ -92,6 +98,10 @@ if(HAVE_LIBAIO)
   target_link_libraries(os ${AIO_LIBRARIES})
 endif(HAVE_LIBAIO)
 
+if(HAVE_LIBZBC)
+  target_link_libraries(os ${ZBC_LIBRARIES})
+endif()
+
 if(WITH_FUSE)
   target_link_libraries(os FUSE::FUSE)
 endif()
diff --git a/src/os/bluestore/Allocator.cc b/src/os/bluestore/Allocator.cc
index 9cc04c59afbd9..5cd5d3b5ca838 100644
--- a/src/os/bluestore/Allocator.cc
+++ b/src/os/bluestore/Allocator.cc
@@ -6,6 +6,9 @@
 #include "BitmapAllocator.h"
 #include "AvlAllocator.h"
 #include "HybridAllocator.h"
+#ifdef HAVE_LIBZBC
+#include "ZonedAllocator.h"
+#endif
 #include "common/debug.h"
 #include "common/admin_socket.h"
 #define dout_subsys ceph_subsys_bluestore
@@ -125,6 +128,10 @@ Allocator *Allocator::create(CephContext* cct, string type,
     return new HybridAllocator(cct, size, block_size,
       cct->_conf.get_val<uint64_t>("bluestore_hybrid_alloc_mem_cap"),
       name);
+#ifdef HAVE_LIBZBC
+  } else if (type == "zoned") {
+    return new ZonedAllocator(cct, size, block_size, name);
+#endif
   }
   if (alloc == nullptr) {
     lderr(cct) << "Allocator::" << __func__ << " unknown alloc type "
diff --git a/src/os/bluestore/BlockDevice.cc b/src/os/bluestore/BlockDevice.cc
index fb87d2f897066..f431aaf190cff 100644
--- a/src/os/bluestore/BlockDevice.cc
+++ b/src/os/bluestore/BlockDevice.cc
@@ -32,6 +32,13 @@
 #include "libpmem.h"
 #endif
 
+#if defined(HAVE_LIBZBC)
+#include "HMSMRDevice.h"
+extern "C" {
+#include <libzbc/zbc.h>
+}
+#endif
+
 #include "common/debug.h"
 #include "common/EventTrace.h"
 #include "common/errno.h"
@@ -123,6 +130,11 @@ BlockDevice *BlockDevice::create(CephContext* cct, const string& path,
   }
 #endif
 #if defined(HAVE_LIBAIO) || defined(HAVE_POSIXAIO)
+#if defined(HAVE_LIBZBC)
+  if (zbc_device_is_zoned(path.c_str(), false, nullptr)) {
+    return new HMSMRDevice(cct, cb, cbpriv, d_cb, d_cbpriv);
+  }
+#endif
   if (type == "kernel") {
     return new KernelDevice(cct, cb, cbpriv, d_cb, d_cbpriv);
   }
diff --git a/src/os/bluestore/BlockDevice.h b/src/os/bluestore/BlockDevice.h
index eae7c97949103..9612c32f6cd85 100644
--- a/src/os/bluestore/BlockDevice.h
+++ b/src/os/bluestore/BlockDevice.h
@@ -144,6 +144,16 @@ protected:
   bool rotational = true;
   bool lock_exclusive = true;
 
+  // HM-SMR specific properties.  In HM-SMR drives the LBA space is divided into
+  // fixed-size zones.  Typically, the first few zones are randomly writable;
+  // they form a conventional region of the drive.  The remaining zones must be
+  // written sequentially and they must be reset before rewritten.  For example,
+  // a 14 TB HGST HSH721414AL drive has 52156 zones each of size is 256 MiB.
+  // The zones 0-523 are randomly writable and they form the conventional region
+  // of the drive.  The zones 524-52155 are sequential zones.
+  uint64_t conventional_region_size = 0;
+  uint64_t zone_size = 0;
+
 public:
   aio_callback_t aio_callback;
   void *aio_callback_priv;
@@ -159,6 +169,17 @@ public:
   virtual bool supported_bdev_label() { return true; }
   virtual bool is_rotational() { return rotational; }
 
+  // HM-SMR-specific calls
+  virtual bool is_smr() const { return false; }
+  virtual uint64_t get_zone_size() const {
+    ceph_assert(is_smr());
+    return zone_size;
+  }
+  virtual uint64_t get_conventional_region_size() const {
+    ceph_assert(is_smr());
+    return conventional_region_size;
+  }
+
   virtual void aio_submit(IOContext *ioc) = 0;
 
   void set_no_exclusive_lock() {
diff --git a/src/os/bluestore/BlueStore.cc b/src/os/bluestore/BlueStore.cc
index 8e10a44769277..7fb4e1ba98f81 100644
--- a/src/os/bluestore/BlueStore.cc
+++ b/src/os/bluestore/BlueStore.cc
@@ -5069,9 +5069,52 @@ int BlueStore::_open_alloc()
 	     << dendl;
   }
 
+  if (bdev->is_smr()) {
+    if (cct->_conf->bluestore_allocator != "zoned") {
+      dout(1) << __func__ << " The drive is HM-SMR but "
+	      << cct->_conf->bluestore_allocator << " allocator is specified. "
+	      << "Only zoned allocator can be used with HM-SMR drive." << dendl;
+      return -EINVAL;
+    }
+
+    // At least for now we want to use large min_alloc_size with HM-SMR drives.
+    // Populating used_blocks bitset on a debug build of ceph-osd takes about 5
+    // minutes with a 14 TB HM-SMR drive and 4 KiB min_alloc_size.
+    if (min_alloc_size < 64 * 1024) {
+      dout(1) << __func__ << " The drive is HM-SMR but min_alloc_size is "
+	      << min_alloc_size << ". "
+	      << "Please set to at least 64 KiB." << dendl;
+      return -EINVAL;
+    }
+
+    // We don't want to defer writes with HM-SMR because it violates sequential
+    // write requirement.
+    if (prefer_deferred_size) {
+      dout(1) << __func__ << " The drive is HM-SMR but prefer_deferred_size is "
+	      << prefer_deferred_size << ". "
+	      << "Please set to 0." << dendl;
+      return -EINVAL;
+    }
+
+    // For now, to avoid interface changes we piggyback zone_size (in MiB) and
+    // the first sequential zone number onto min_alloc_size and pass it to
+    // Allocator::create.
+    uint64_t zone_size = bdev->get_zone_size();
+    uint64_t zone_size_mb = zone_size / (1024 * 1024);
+    uint64_t first_seq_zone = bdev->get_conventional_region_size() / zone_size;
+
+    min_alloc_size |= (zone_size_mb << 32);
+    min_alloc_size |= (first_seq_zone << 48);
+  }
+
   alloc = Allocator::create(cct, cct->_conf->bluestore_allocator,
                             bdev->get_size(),
                             min_alloc_size, "block");
+
+  if (bdev->is_smr()) {
+    min_alloc_size &= 0x00000000ffffffff;
+  }
+
   if (!alloc) {
     lderr(cct) << __func__ << " Allocator::unknown alloc type "
                << cct->_conf->bluestore_allocator
@@ -5336,25 +5379,32 @@ int BlueStore::_minimal_open_bluefs(bool create)
   }
   if (create) {
     // note: we always leave the first SUPER_RESERVED (8k) of the device unused
-    uint64_t initial =
-      bdev->get_size() * (cct->_conf->bluestore_bluefs_min_ratio +
-			  cct->_conf->bluestore_bluefs_gift_ratio);
-    initial = std::max(initial, cct->_conf->bluestore_bluefs_min);
-    uint64_t alloc_size = cct->_conf->bluefs_shared_alloc_size;
-    if (alloc_size % min_alloc_size) {
-      derr << __func__ << " bluefs_shared_alloc_size 0x" << std::hex
-	    << alloc_size << " is not a multiple of "
-	    << "min_alloc_size 0x" << min_alloc_size << std::dec << dendl;
-      r = -EINVAL;
-      goto free_bluefs;
+    uint64_t start, initial;
+    if (bdev->is_smr()) {
+      // For now, we allocate all of the conventional region on HM-SMR drive to
+      // bluefs and assume that it never outgrows that region.
+      start = _get_ondisk_reserved();
+      initial = bdev->get_conventional_region_size() - start;
+    } else {
+      initial = bdev->get_size() * (cct->_conf->bluestore_bluefs_min_ratio +
+				    cct->_conf->bluestore_bluefs_gift_ratio);
+      initial = std::max(initial, cct->_conf->bluestore_bluefs_min);
+      uint64_t alloc_size = cct->_conf->bluefs_shared_alloc_size;
+      if (alloc_size % min_alloc_size) {
+	derr << __func__ << " bluefs_shared_alloc_size 0x" << std::hex
+	     << alloc_size << " is not a multiple of "
+	     << "min_alloc_size 0x" << min_alloc_size << std::dec << dendl;
+	r = -EINVAL;
+	goto free_bluefs;
+      }
+      // align to bluefs's alloc_size
+      initial = p2roundup(initial, alloc_size);
+      // put bluefs in the middle of the device in case it is an HDD
+      start = p2align((bdev->get_size() - initial) / 2, alloc_size);
+      //avoiding superblock overwrite
+      start = std::max(alloc_size, start);
+      ceph_assert(start >=_get_ondisk_reserved());
     }
-    // align to bluefs's alloc_size
-    initial = p2roundup(initial, alloc_size);
-    // put bluefs in the middle of the device in case it is an HDD
-    uint64_t start = p2align((bdev->get_size() - initial) / 2, alloc_size);
-    //avoiding superblock overwrite
-    start = std::max(alloc_size, start);
-    ceph_assert(start >=_get_ondisk_reserved());
 
     bluefs->add_block_extent(bluefs_layout.shared_bdev, start, initial);
     bluefs_extents.insert(start, initial);
@@ -12350,6 +12400,15 @@ int BlueStore::queue_transactions(
   TransContext *txc = _txc_create(static_cast<Collection*>(ch.get()), osr,
 				  &on_commit);
 
+  // With HM-SMR drives (and ZNS SSDs) we want the I/O allocation and I/O
+  // submission to happen atomically because if I/O submission happens in a
+  // different order than I/O allocation, we end up issuing non-sequential
+  // writes to the drive.  This is a temporary solution until ZONE APPEND
+  // support matures in the kernel.  For more information please see:
+  // https://www.usenix.org/conference/vault20/presentation/bjorling
+  if (bdev->is_smr()) {
+    atomic_alloc_and_submit_lock.lock();
+  }
   for (vector<Transaction>::iterator p = tls.begin(); p != tls.end(); ++p) {
     txc->bytes += (*p).get_num_bytes();
     _txc_add_transaction(txc, &(*p));
@@ -12404,6 +12463,10 @@ int BlueStore::queue_transactions(
   // execute (start)
   _txc_state_proc(txc);
 
+  if (bdev->is_smr()) {
+    atomic_alloc_and_submit_lock.unlock();
+  }
+
   // we're immediately readable (unlike FileStore)
   for (auto c : on_applied_sync) {
     c->complete(0);
@@ -12893,6 +12956,23 @@ void BlueStore::_do_write_small(
   dout(10) << __func__ << " 0x" << std::hex << offset << "~" << length
 	   << std::dec << dendl;
   ceph_assert(length < min_alloc_size);
+
+  // On zoned devices, the first goal is to support non-overwrite workloads,
+  // such as RGW, with large, aligned objects.  Therefore, for user writes
+  // _do_write_small should not trigger.  OSDs, however, write and update a tiny
+  // amount of metadata, such as OSD maps, to disk.  For those cases, we
+  // temporarily just pad them to min_alloc_size and write them to a new place
+  // on every update.
+  if (bdev->is_smr()) {
+    BlobRef b = c->new_blob();
+    uint64_t b_off = 0, b_off0 = 0;
+    bufferlist l;
+    blp.copy(length, l);
+    _pad_zeros(&l, &b_off0, min_alloc_size);
+    wctx->write(offset, b, min_alloc_size, b_off0, l, b_off, length, false, true);
+    return;
+  }
+
   uint64_t end_offs = offset + length;
 
   logger->inc(l_bluestore_write_small);
diff --git a/src/os/bluestore/BlueStore.h b/src/os/bluestore/BlueStore.h
index 4ae14fa5c60d3..8685fb8782a58 100644
--- a/src/os/bluestore/BlueStore.h
+++ b/src/os/bluestore/BlueStore.h
@@ -2001,6 +2001,8 @@ private:
   interval_set<uint64_t> bluefs_extents_reclaiming; ///< currently reclaiming
 
   ceph::mutex deferred_lock = ceph::make_mutex("BlueStore::deferred_lock");
+  ceph::mutex atomic_alloc_and_submit_lock =
+      ceph::make_mutex("BlueStore::atomic_alloc_and_submit_lock");
   std::atomic<uint64_t> deferred_seq = {0};
   deferred_osr_queue_t deferred_queue; ///< osr's with deferred io pending
   int deferred_queue_size = 0;         ///< num txc's queued across all osrs
diff --git a/src/os/bluestore/HMSMRDevice.cc b/src/os/bluestore/HMSMRDevice.cc
new file mode 100644
index 0000000000000..4fd6e8a37f02f
--- /dev/null
+++ b/src/os/bluestore/HMSMRDevice.cc
@@ -0,0 +1,1211 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2014 Red Hat
+ * Copyright (C) 2020 Abutalib Aghayev
+ *
+ * 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.
+ *
+ */
+
+#include <unistd.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <sys/file.h>
+
+#include "HMSMRDevice.h"
+#include "include/intarith.h"
+#include "include/types.h"
+#include "include/compat.h"
+#include "include/scope_guard.h"
+#include "include/stringify.h"
+#include "common/blkdev.h"
+#include "common/errno.h"
+#if defined(__FreeBSD__)
+#include "bsm/audit_errno.h"
+#endif
+#include "common/debug.h"
+#include "common/numa.h"
+
+#include "global/global_context.h"
+#include "ceph_io_uring.h"
+
+extern "C" {
+#include <libzbc/zbc.h>
+}
+
+#define dout_context cct
+#define dout_subsys ceph_subsys_bdev
+#undef dout_prefix
+#define dout_prefix *_dout << "smrbdev(" << this << " " << path << ") "
+
+HMSMRDevice::HMSMRDevice(CephContext* cct, aio_callback_t cb, void *cbpriv, aio_callback_t d_cb, void *d_cbpriv)
+  : BlockDevice(cct, cb, cbpriv),
+    aio(false), dio(false),
+    discard_callback(d_cb),
+    discard_callback_priv(d_cbpriv),
+    aio_stop(false),
+    discard_started(false),
+    discard_stop(false),
+    aio_thread(this),
+    discard_thread(this),
+    injecting_crash(0)
+{
+  fd_directs.resize(WRITE_LIFE_MAX, -1);
+  fd_buffereds.resize(WRITE_LIFE_MAX, -1);
+
+  bool use_ioring = g_ceph_context->_conf.get_val<bool>("bluestore_ioring");
+  unsigned int iodepth = cct->_conf->bdev_aio_max_queue_depth;
+
+  if (use_ioring && ioring_queue_t::supported()) {
+    io_queue = std::make_unique<ioring_queue_t>(iodepth);
+  } else {
+    static bool once;
+    if (use_ioring && !once) {
+      derr << "WARNING: io_uring API is not supported! Fallback to libaio!"
+           << dendl;
+      once = true;
+    }
+    io_queue = std::make_unique<aio_queue_t>(iodepth);
+  }
+}
+
+int HMSMRDevice::_lock()
+{
+  dout(10) << __func__ << " " << fd_directs[WRITE_LIFE_NOT_SET] << dendl;
+  int r = ::flock(fd_directs[WRITE_LIFE_NOT_SET], LOCK_EX | LOCK_NB);
+  if (r < 0) {
+    derr << __func__ << " flock failed on " << path << dendl;
+    return -errno;
+  }
+  return 0;
+}
+
+bool HMSMRDevice::set_smr_params(const std::string& path) {
+  dout(10) << __func__ << " opening " << path << dendl;
+
+  zbc_device *dev;
+  if (zbc_open(path.c_str(), O_RDWR | O_DIRECT, &dev) != 0) {
+    return false;
+  }
+  auto close_dev = make_scope_guard([dev] { zbc_close(dev); });
+
+  unsigned int nr_zones = 0;
+  if (zbc_report_nr_zones(dev, 0, ZBC_RO_NOT_WP, &nr_zones) != 0) {
+    return false;
+  }
+
+  std::vector<zbc_zone> zones(nr_zones);
+  if (zbc_report_zones(dev, 0, ZBC_RO_NOT_WP, zones.data(), &nr_zones) != 0) {
+    return false;
+  }
+
+  zone_size = 512 * zbc_zone_length(&zones[0]); // on HM-SMR zones are equisized
+  conventional_region_size = nr_zones * zone_size;
+
+  dout(10) << __func__ << " setting zone size to " << zone_size
+	   << " and conventional region size to " << conventional_region_size
+           << dendl;
+
+  return true;
+}
+
+int HMSMRDevice::open(const string& p)
+{
+  path = p;
+  int r = 0, i = 0;
+  dout(1) << __func__ << " path " << path << dendl;
+
+  for (i = 0; i < WRITE_LIFE_MAX; i++) {
+    int fd = ::open(path.c_str(), O_RDWR | O_DIRECT);
+    if (fd  < 0) {
+      r = -errno;
+      break;
+    }
+    fd_directs[i] = fd;
+
+    fd  = ::open(path.c_str(), O_RDWR | O_CLOEXEC);
+    if (fd  < 0) {
+      r = -errno;
+      break;
+    }
+    fd_buffereds[i] = fd;
+  }
+
+  if (i != WRITE_LIFE_MAX) {
+    derr << __func__ << " open got: " << cpp_strerror(r) << dendl;
+    goto out_fail;
+  }
+
+  if (!set_smr_params(p)) {
+    derr << __func__ << " failed to set HM-SMR parameters" << dendl;
+    goto out_fail;
+  }
+
+#if defined(F_SET_FILE_RW_HINT)
+  for (i = WRITE_LIFE_NONE; i < WRITE_LIFE_MAX; i++) {
+    if (fcntl(fd_directs[i], F_SET_FILE_RW_HINT, &i) < 0) {
+      r = -errno;
+      break;
+    }
+    if (fcntl(fd_buffereds[i], F_SET_FILE_RW_HINT, &i) < 0) {
+      r = -errno;
+      break;
+    }
+  }
+  if (i != WRITE_LIFE_MAX) {
+    enable_wrt = false;
+    dout(0) << "ioctl(F_SET_FILE_RW_HINT) on " << path << " failed: " << cpp_strerror(r) << dendl;
+  }
+#endif
+
+  dio = true;
+  aio = cct->_conf->bdev_aio;
+  if (!aio) {
+    ceph_abort_msg("non-aio not supported");
+  }
+
+  // disable readahead as it will wreak havoc on our mix of
+  // directio/aio and buffered io.
+  r = posix_fadvise(fd_buffereds[WRITE_LIFE_NOT_SET], 0, 0, POSIX_FADV_RANDOM);
+  if (r) {
+    r = -r;
+    derr << __func__ << " posix_fadvise got: " << cpp_strerror(r) << dendl;
+    goto out_fail;
+  }
+
+  if (lock_exclusive) {
+    r = _lock();
+    if (r < 0) {
+      derr << __func__ << " failed to lock " << path << ": " << cpp_strerror(r)
+	   << dendl;
+      goto out_fail;
+    }
+  }
+
+  struct stat st;
+  r = ::fstat(fd_directs[WRITE_LIFE_NOT_SET], &st);
+  if (r < 0) {
+    r = -errno;
+    derr << __func__ << " fstat got " << cpp_strerror(r) << dendl;
+    goto out_fail;
+  }
+
+  // Operate as though the block size is 4 KB.  The backing file
+  // blksize doesn't strictly matter except that some file systems may
+  // require a read/modify/write if we write something smaller than
+  // it.
+  block_size = cct->_conf->bdev_block_size;
+  if (block_size != (unsigned)st.st_blksize) {
+    dout(1) << __func__ << " backing device/file reports st_blksize "
+	    << st.st_blksize << ", using bdev_block_size "
+	    << block_size << " anyway" << dendl;
+  }
+
+
+  {
+    BlkDev blkdev_direct(fd_directs[WRITE_LIFE_NOT_SET]);
+    BlkDev blkdev_buffered(fd_buffereds[WRITE_LIFE_NOT_SET]);
+
+    if (S_ISBLK(st.st_mode)) {
+      int64_t s;
+      r = blkdev_direct.get_size(&s);
+      if (r < 0) {
+	goto out_fail;
+      }
+      size = s;
+    } else {
+      size = st.st_size;
+    }
+
+    char partition[PATH_MAX], devname[PATH_MAX];
+    if ((r = blkdev_buffered.partition(partition, PATH_MAX)) ||
+	(r = blkdev_buffered.wholedisk(devname, PATH_MAX))) {
+      derr << "unable to get device name for " << path << ": "
+	<< cpp_strerror(r) << dendl;
+      rotational = true;
+    } else {
+      dout(20) << __func__ << " devname " << devname << dendl;
+      rotational = blkdev_buffered.is_rotational();
+      support_discard = blkdev_buffered.support_discard();
+      this->devname = devname;
+      _detect_vdo();
+    }
+  }
+
+  r = _aio_start();
+  if (r < 0) {
+    goto out_fail;
+  }
+  _discard_start();
+
+  // round size down to an even block
+  size &= ~(block_size - 1);
+
+  dout(1) << __func__
+	  << " size " << size
+	  << " (0x" << std::hex << size << std::dec << ", "
+	  << byte_u_t(size) << ")"
+	  << " block_size " << block_size
+	  << " (" << byte_u_t(block_size) << ")"
+	  << " " << (rotational ? "rotational" : "non-rotational")
+      << " discard " << (support_discard ? "supported" : "not supported")
+	  << dendl;
+  return 0;
+
+out_fail:
+  for (i = 0; i < WRITE_LIFE_MAX; i++) {
+    if (fd_directs[i] >= 0) {
+      VOID_TEMP_FAILURE_RETRY(::close(fd_directs[i]));
+      fd_directs[i] = -1;
+    } else {
+      break;
+    }
+    if (fd_buffereds[i] >= 0) {
+      VOID_TEMP_FAILURE_RETRY(::close(fd_buffereds[i]));
+      fd_buffereds[i] = -1;
+    } else {
+      break;
+    }
+  }
+  return r;
+}
+
+int HMSMRDevice::get_devices(std::set<std::string> *ls) const
+{
+  if (devname.empty()) {
+    return 0;
+  }
+  get_raw_devices(devname, ls);
+  return 0;
+}
+
+void HMSMRDevice::close()
+{
+  dout(1) << __func__ << dendl;
+  _aio_stop();
+  _discard_stop();
+
+  if (vdo_fd >= 0) {
+    VOID_TEMP_FAILURE_RETRY(::close(vdo_fd));
+    vdo_fd = -1;
+  }
+
+  for (int i = 0; i < WRITE_LIFE_MAX; i++) {
+    assert(fd_directs[i] >= 0);
+    VOID_TEMP_FAILURE_RETRY(::close(fd_directs[i]));
+    fd_directs[i] = -1;
+
+    assert(fd_buffereds[i] >= 0);
+    VOID_TEMP_FAILURE_RETRY(::close(fd_buffereds[i]));
+    fd_buffereds[i] = -1;
+  }
+  path.clear();
+}
+
+int HMSMRDevice::collect_metadata(const string& prefix, map<string,string> *pm) const
+{
+  (*pm)[prefix + "support_discard"] = stringify((int)(bool)support_discard);
+  (*pm)[prefix + "rotational"] = stringify((int)(bool)rotational);
+  (*pm)[prefix + "size"] = stringify(get_size());
+  (*pm)[prefix + "block_size"] = stringify(get_block_size());
+  (*pm)[prefix + "driver"] = "HMSMRDevice";
+  if (rotational) {
+    (*pm)[prefix + "type"] = "hdd";
+  } else {
+    (*pm)[prefix + "type"] = "ssd";
+  }
+  if (vdo_fd >= 0) {
+    (*pm)[prefix + "vdo"] = "true";
+    uint64_t total, avail;
+    get_vdo_utilization(vdo_fd, &total, &avail);
+    (*pm)[prefix + "vdo_physical_size"] = stringify(total);
+  }
+
+  {
+    string res_names;
+    std::set<std::string> devnames;
+    if (get_devices(&devnames) == 0) {
+      for (auto& dev : devnames) {
+	if (!res_names.empty()) {
+	  res_names += ",";
+	}
+	res_names += dev;
+      }
+      if (res_names.size()) {
+	(*pm)[prefix + "devices"] = res_names;
+      }
+    }
+  }
+
+  struct stat st;
+  int r = ::fstat(fd_buffereds[WRITE_LIFE_NOT_SET], &st);
+  if (r < 0)
+    return -errno;
+  if (S_ISBLK(st.st_mode)) {
+    (*pm)[prefix + "access_mode"] = "blk";
+
+    char buffer[1024] = {0};
+    BlkDev blkdev{fd_buffereds[WRITE_LIFE_NOT_SET]};
+    if (r = blkdev.partition(buffer, sizeof(buffer)); r) {
+      (*pm)[prefix + "partition_path"] = "unknown";
+    } else {
+      (*pm)[prefix + "partition_path"] = buffer;
+    }
+    buffer[0] = '\0';
+    if (r = blkdev.partition(buffer, sizeof(buffer)); r) {
+      (*pm)[prefix + "dev_node"] = "unknown";
+    } else {
+      (*pm)[prefix + "dev_node"] = buffer;
+    }
+    if (!r) {
+      return 0;
+    }
+    buffer[0] = '\0';
+    blkdev.model(buffer, sizeof(buffer));
+    (*pm)[prefix + "model"] = buffer;
+
+    buffer[0] = '\0';
+    blkdev.dev(buffer, sizeof(buffer));
+    (*pm)[prefix + "dev"] = buffer;
+
+    // nvme exposes a serial number
+    buffer[0] = '\0';
+    blkdev.serial(buffer, sizeof(buffer));
+    (*pm)[prefix + "serial"] = buffer;
+
+    // numa
+    int node;
+    r = blkdev.get_numa_node(&node);
+    if (r >= 0) {
+      (*pm)[prefix + "numa_node"] = stringify(node);
+    }
+  } else {
+    (*pm)[prefix + "access_mode"] = "file";
+    (*pm)[prefix + "path"] = path;
+  }
+  return 0;
+}
+
+void HMSMRDevice::_detect_vdo()
+{
+  vdo_fd = get_vdo_stats_handle(devname.c_str(), &vdo_name);
+  if (vdo_fd >= 0) {
+    dout(1) << __func__ << " VDO volume " << vdo_name
+	    << " maps to " << devname << dendl;
+  } else {
+    dout(20) << __func__ << " no VDO volume maps to " << devname << dendl;
+  }
+  return;
+}
+
+bool HMSMRDevice::get_thin_utilization(uint64_t *total, uint64_t *avail) const
+{
+  if (vdo_fd < 0) {
+    return false;
+  }
+  return get_vdo_utilization(vdo_fd, total, avail);
+}
+
+int HMSMRDevice::choose_fd(bool buffered, int write_hint) const
+{
+  assert(write_hint >= WRITE_LIFE_NOT_SET && write_hint < WRITE_LIFE_MAX);
+  if (!enable_wrt)
+    write_hint = WRITE_LIFE_NOT_SET;
+  return buffered ? fd_buffereds[write_hint] : fd_directs[write_hint];
+}
+
+int HMSMRDevice::flush()
+{
+  // protect flush with a mutex.  note that we are not really protecting
+  // data here.  instead, we're ensuring that if any flush() caller
+  // sees that io_since_flush is true, they block any racing callers
+  // until the flush is observed.  that allows racing threads to be
+  // calling flush while still ensuring that *any* of them that got an
+  // aio completion notification will not return before that aio is
+  // stable on disk: whichever thread sees the flag first will block
+  // followers until the aio is stable.
+  std::lock_guard l(flush_mutex);
+
+  bool expect = true;
+  if (!io_since_flush.compare_exchange_strong(expect, false)) {
+    dout(10) << __func__ << " no-op (no ios since last flush), flag is "
+	     << (int)io_since_flush.load() << dendl;
+    return 0;
+  }
+
+  dout(10) << __func__ << " start" << dendl;
+  if (cct->_conf->bdev_inject_crash) {
+    ++injecting_crash;
+    // sleep for a moment to give other threads a chance to submit or
+    // wait on io that races with a flush.
+    derr << __func__ << " injecting crash. first we sleep..." << dendl;
+    sleep(cct->_conf->bdev_inject_crash_flush_delay);
+    derr << __func__ << " and now we die" << dendl;
+    cct->_log->flush();
+    _exit(1);
+  }
+  utime_t start = ceph_clock_now();
+  int r = ::fdatasync(fd_directs[WRITE_LIFE_NOT_SET]);
+  utime_t end = ceph_clock_now();
+  utime_t dur = end - start;
+  if (r < 0) {
+    r = -errno;
+    derr << __func__ << " fdatasync got: " << cpp_strerror(r) << dendl;
+    ceph_abort();
+  }
+  dout(5) << __func__ << " in " << dur << dendl;;
+  return r;
+}
+
+int HMSMRDevice::_aio_start()
+{
+  if (aio) {
+    dout(10) << __func__ << dendl;
+    int r = io_queue->init(fd_directs);
+    if (r < 0) {
+      if (r == -EAGAIN) {
+	derr << __func__ << " io_setup(2) failed with EAGAIN; "
+	     << "try increasing /proc/sys/fs/aio-max-nr" << dendl;
+      } else {
+	derr << __func__ << " io_setup(2) failed: " << cpp_strerror(r) << dendl;
+      }
+      return r;
+    }
+    aio_thread.create("bstore_aio");
+  }
+  return 0;
+}
+
+void HMSMRDevice::_aio_stop()
+{
+  if (aio) {
+    dout(10) << __func__ << dendl;
+    aio_stop = true;
+    aio_thread.join();
+    aio_stop = false;
+    io_queue->shutdown();
+  }
+}
+
+int HMSMRDevice::_discard_start()
+{
+    discard_thread.create("bstore_discard");
+    return 0;
+}
+
+void HMSMRDevice::_discard_stop()
+{
+  dout(10) << __func__ << dendl;
+  {
+    std::unique_lock l(discard_lock);
+    while (!discard_started) {
+      discard_cond.wait(l);
+    }
+    discard_stop = true;
+    discard_cond.notify_all();
+  }
+  discard_thread.join();
+  {
+    std::lock_guard l(discard_lock);
+    discard_stop = false;
+  }
+  dout(10) << __func__ << " stopped" << dendl;
+}
+
+void HMSMRDevice::discard_drain()
+{
+  dout(10) << __func__ << dendl;
+  std::unique_lock l(discard_lock);
+  while (!discard_queued.empty() || discard_running) {
+    discard_cond.wait(l);
+  }
+}
+
+static bool is_expected_ioerr(const int r)
+{
+  // https://lxr.missinglinkelectronics.com/linux+v4.15/block/blk-core.c#L135
+  return (r == -EOPNOTSUPP || r == -ETIMEDOUT || r == -ENOSPC ||
+	  r == -ENOLINK || r == -EREMOTEIO  || r == -EAGAIN || r == -EIO ||
+	  r == -ENODATA || r == -EILSEQ || r == -ENOMEM ||
+#if defined(__linux__)
+	  r == -EREMCHG || r == -EBADE
+#elif defined(__FreeBSD__)
+	  r == - BSM_ERRNO_EREMCHG || r == -BSM_ERRNO_EBADE
+#endif
+	  );
+}
+
+void HMSMRDevice::_aio_thread()
+{
+  dout(10) << __func__ << " start" << dendl;
+  int inject_crash_count = 0;
+  while (!aio_stop) {
+    dout(40) << __func__ << " polling" << dendl;
+    int max = cct->_conf->bdev_aio_reap_max;
+    aio_t *aio[max];
+    int r = io_queue->get_next_completed(cct->_conf->bdev_aio_poll_ms,
+					 aio, max);
+    if (r < 0) {
+      derr << __func__ << " got " << cpp_strerror(r) << dendl;
+      ceph_abort_msg("got unexpected error from io_getevents");
+    }
+    if (r > 0) {
+      dout(30) << __func__ << " got " << r << " completed aios" << dendl;
+      for (int i = 0; i < r; ++i) {
+	IOContext *ioc = static_cast<IOContext*>(aio[i]->priv);
+	_aio_log_finish(ioc, aio[i]->offset, aio[i]->length);
+	if (aio[i]->queue_item.is_linked()) {
+	  std::lock_guard l(debug_queue_lock);
+	  debug_aio_unlink(*aio[i]);
+	}
+
+	// set flag indicating new ios have completed.  we do this *before*
+	// any completion or notifications so that any user flush() that
+	// follows the observed io completion will include this io.  Note
+	// that an earlier, racing flush() could observe and clear this
+	// flag, but that also ensures that the IO will be stable before the
+	// later flush() occurs.
+	io_since_flush.store(true);
+
+	long r = aio[i]->get_return_value();
+        if (r < 0) {
+          derr << __func__ << " got r=" << r << " (" << cpp_strerror(r) << ")"
+	       << dendl;
+          if (ioc->allow_eio && is_expected_ioerr(r)) {
+            derr << __func__ << " translating the error to EIO for upper layer"
+		 << dendl;
+            ioc->set_return_value(-EIO);
+          } else {
+	    if (is_expected_ioerr(r)) {
+	      note_io_error_event(
+		devname.c_str(),
+		path.c_str(),
+		r,
+#if defined(HAVE_POSIXAIO)
+                aio[i]->aio.aiocb.aio_lio_opcode,
+#else
+                aio[i]->iocb.aio_lio_opcode,
+#endif
+		aio[i]->offset,
+		aio[i]->length);
+	      ceph_abort_msg(
+		"Unexpected IO error. "
+		"This may suggest a hardware issue. "
+		"Please check your kernel log!");
+	    }
+	    ceph_abort_msg(
+	      "Unexpected IO error. "
+	      "This may suggest HW issue. Please check your dmesg!");
+          }
+        } else if (aio[i]->length != (uint64_t)r) {
+          derr << "aio to 0x" << std::hex << aio[i]->offset
+	       << "~" << aio[i]->length << std::dec
+               << " but returned: " << r << dendl;
+          ceph_abort_msg("unexpected aio return value: does not match length");
+        }
+
+        dout(10) << __func__ << " finished aio " << aio[i] << " r " << r
+                 << " ioc " << ioc
+                 << " with " << (ioc->num_running.load() - 1)
+                 << " aios left" << dendl;
+
+	// NOTE: once num_running and we either call the callback or
+	// call aio_wake we cannot touch ioc or aio[] as the caller
+	// may free it.
+	if (ioc->priv) {
+	  if (--ioc->num_running == 0) {
+	    aio_callback(aio_callback_priv, ioc->priv);
+	  }
+	} else {
+          ioc->try_aio_wake();
+	}
+      }
+    }
+    if (cct->_conf->bdev_debug_aio) {
+      utime_t now = ceph_clock_now();
+      std::lock_guard l(debug_queue_lock);
+      if (debug_oldest) {
+	if (debug_stall_since == utime_t()) {
+	  debug_stall_since = now;
+	} else {
+	  if (cct->_conf->bdev_debug_aio_suicide_timeout) {
+            utime_t cutoff = now;
+	    cutoff -= cct->_conf->bdev_debug_aio_suicide_timeout;
+	    if (debug_stall_since < cutoff) {
+	      derr << __func__ << " stalled aio " << debug_oldest
+		   << " since " << debug_stall_since << ", timeout is "
+		   << cct->_conf->bdev_debug_aio_suicide_timeout
+		   << "s, suicide" << dendl;
+	      ceph_abort_msg("stalled aio... buggy kernel or bad device?");
+	    }
+	  }
+	}
+      }
+    }
+    reap_ioc();
+    if (cct->_conf->bdev_inject_crash) {
+      ++inject_crash_count;
+      if (inject_crash_count * cct->_conf->bdev_aio_poll_ms / 1000 >
+	  cct->_conf->bdev_inject_crash + cct->_conf->bdev_inject_crash_flush_delay) {
+	derr << __func__ << " bdev_inject_crash trigger from aio thread"
+	     << dendl;
+	cct->_log->flush();
+	_exit(1);
+      }
+    }
+  }
+  reap_ioc();
+  dout(10) << __func__ << " end" << dendl;
+}
+
+void HMSMRDevice::_discard_thread()
+{
+  std::unique_lock l(discard_lock);
+  ceph_assert(!discard_started);
+  discard_started = true;
+  discard_cond.notify_all();
+  while (true) {
+    ceph_assert(discard_finishing.empty());
+    if (discard_queued.empty()) {
+      if (discard_stop)
+	break;
+      dout(20) << __func__ << " sleep" << dendl;
+      discard_cond.notify_all(); // for the thread trying to drain...
+      discard_cond.wait(l);
+      dout(20) << __func__ << " wake" << dendl;
+    } else {
+      discard_finishing.swap(discard_queued);
+      discard_running = true;
+      l.unlock();
+      dout(20) << __func__ << " finishing" << dendl;
+      for (auto p = discard_finishing.begin();p != discard_finishing.end(); ++p) {
+	discard(p.get_start(), p.get_len());
+      }
+
+      discard_callback(discard_callback_priv, static_cast<void*>(&discard_finishing));
+      discard_finishing.clear();
+      l.lock();
+      discard_running = false;
+    }
+  }
+  dout(10) << __func__ << " finish" << dendl;
+  discard_started = false;
+}
+
+int HMSMRDevice::queue_discard(interval_set<uint64_t> &to_release)
+{
+  if (!support_discard)
+    return -1;
+
+  if (to_release.empty())
+    return 0;
+
+  std::lock_guard l(discard_lock);
+  discard_queued.insert(to_release);
+  discard_cond.notify_all();
+  return 0;
+}
+
+void HMSMRDevice::_aio_log_start(
+  IOContext *ioc,
+  uint64_t offset,
+  uint64_t length)
+{
+  dout(20) << __func__ << " 0x" << std::hex << offset << "~" << length
+	   << std::dec << dendl;
+  if (cct->_conf->bdev_debug_inflight_ios) {
+    std::lock_guard l(debug_lock);
+    if (debug_inflight.intersects(offset, length)) {
+      derr << __func__ << " inflight overlap of 0x"
+	   << std::hex
+	   << offset << "~" << length << std::dec
+	   << " with " << debug_inflight << dendl;
+      ceph_abort();
+    }
+    debug_inflight.insert(offset, length);
+  }
+}
+
+void HMSMRDevice::debug_aio_link(aio_t& aio)
+{
+  if (debug_queue.empty()) {
+    debug_oldest = &aio;
+  }
+  debug_queue.push_back(aio);
+}
+
+void HMSMRDevice::debug_aio_unlink(aio_t& aio)
+{
+  if (aio.queue_item.is_linked()) {
+    debug_queue.erase(debug_queue.iterator_to(aio));
+    if (debug_oldest == &aio) {
+      auto age = cct->_conf->bdev_debug_aio_log_age;
+      if (age && debug_stall_since != utime_t()) {
+        utime_t cutoff = ceph_clock_now();
+	cutoff -= age;
+	if (debug_stall_since < cutoff) {
+	  derr << __func__ << " stalled aio " << debug_oldest
+		<< " since " << debug_stall_since << ", timeout is "
+		<< age
+		<< "s" << dendl;
+	}
+      }
+
+      if (debug_queue.empty()) {
+	debug_oldest = nullptr;
+      } else {
+	debug_oldest = &debug_queue.front();
+      }
+      debug_stall_since = utime_t();
+    }
+  }
+}
+
+void HMSMRDevice::_aio_log_finish(
+  IOContext *ioc,
+  uint64_t offset,
+  uint64_t length)
+{
+  dout(20) << __func__ << " " << aio << " 0x"
+	   << std::hex << offset << "~" << length << std::dec << dendl;
+  if (cct->_conf->bdev_debug_inflight_ios) {
+    std::lock_guard l(debug_lock);
+    debug_inflight.erase(offset, length);
+  }
+}
+
+void HMSMRDevice::aio_submit(IOContext *ioc)
+{
+  dout(20) << __func__ << " ioc " << ioc
+	   << " pending " << ioc->num_pending.load()
+	   << " running " << ioc->num_running.load()
+	   << dendl;
+
+  if (ioc->num_pending.load() == 0) {
+    return;
+  }
+
+  // move these aside, and get our end iterator position now, as the
+  // aios might complete as soon as they are submitted and queue more
+  // wal aio's.
+  list<aio_t>::iterator e = ioc->running_aios.begin();
+  ioc->running_aios.splice(e, ioc->pending_aios);
+
+  int pending = ioc->num_pending.load();
+  ioc->num_running += pending;
+  ioc->num_pending -= pending;
+  ceph_assert(ioc->num_pending.load() == 0);  // we should be only thread doing this
+  ceph_assert(ioc->pending_aios.size() == 0);
+
+  if (cct->_conf->bdev_debug_aio) {
+    list<aio_t>::iterator p = ioc->running_aios.begin();
+    while (p != e) {
+      dout(30) << __func__ << " " << *p << dendl;
+      std::lock_guard l(debug_queue_lock);
+      debug_aio_link(*p++);
+    }
+  }
+
+  void *priv = static_cast<void*>(ioc);
+  int r, retries = 0;
+  r = io_queue->submit_batch(ioc->running_aios.begin(), e,
+			     pending, priv, &retries);
+
+  if (retries)
+    derr << __func__ << " retries " << retries << dendl;
+  if (r < 0) {
+    derr << " aio submit got " << cpp_strerror(r) << dendl;
+    ceph_assert(r == 0);
+  }
+}
+
+int HMSMRDevice::_sync_write(uint64_t off, bufferlist &bl, bool buffered, int write_hint)
+{
+  uint64_t len = bl.length();
+  dout(5) << __func__ << " 0x" << std::hex << off << "~" << len
+	  << std::dec << (buffered ? " (buffered)" : " (direct)") << dendl;
+  if (cct->_conf->bdev_inject_crash &&
+      rand() % cct->_conf->bdev_inject_crash == 0) {
+    derr << __func__ << " bdev_inject_crash: dropping io 0x" << std::hex
+	 << off << "~" << len << std::dec << dendl;
+    ++injecting_crash;
+    return 0;
+  }
+  vector<iovec> iov;
+  bl.prepare_iov(&iov);
+  int r = ::pwritev(choose_fd(buffered, write_hint),
+		    &iov[0], iov.size(), off);
+
+  if (r < 0) {
+    r = -errno;
+    derr << __func__ << " pwritev error: " << cpp_strerror(r) << dendl;
+    return r;
+  }
+#ifdef HAVE_SYNC_FILE_RANGE
+  if (buffered) {
+    // initiate IO and wait till it completes
+    r = ::sync_file_range(fd_buffereds[WRITE_LIFE_NOT_SET], off, len, SYNC_FILE_RANGE_WRITE|SYNC_FILE_RANGE_WAIT_AFTER|SYNC_FILE_RANGE_WAIT_BEFORE);
+    if (r < 0) {
+      r = -errno;
+      derr << __func__ << " sync_file_range error: " << cpp_strerror(r) << dendl;
+      return r;
+    }
+  }
+#endif
+
+  io_since_flush.store(true);
+
+  return 0;
+}
+
+int HMSMRDevice::write(
+  uint64_t off,
+  bufferlist &bl,
+  bool buffered,
+  int write_hint)
+{
+  uint64_t len = bl.length();
+  dout(20) << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+	   << (buffered ? " (buffered)" : " (direct)")
+	   << dendl;
+  ceph_assert(is_valid_io(off, len));
+  if (cct->_conf->objectstore_blackhole) {
+    lderr(cct) << __func__ << " objectstore_blackhole=true, throwing out IO"
+	       << dendl;
+    return 0;
+  }
+
+  if ((!buffered || bl.get_num_buffers() >= IOV_MAX) &&
+      bl.rebuild_aligned_size_and_memory(block_size, block_size, IOV_MAX)) {
+    dout(20) << __func__ << " rebuilding buffer to be aligned" << dendl;
+  }
+  dout(40) << "data: ";
+  bl.hexdump(*_dout);
+  *_dout << dendl;
+
+  return _sync_write(off, bl, buffered, write_hint);
+}
+
+int HMSMRDevice::aio_write(
+  uint64_t off,
+  bufferlist &bl,
+  IOContext *ioc,
+  bool buffered,
+  int write_hint)
+{
+  uint64_t len = bl.length();
+  dout(20) << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+	   << (buffered ? " (buffered)" : " (direct)")
+	   << dendl;
+  ceph_assert(is_valid_io(off, len));
+  if (cct->_conf->objectstore_blackhole) {
+    lderr(cct) << __func__ << " objectstore_blackhole=true, throwing out IO"
+	       << dendl;
+    return 0;
+  }
+
+  if ((!buffered || bl.get_num_buffers() >= IOV_MAX) &&
+      bl.rebuild_aligned_size_and_memory(block_size, block_size, IOV_MAX)) {
+    dout(20) << __func__ << " rebuilding buffer to be aligned" << dendl;
+  }
+  dout(40) << "data: ";
+  bl.hexdump(*_dout);
+  *_dout << dendl;
+
+  _aio_log_start(ioc, off, len);
+
+#ifdef HAVE_LIBAIO
+  if (aio && dio && !buffered) {
+    if (cct->_conf->bdev_inject_crash &&
+	rand() % cct->_conf->bdev_inject_crash == 0) {
+      derr << __func__ << " bdev_inject_crash: dropping io 0x" << std::hex
+	   << off << "~" << len << std::dec
+	   << dendl;
+      // generate a real io so that aio_wait behaves properly, but make it
+      // a read instead of write, and toss the result.
+      ioc->pending_aios.push_back(aio_t(ioc, choose_fd(false, write_hint)));
+      ++ioc->num_pending;
+      auto& aio = ioc->pending_aios.back();
+      bufferptr p = buffer::create_small_page_aligned(len);
+      aio.bl.append(std::move(p));
+      aio.bl.prepare_iov(&aio.iov);
+      aio.preadv(off, len);
+      ++injecting_crash;
+    } else {
+      if (bl.length() <= RW_IO_MAX) {
+	// fast path (non-huge write)
+	ioc->pending_aios.push_back(aio_t(ioc, choose_fd(false, write_hint)));
+	++ioc->num_pending;
+	auto& aio = ioc->pending_aios.back();
+	bl.prepare_iov(&aio.iov);
+	aio.bl.claim_append(bl);
+	aio.pwritev(off, len);
+	dout(30) << aio << dendl;
+	dout(5) << __func__ << " 0x" << std::hex << off << "~" << len
+		<< std::dec << " aio " << &aio << dendl;
+      } else {
+	// write in RW_IO_MAX-sized chunks
+	uint64_t prev_len = 0;
+	while (prev_len < bl.length()) {
+	  bufferlist tmp;
+	  if (prev_len + RW_IO_MAX < bl.length()) {
+	    tmp.substr_of(bl, prev_len, RW_IO_MAX);
+	  } else {
+	    tmp.substr_of(bl, prev_len, bl.length() - prev_len);
+	  }
+	  auto len = tmp.length();
+	  ioc->pending_aios.push_back(aio_t(ioc, choose_fd(false, write_hint)));
+	  ++ioc->num_pending;
+	  auto& aio = ioc->pending_aios.back();
+	  tmp.prepare_iov(&aio.iov);
+	  aio.bl.claim_append(tmp);
+	  aio.pwritev(off + prev_len, len);
+	  dout(30) << aio << dendl;
+	  dout(5) << __func__ << " 0x" << std::hex << off + prev_len
+		  << "~" << len
+		  << std::dec << " aio " << &aio << " (piece)" << dendl;
+	  prev_len += len;
+	}
+      }
+    }
+  } else
+#endif
+  {
+    int r = _sync_write(off, bl, buffered, write_hint);
+    _aio_log_finish(ioc, off, len);
+    if (r < 0)
+      return r;
+  }
+  return 0;
+}
+
+int HMSMRDevice::discard(uint64_t offset, uint64_t len)
+{
+  int r = 0;
+  if (cct->_conf->objectstore_blackhole) {
+    lderr(cct) << __func__ << " objectstore_blackhole=true, throwing out IO"
+	       << dendl;
+    return 0;
+  }
+  if (support_discard) {
+      dout(10) << __func__
+	       << " 0x" << std::hex << offset << "~" << len << std::dec
+	       << dendl;
+
+      r = BlkDev{fd_directs[WRITE_LIFE_NOT_SET]}.discard((int64_t)offset, (int64_t)len);
+  }
+  return r;
+}
+
+int HMSMRDevice::read(uint64_t off, uint64_t len, bufferlist *pbl,
+		      IOContext *ioc,
+		      bool buffered)
+{
+  dout(5) << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+	  << (buffered ? " (buffered)" : " (direct)")
+	  << dendl;
+  ceph_assert(is_valid_io(off, len));
+
+  _aio_log_start(ioc, off, len);
+
+  auto start1 = mono_clock::now();
+
+  auto p = buffer::ptr_node::create(buffer::create_small_page_aligned(len));
+  int r = ::pread(buffered ? fd_buffereds[WRITE_LIFE_NOT_SET] : fd_directs[WRITE_LIFE_NOT_SET],
+		  p->c_str(), len, off);
+  auto age = cct->_conf->bdev_debug_aio_log_age;
+  if (mono_clock::now() - start1 >= make_timespan(age)) {
+    derr << __func__ << " stalled read "
+         << " 0x" << std::hex << off << "~" << len << std::dec
+         << (buffered ? " (buffered)" : " (direct)")
+	 << " since " << start1 << ", timeout is "
+	 << age
+	 << "s" << dendl;
+  }
+
+  if (r < 0) {
+    if (ioc->allow_eio && is_expected_ioerr(r)) {
+      r = -EIO;
+    } else {
+      r = -errno;
+    }
+    goto out;
+  }
+  ceph_assert((uint64_t)r == len);
+  pbl->push_back(std::move(p));
+
+  dout(40) << "data: ";
+  pbl->hexdump(*_dout);
+  *_dout << dendl;
+
+ out:
+  _aio_log_finish(ioc, off, len);
+  return r < 0 ? r : 0;
+}
+
+int HMSMRDevice::aio_read(
+  uint64_t off,
+  uint64_t len,
+  bufferlist *pbl,
+  IOContext *ioc)
+{
+  dout(5) << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+	  << dendl;
+
+  int r = 0;
+#ifdef HAVE_LIBAIO
+  if (aio && dio) {
+    ceph_assert(is_valid_io(off, len));
+    _aio_log_start(ioc, off, len);
+    ioc->pending_aios.push_back(aio_t(ioc, fd_directs[WRITE_LIFE_NOT_SET]));
+    ++ioc->num_pending;
+    aio_t& aio = ioc->pending_aios.back();
+    bufferptr p = buffer::create_small_page_aligned(len);
+    aio.bl.append(std::move(p));
+    aio.bl.prepare_iov(&aio.iov);
+    aio.preadv(off, len);
+    dout(30) << aio << dendl;
+    pbl->append(aio.bl);
+    dout(5) << __func__ << " 0x" << std::hex << off << "~" << len
+	    << std::dec << " aio " << &aio << dendl;
+  } else
+#endif
+  {
+    r = read(off, len, pbl, ioc, false);
+  }
+
+  return r;
+}
+
+int HMSMRDevice::direct_read_unaligned(uint64_t off, uint64_t len, char *buf)
+{
+  uint64_t aligned_off = p2align(off, block_size);
+  uint64_t aligned_len = p2roundup(off+len, block_size) - aligned_off;
+  bufferptr p = buffer::create_small_page_aligned(aligned_len);
+  int r = 0;
+
+  auto start1 = mono_clock::now();
+  r = ::pread(fd_directs[WRITE_LIFE_NOT_SET], p.c_str(), aligned_len, aligned_off);
+  auto age = cct->_conf->bdev_debug_aio_log_age;
+  if (mono_clock::now() - start1 >= make_timespan(age)) {
+    derr << __func__ << " stalled read "
+         << " 0x" << std::hex << off << "~" << len << std::dec
+	 << " since " << start1 << ", timeout is "
+	 << age
+	 << "s" << dendl;
+  }
+
+  if (r < 0) {
+    r = -errno;
+    derr << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+      << " error: " << cpp_strerror(r) << dendl;
+    goto out;
+  }
+  ceph_assert((uint64_t)r == aligned_len);
+  memcpy(buf, p.c_str() + (off - aligned_off), len);
+
+  dout(40) << __func__ << " data: ";
+  bufferlist bl;
+  bl.append(buf, len);
+  bl.hexdump(*_dout);
+  *_dout << dendl;
+
+ out:
+  return r < 0 ? r : 0;
+}
+
+int HMSMRDevice::read_random(uint64_t off, uint64_t len, char *buf,
+                       bool buffered)
+{
+  dout(5) << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+          << "buffered " << buffered
+	  << dendl;
+  ceph_assert(len > 0);
+  ceph_assert(off < size);
+  ceph_assert(off + len <= size);
+  int r = 0;
+  auto age = cct->_conf->bdev_debug_aio_log_age;
+
+  //if it's direct io and unaligned, we have to use a internal buffer
+  if (!buffered && ((off % block_size != 0)
+                    || (len % block_size != 0)
+                    || (uintptr_t(buf) % CEPH_PAGE_SIZE != 0)))
+    return direct_read_unaligned(off, len, buf);
+
+  auto start1 = mono_clock::now();
+  if (buffered) {
+    //buffered read
+    auto off0 = off;
+    char *t = buf;
+    uint64_t left = len;
+    while (left > 0) {
+      r = ::pread(fd_buffereds[WRITE_LIFE_NOT_SET], t, left, off);
+      if (r < 0) {
+	r = -errno;
+        derr << __func__ << " 0x" << std::hex << off << "~" << left
+          << std::dec << " error: " << cpp_strerror(r) << dendl;
+	goto out;
+      }
+      off += r;
+      t += r;
+      left -= r;
+    }
+    if (mono_clock::now() - start1 >= make_timespan(age)) {
+      derr << __func__ << " stalled read "
+	   << " 0x" << std::hex << off0 << "~" << len << std::dec
+           << " (buffered) since " << start1 << ", timeout is "
+	   << age
+	   << "s" << dendl;
+    }
+  } else {
+    //direct and aligned read
+    r = ::pread(fd_directs[WRITE_LIFE_NOT_SET], buf, len, off);
+    if (mono_clock::now() - start1 >= make_timespan(age)) {
+      derr << __func__ << " stalled read "
+	   << " 0x" << std::hex << off << "~" << len << std::dec
+           << " (direct) since " << start1 << ", timeout is "
+	   << age
+	   << "s" << dendl;
+    }
+    if (r < 0) {
+      r = -errno;
+      derr << __func__ << " direct_aligned_read" << " 0x" << std::hex
+        << off << "~" << left << std::dec << " error: " << cpp_strerror(r)
+        << dendl;
+      goto out;
+    }
+    ceph_assert((uint64_t)r == len);
+  }
+
+  dout(40) << __func__ << " data: ";
+  bufferlist bl;
+  bl.append(buf, len);
+  bl.hexdump(*_dout);
+  *_dout << dendl;
+
+ out:
+  return r < 0 ? r : 0;
+}
+
+int HMSMRDevice::invalidate_cache(uint64_t off, uint64_t len)
+{
+  dout(5) << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+	  << dendl;
+  ceph_assert(off % block_size == 0);
+  ceph_assert(len % block_size == 0);
+  int r = posix_fadvise(fd_buffereds[WRITE_LIFE_NOT_SET], off, len, POSIX_FADV_DONTNEED);
+  if (r) {
+    r = -r;
+    derr << __func__ << " 0x" << std::hex << off << "~" << len << std::dec
+	 << " error: " << cpp_strerror(r) << dendl;
+  }
+  return r;
+}
diff --git a/src/os/bluestore/HMSMRDevice.h b/src/os/bluestore/HMSMRDevice.h
new file mode 100644
index 0000000000000..4fd83db8032b4
--- /dev/null
+++ b/src/os/bluestore/HMSMRDevice.h
@@ -0,0 +1,162 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2014 Red Hat
+ * Copyright (C) 2020 Abutalib Aghayev
+ *
+ * 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.
+ *
+ */
+
+// Copied from KernelDevice with HM-SMR specific functionality added.  Will be
+// further specialized for HM-SMR.
+
+#ifndef CEPH_OS_BLUESTORE_HMSMRDEVICE_H
+#define CEPH_OS_BLUESTORE_HMSMRDEVICE_H
+
+#include <atomic>
+
+#include "include/types.h"
+#include "include/interval_set.h"
+#include "common/Thread.h"
+#include "include/utime.h"
+
+#include "ceph_aio.h"
+#include "BlockDevice.h"
+
+#define RW_IO_MAX (INT_MAX & CEPH_PAGE_MASK)
+
+class HMSMRDevice final : public BlockDevice {
+  std::vector<int> fd_directs, fd_buffereds;
+  bool enable_wrt = true;
+  std::string path;
+  bool aio, dio;
+
+  int vdo_fd = -1;      ///< fd for vdo sysfs directory
+  string vdo_name;
+
+  std::string devname;  ///< kernel dev name (/sys/block/$devname), if any
+
+  ceph::mutex debug_lock = ceph::make_mutex("HMSMRDevice::debug_lock");
+  interval_set<uint64_t> debug_inflight;
+
+  std::atomic<bool> io_since_flush = {false};
+  ceph::mutex flush_mutex = ceph::make_mutex("HMSMRDevice::flush_mutex");
+
+  std::unique_ptr<io_queue_t> io_queue;
+  aio_callback_t discard_callback;
+  void *discard_callback_priv;
+  bool aio_stop;
+  bool discard_started;
+  bool discard_stop;
+
+  ceph::mutex discard_lock = ceph::make_mutex("HMSMRDevice::discard_lock");
+  ceph::condition_variable discard_cond;
+  bool discard_running = false;
+  interval_set<uint64_t> discard_queued;
+  interval_set<uint64_t> discard_finishing;
+
+  struct AioCompletionThread : public Thread {
+    HMSMRDevice *bdev;
+    explicit AioCompletionThread(HMSMRDevice *b) : bdev(b) {}
+    void *entry() override {
+      bdev->_aio_thread();
+      return NULL;
+    }
+  } aio_thread;
+
+  struct DiscardThread : public Thread {
+    HMSMRDevice *bdev;
+    explicit DiscardThread(HMSMRDevice *b) : bdev(b) {}
+    void *entry() override {
+      bdev->_discard_thread();
+      return NULL;
+    }
+  } discard_thread;
+
+  std::atomic_int injecting_crash;
+
+  void _aio_thread();
+  void _discard_thread();
+  int queue_discard(interval_set<uint64_t> &to_release) final;
+
+  int _aio_start();
+  void _aio_stop();
+
+  int _discard_start();
+  void _discard_stop();
+
+  void _aio_log_start(IOContext *ioc, uint64_t offset, uint64_t length);
+  void _aio_log_finish(IOContext *ioc, uint64_t offset, uint64_t length);
+
+  int _sync_write(uint64_t off, bufferlist& bl, bool buffered, int write_hint);
+
+  int _lock();
+
+  int direct_read_unaligned(uint64_t off, uint64_t len, char *buf);
+
+  // stalled aio debugging
+  aio_list_t debug_queue;
+  ceph::mutex debug_queue_lock =
+      ceph::make_mutex("HMSMRDevice::debug_queue_lock");
+  aio_t *debug_oldest = nullptr;
+  utime_t debug_stall_since;
+  void debug_aio_link(aio_t& aio);
+  void debug_aio_unlink(aio_t& aio);
+
+  void _detect_vdo();
+  int choose_fd(bool buffered, int write_hint) const;
+
+  bool set_smr_params(const std::string& path);
+
+public:
+  HMSMRDevice(CephContext* cct, aio_callback_t cb, void *cbpriv,
+              aio_callback_t d_cb, void *d_cbpriv);
+
+  void aio_submit(IOContext *ioc) final;
+  void discard_drain() final;
+
+  int collect_metadata(const std::string& prefix,
+                       map<std::string,std::string> *pm) const final;
+  int get_devname(std::string *s) const final {
+    if (devname.empty()) {
+      return -ENOENT;
+    }
+    *s = devname;
+    return 0;
+  }
+  int get_devices(std::set<std::string> *ls) const final;
+
+  bool is_smr() const final { return true; }
+
+  bool get_thin_utilization(uint64_t *total, uint64_t *avail) const final;
+
+  int read(uint64_t off, uint64_t len, bufferlist *pbl,
+	   IOContext *ioc,
+	   bool buffered) final;
+  int aio_read(uint64_t off, uint64_t len, bufferlist *pbl,
+	       IOContext *ioc) final;
+  int read_random(uint64_t off, uint64_t len, char *buf,
+                  bool buffered) final;
+
+  int write(uint64_t off, bufferlist& bl, bool buffered,
+            int write_hint = WRITE_LIFE_NOT_SET) final;
+  int aio_write(uint64_t off, bufferlist& bl,
+		IOContext *ioc,
+		bool buffered,
+		int write_hint = WRITE_LIFE_NOT_SET) final;
+  int flush() final;
+  int discard(uint64_t offset, uint64_t len) final;
+
+  // for managing buffered readers/writers
+  int invalidate_cache(uint64_t off, uint64_t len) final;
+  int open(const std::string& path) final;
+  void close() final;
+};
+
+#endif
diff --git a/src/os/bluestore/ZonedAllocator.cc b/src/os/bluestore/ZonedAllocator.cc
new file mode 100644
index 0000000000000..90468e4f00a8b
--- /dev/null
+++ b/src/os/bluestore/ZonedAllocator.cc
@@ -0,0 +1,161 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+// 
+// A simple allocator that just hands out space from the next empty zone.  This
+// is temporary, just to get the simplest append-only write workload to work.
+//
+// Copyright (C) 2020 Abutalib Aghayev
+//
+
+#include "ZonedAllocator.h"
+#include "bluestore_types.h"
+#include "common/debug.h"
+
+#define dout_context cct
+#define dout_subsys ceph_subsys_bluestore
+#undef dout_prefix
+#define dout_prefix *_dout << "ZonedAllocator " << this << " "
+
+ZonedAllocator::ZonedAllocator(CephContext* cct,
+			       int64_t size,
+			       int64_t block_size,
+			       const std::string& name)
+    : Allocator(name),
+      cct(cct),
+      num_free(0),
+      size(size),
+      block_size((block_size & 0x00000000ffffffff)),
+      zone_size(((block_size & 0x0000ffff00000000) >> 32) * 1024 * 1024),
+      starting_zone((block_size & 0xffff000000000000) >> 48),
+      nr_zones(size / zone_size),
+      write_pointers(nr_zones) {
+  ldout(cct, 10) << __func__ << " size 0x" << std::hex << size
+		 << " zone size 0x" << zone_size << std::dec
+		 << " number of zones " << nr_zones
+                 << " first sequential zone " << starting_zone
+		 << dendl;
+  ceph_assert(size % zone_size == 0);
+}
+
+ZonedAllocator::~ZonedAllocator() {}
+
+int64_t ZonedAllocator::allocate(
+  uint64_t want_size,
+  uint64_t alloc_unit,
+  uint64_t max_alloc_size,
+  int64_t hint,
+  PExtentVector *extents) {
+  std::lock_guard l(lock);
+
+  ceph_assert(want_size % 4096 == 0);
+
+  ldout(cct, 10) << __func__ << " trying to allocate "
+		 << std::hex << want_size << dendl;
+
+  uint64_t zone = starting_zone;
+  for ( ; zone < nr_zones; ++zone) {
+    if (fits(want_size, zone))
+      break;
+    ldout(cct, 10) << __func__ << " skipping zone " << zone
+		   << " because there is not enough space: "
+		   << " want_size = " << want_size
+		   << " available = " << zone_free_space(zone) << dendl;
+  }
+
+  if (zone == nr_zones) {
+    ldout(cct, 10) << __func__ << " failed to allocate" << dendl;
+    return -ENOSPC;
+  }
+
+  uint64_t offset = zone_offset(zone);
+  ldout(cct, 10) << __func__ << " advancing zone " << zone
+		 << " write pointer from " << std::hex << offset
+		 << " to " << offset + want_size << dendl;
+  advance_wp(zone, want_size);
+
+  if (zone_free_space(zone) == 0) {
+    starting_zone = zone + 1;
+  }
+
+  ldout(cct, 10) << __func__ << " zone " << zone << " offset is now "
+		 << std::hex << zone_wp(zone) << dendl;
+
+  ldout(cct, 10) << __func__ << " allocated " << std::hex << want_size
+		 << " bytes at offset " << offset
+		 << " located at zone " << zone
+		 << " and zone offset " << offset % zone_size << dendl;
+
+  extents->emplace_back(bluestore_pextent_t(offset, want_size));
+  return want_size;
+}
+
+void ZonedAllocator::release(const interval_set<uint64_t>& release_set) {
+  std::lock_guard l(lock);
+}
+
+uint64_t ZonedAllocator::get_free() {
+  std::lock_guard l(lock);
+  return num_free;
+}
+
+void ZonedAllocator::dump() {
+  std::lock_guard l(lock);
+}
+
+void ZonedAllocator::dump(std::function<void(uint64_t offset,
+					     uint64_t length)> notify) {
+  std::lock_guard l(lock);
+}
+
+void ZonedAllocator::init_add_free(uint64_t offset, uint64_t length) {
+  std::lock_guard l(lock);
+  ldout(cct, 10) << __func__ << " " << std::hex
+		 << offset << "~" << length << dendl;
+
+  num_free += length;
+  uint64_t zone = offset / zone_size;
+  offset %= zone_size;
+  write_pointers[zone] = offset;
+  ldout(cct, 10) << __func__ << " set zone " << std::hex
+		 << zone << " write pointer to 0x" << offset << dendl;
+
+  length -= zone_size - offset;
+  ceph_assert(length % zone_size == 0);
+
+  for ( ; length; length -= zone_size) {
+    write_pointers[++zone] = 0;
+    ldout(cct, 30) << __func__ << " set zone 0x" << std::hex
+		   << zone << " write pointer to 0x" << 0 << dendl;
+  }
+}
+
+void ZonedAllocator::init_rm_free(uint64_t offset, uint64_t length) {
+  std::lock_guard l(lock);
+  ldout(cct, 10) << __func__ << " 0x" << std::hex
+		 << offset << "~" << length << dendl;
+
+  num_free -= length;
+  ceph_assert(num_free >= 0);
+
+  uint64_t zone = offset / zone_size;
+  offset %= zone_size;
+  ceph_assert(write_pointers[zone] == offset);
+  write_pointers[zone] = zone_size;
+  ldout(cct, 10) << __func__ << " set zone 0x" << std::hex
+		 << zone << " write pointer to 0x" << zone_size << dendl;
+
+  length -= zone_size - offset;
+  ceph_assert(length % zone_size == 0);
+
+  for ( ; length; length -= zone_size) {
+    write_pointers[++zone] = zone_size;
+    ldout(cct, 10) << __func__ << " set zone 0x" << std::hex
+		   << zone << " write pointer to 0x" << zone_size << dendl;
+  }
+}
+
+
+void ZonedAllocator::shutdown() {
+  ldout(cct, 1) << __func__ << dendl;
+}
diff --git a/src/os/bluestore/ZonedAllocator.h b/src/os/bluestore/ZonedAllocator.h
new file mode 100644
index 0000000000000..2851c6e466a96
--- /dev/null
+++ b/src/os/bluestore/ZonedAllocator.h
@@ -0,0 +1,89 @@
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+// 
+// A simple allocator that just hands out space from the next empty zone.  This
+// is temporary, just to get the simplest append-only write workload to work.
+//
+// Copyright (C) 2020 Abutalib Aghayev
+//
+
+#ifndef CEPH_OS_BLUESTORE_ZONEDALLOCATOR_H
+#define CEPH_OS_BLUESTORE_ZONEDALLOCATOR_H
+
+#include <mutex>
+
+#include "Allocator.h"
+#include "common/ceph_mutex.h"
+#include "include/btree_map.h"
+#include "include/interval_set.h"
+#include "include/mempool.h"
+#include "os/bluestore/bluestore_types.h"
+
+class ZonedAllocator : public Allocator {
+  CephContext* cct;
+
+  // Currently only one thread at a time calls into ZonedAllocator due to
+  // atomic_alloc_and_submit_lock in BlueStore.cc, but we do locking anyway
+  // because eventually ZONE_APPEND support will land and
+  // atomic_alloc_and_submit_lock will be removed.
+  ceph::mutex lock = ceph::make_mutex("ZonedAllocator::lock");
+
+  int64_t num_free;     ///< total bytes in freelist
+  uint64_t size;
+  uint64_t block_size;
+  uint64_t zone_size;
+  uint64_t starting_zone;
+  uint64_t nr_zones;
+  std::vector<uint64_t> write_pointers;
+
+  inline uint64_t zone_offset(uint64_t zone) {
+    ceph_assert(zone < nr_zones);
+    return zone * zone_size + zone_wp(zone);
+  }
+
+  inline uint64_t zone_wp(uint64_t zone) {
+    ceph_assert(zone < nr_zones);
+    return write_pointers[zone];
+  }
+
+  inline uint64_t zone_free_space(uint64_t zone) {
+    ceph_assert(zone < nr_zones);
+    return zone_size - zone_wp(zone);
+  }
+
+  inline void advance_wp(uint64_t zone, uint64_t size) {
+    ceph_assert(zone < nr_zones);
+    write_pointers[zone] += size;
+    ceph_assert(write_pointers[zone] <= zone_size);
+  }
+
+  inline bool fits(uint64_t want_size, uint64_t zone) {
+    ceph_assert(zone < nr_zones);
+    return want_size <= zone_free_space(zone);
+  }
+
+public:
+  ZonedAllocator(CephContext* cct, int64_t size, int64_t block_size,
+                 const std::string& name);
+  ~ZonedAllocator() override;
+
+  int64_t allocate(
+    uint64_t want_size, uint64_t alloc_unit, uint64_t max_alloc_size,
+    int64_t hint, PExtentVector *extents) override;
+
+  void release(const interval_set<uint64_t>& release_set) override;
+
+  uint64_t get_free() override;
+
+  void dump() override;
+  void dump(std::function<void(uint64_t offset,
+                               uint64_t length)> notify) override;
+
+  void init_add_free(uint64_t offset, uint64_t length) override;
+  void init_rm_free(uint64_t offset, uint64_t length) override;
+
+  void shutdown() override;
+};
+
+#endif
diff --git a/src/vstart.sh b/src/vstart.sh
index c3e1880a3f402..ff03a7bde6a53 100755
--- a/src/vstart.sh
+++ b/src/vstart.sh
@@ -162,6 +162,7 @@ rgw_frontend="beast"
 rgw_compression=""
 lockdep=${LOCKDEP:-1}
 spdk_enabled=0 #disable SPDK by default
+zoned_enabled=0
 pci_id=""
 
 with_mgr_dashboard=true
@@ -228,6 +229,7 @@ usage=$usage"\t--msgr21: use msgr2 and msgr1\n"
 usage=$usage"\t--crimson: use crimson-osd instead of ceph-osd\n"
 usage=$usage"\t--osd-args: specify any extra osd specific options\n"
 usage=$usage"\t--bluestore-devs: comma-separated list of blockdevs to use for bluestore\n"
+usage=$usage"\t--bluestore-zoned: blockdevs listed by --bluestore-devs are zoned devices (HM-SMR HDD or ZNS SSD)\n"
 usage=$usage"\t--inc-osd: append some more osds into existing vcluster\n"
 usage=$usage"\t--cephadm: enable cephadm orchestrator with ~/.ssh/id_rsa[.pub]\n"
 usage=$usage"\t--no-parallel: dont start all OSDs in parallel\n"
@@ -439,6 +441,9 @@ case $1 in
         done
         shift
         ;;
+    --bluestore-zoned )
+        zoned_enabled=1
+        ;;
     * )
         usage_exit
 esac
@@ -655,6 +660,14 @@ EOF
         bluestore block wal size = 1048576000
         bluestore block wal create = true"
         fi
+        if [ "$zoned_enabled" -eq 1 ]; then
+            BLUESTORE_OPTS="${BLUESTORE_OPTS}
+        bluestore min alloc size = 65536
+        bluestore prefer deferred size = 0
+        bluestore prefer deferred size hdd = 0
+        bluestore prefer deferred size ssd = 0
+        bluestore allocator = zoned"
+        fi
     fi
     wconf <<EOF
 [client]