sst.c

/*
 * Copyright (c) 2016--2021  Wu, Xingbo <wuxb45@gmail.com>
 *
 * All rights reserved. No warranty, explicit or implicit, provided.
 */
#define _GNU_SOURCE

// headers {{{
#include "lib.h"
#include "ctypes.h"
#include "kv.h"
#include <assert.h> // static_assert
#include <dirent.h> // opendir
#include <sys/uio.h> // writev
#include "sst.h"

// }}} headers

// define {{{
#define SST_MAX_PPB    ((16)) // up to 16 pages per block
#define SST_MAX_BLKSZ  ((PGSZ * SST_MAX_PPB)) // 64kB
#define SST_MAX_KVSZ   ((SST_MAX_BLKSZ - (sizeof(u16) * 2)))
#define SST_MAX_PAGEID ((UINT16_MAX - SST_MAX_PPB))
#define MSST_NRUN      ((16)) // up to 16 runs

#define SSTY_STALE     ((0x80u))
#define SSTY_TOMBSTONE ((0x40u))
#define SSTY_RANK      ((0x3fu))
#define SSTY_INVALID   ((0xffu))

// 4 for 16; 5 for 32; TODO: 6 for 64
#define SSTY_DBITS ((5))
#define SSTY_DIST ((1u << SSTY_DBITS))
static_assert(SSTY_DBITS >= 4 && SSTY_DBITS <= 5, "Supported SSTY_DBITS: 4 and 5; TODO: 6");

#if defined(__linux__)
#define SSTY_MMAP_FLAGS ((MAP_PRIVATE|MAP_POPULATE))
#else
#define SSTY_MMAP_FLAGS ((MAP_PRIVATE))
#endif

// turn on IO-optimized binary search by default; comment out to disable
#define MSSTY_SEEK_BISECT_OPT

const struct msstz_cfg msstz_cfg_default = {
  .major_switch = 2,
  .major_trigger = 8,
  .estimate_safe = 14,
  .max_pages = 20400, // maximum number of pages in a run
  .tgt_pages = 1, // target number of pages in a block
  .pages_accept = 5000, // tune with max_pages
  .ckeys = true,
  .tags = true,
};
// }}} define

// kv {{{
  inline size_t
sst_kv_vi128_estimate(const struct kv * const kv)
{
  return vi128_estimate_u32(kv->klen) + vi128_estimate_u32(kv->vlen) + kv->klen + (kv->vlen & SST_VLEN_MASK);
}

  u8 *
sst_kv_vi128_encode(u8 * ptr, const struct kv * const kv)
{
  ptr = vi128_encode_u32(ptr, kv->klen);
  ptr = vi128_encode_u32(ptr, kv->vlen);
  const u32 kvlen = kv->klen + (kv->vlen & SST_VLEN_MASK);
  memcpy(ptr, kv->kv, kvlen);
  return ptr + kvlen;
}

  inline size_t
sst_kv_size(const struct kv * const kv)
{
  return sizeof(*kv) + kv->klen + (kv->vlen & SST_VLEN_MASK);
}

  struct kv *
sst_kvref_dup2_kv(struct kvref * const kvref, struct kv * const out)
{
  const size_t sz = sst_kv_size(&kvref->hdr);
  struct kv * const new = out ? out : malloc(sz);
  if (new) {
    *new = kvref->hdr;
    memcpy(new->kv, kvref->kptr, new->klen);
    memcpy(new->kv + new->klen, kvref->vptr, new->vlen & SST_VLEN_MASK);
  }
  return new;
}
// }}} kv

// mm {{{
  struct kv *
kvmap_mm_in_ts(struct kv * const kv, void * const priv)
{
  (void)priv;
  if (kv == NULL)
    return NULL;

  const size_t sz = sst_kv_size(kv);
  struct kv * const new = malloc(sz);
  if (new)
    memcpy(new, kv, sz);
  return new;
}

  struct kv *
kvmap_mm_out_ts(struct kv * const kv, struct kv * const out)
{
  if (kv == NULL)
    return NULL;
  const size_t sz = sst_kv_size(kv);
  struct kv * const new = out ? out : malloc(sz);
  if (new)
    memcpy(new, kv, sz);
  return new;
}

// for memtable
const struct kvmap_mm kvmap_mm_ts = {
  .in = kvmap_mm_in_ts,
  .out = kvmap_mm_out_ts,
  .free = kvmap_mm_free_free,
  .priv = NULL,
};
// }}} mm

// sst {{{
struct sst_blkmeta { // the first two bytes in each block
  u8 nkeys; // number of keys
  u8 npages; // number of 4K pages
};

struct sst_meta {
  u32 nblks; // == 0 for empty sst in the place of bms[0]
  u32 npages;
  u64 seq; // the original seq; a linking will generate greater seq in file name
  u32 run; // this is always valid after linking
  u32 totkv;
  u32 bmsoff;
  u32 ioffsoff;
  u32 ckeysoff;
  u32 ckeyssz;
};

struct sst {
  const struct sst_blkmeta * bms; // block metadata (2-byte each)
  u32 nblks; // number of index keys in ioffs
  u32 npages; // number of 4kB data blocks
  int fd;
  u32 refcnt; // not atomic; an sst can be referenced by multiple msst (versions of partitions)
  struct rcache * rc;
  const u32 * ioffs; // offsets of the index keys
  u8 * mem; // pointer to the mmap area
  u32 fsize;
  u32 totkv;
};

  static bool
sst_init_at(const int dfd, const u64 seq, const u32 run, struct sst * const sst)
{
  char fn[24];
  const u64 magic = seq * 100lu + run;
  sprintf(fn, "%03lu.sstx", magic);
  const int fd = openat(dfd, fn, O_RDONLY);
  if (fd < 0)
    return false;

  const size_t fsize = fdsize(fd);
  if (fsize == 0 || fsize >= UINT32_MAX) {
    close(fd);
    return false;
  }

  // Hugepages make replacement hard; some file systems don't support hugepages
  //MAP_HUGETLB|MAP_HUGE_2MB
  u8 * const mem = mmap(NULL, fsize, PROT_READ, MAP_PRIVATE, fd, 0);
  if (mem == MAP_FAILED)
    return false;

  debug_assert(mem);
  sst->mem = mem;
  sst->fsize = (u32)fsize;
  const struct sst_meta * const meta = sst_meta(sst);
  debug_assert((meta->seq < seq) || ((meta->seq == seq) && (meta->run == run)));
  sst->bms = (typeof(sst->bms))(mem + meta->bmsoff);
  sst->nblks = meta->nblks;
  sst->npages = meta->npages;
  sst->fd = fd; // keep fd open
  sst->refcnt = 1;
  sst->rc = NULL;
  sst->ioffs = (typeof(sst->ioffs))(mem + meta->ioffsoff);
  sst->totkv = meta->totkv;
  //const u32 datasz = PGSZ * sst->npages;
  //madvise(mem, datasz, MADV_RANDOM);
  //pages_lock(mem + datasz, fsize - datasz); // mlock the metadata area; not necessary with ssty
  pages_lock((void *)sst->bms, sizeof(sst->bms[0]) * meta->npages); // mlock the bms
  return true;
}

  const struct sst_meta *
sst_meta(struct sst * const sst)
{
  const struct sst_meta * const meta = (typeof(meta))(sst->mem + sst->fsize - sizeof(*meta));
  return meta;
}

  inline void
sst_rcache(struct sst * const sst, struct rcache * const rc)
{
  sst->rc = rc;
}

  static struct sst *
sst_open_at(const int dfd, const u64 seq, const u32 run)
{
  struct sst * const sst = yalloc(sizeof(*sst));
  if (sst == NULL)
    return NULL;
  if (sst_init_at(dfd, seq, run, sst)) {
    return sst;
  } else {
    free(sst);
    return NULL;
  }
}

  struct sst *
sst_open(const char * const dirname, const u64 seq, const u32 run)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return NULL;
  struct sst * const sst = sst_open_at(dfd, seq, run);
  close(dfd);
  return sst;
}

  static inline u32
k128_search_le(const u8 * const base, const u32 * const ioffs, const struct kref * const key, u32 l, u32 r)
{
  while ((l + 1) < r) {
    const u32 m = (l + r) >> 1;
    const int cmp = kref_k128_compare(key, base + ioffs[m]);
    if (cmp < 0) // search-key < [m]
      r = m;
    else if (cmp > 0) // search-key > [m]
      l = m;
    else
      return m;
  }
  return l;
}

  static u16
sst_search_pageid(struct sst * const map, const struct kref * const key)
{
  const u32 ikeyid = k128_search_le(map->mem + sizeof(u16), map->ioffs, key, 0, map->nblks);
  const u16 pageid = *(const u16 *)(map->mem + map->ioffs[ikeyid]);
  return pageid;
}

// access data blocks from here
  static inline const u8 *
sst_blk_acquire(struct sst * const map, const u32 pageid)
{
  if (map->rc && (map->bms[pageid].npages == 1)) {
    const u8 * const ptr = rcache_acquire(map->rc, map->fd, pageid);
    debug_assert(ptr);
    return ptr;
  }
  return map->mem + (PGSZ * pageid);
}

  static inline u64
sst_blk_retain(struct rcache * const rc, const u8 * blk)
{
  debug_assert(blk && (((u64)blk) & 0xffflu) == 0);
  if (rc && (blk[1] == 1))
    rcache_retain(rc, blk);
  return (u64)blk;
}

  static inline void
sst_blk_release(struct rcache * const rc, const u8 * blk)
{
  debug_assert(blk && (((u64)blk) & 0xffflu) == 0);
  if (rc && (blk[1] == 1))
    rcache_release(rc, blk);
}

// the highest bit is set if there is a match
// return 0 to nkeys (low bits)
  static u32
sst_search_block_ge(const u8 * const blk, const struct kref * const key)
{
  const u16 * const offs = (typeof(offs))(blk+sizeof(u16));
  u32 l = 0;
  u32 r = ((struct sst_blkmeta *)blk)->nkeys; // blkmeta.nkeys
  while (l < r) {
    const u32 m = (l + r) >> 1;
    const int cmp = kref_kv128_compare(key, blk + offs[m]);
    if (cmp < 0)
      r = m;
    else if (cmp > 0)
      l = m + 1;
    else
      return m | (1u << 31); // match
  }
  return l;
}

  static inline const u8 *
sst_blk_get_kvptr(const u8 * const blk, const u32 id)
{
  debug_assert(id < blk[0]);
  const u16 * const offs = (typeof(offs))(blk + sizeof(u8) + sizeof(u8));
  return blk + offs[id];
}

  struct kv *
sst_get(struct sst * const map, const struct kref * const key, struct kv * const out)
{
  const u16 pageid = sst_search_pageid(map, key);

  // search in the block
  const u8 * const blk = sst_blk_acquire(map, pageid);
  const u32 r = sst_search_block_ge(blk, key);
  if ((r >> 31) == 0) { // not found
    sst_blk_release(map->rc, blk);
    return NULL;
  }

  // found
  const u8 * ptr = sst_blk_get_kvptr(blk, r & 0xffffu);
  u32 klen, vlen;
  ptr = vi128_decode_u32(ptr, &klen);
  ptr = vi128_decode_u32(ptr, &vlen);

  const u32 vlen1 = vlen & SST_VLEN_MASK;
  const u32 kvlen = klen + vlen1;
  struct kv * const ret = out ? out : malloc(sizeof(*ret) + kvlen);
  ret->klen = klen;
  ret->vlen = vlen;
  memcpy(ret->kv, ptr, kvlen);
  sst_blk_release(map->rc, blk);
  return ret;
}

  bool
sst_probe(struct sst * const map, const struct kref * const key)
{
  const u16 pageid = sst_search_pageid(map, key);

  // search in the block
  const u8 * const blk = sst_blk_acquire(map, pageid);
  const u32 r = sst_search_block_ge(blk, key);
  sst_blk_release(map->rc, blk);
  return (r >> 31);
}

  struct kv *
sst_first_key(struct sst * const map, struct kv * const out)
{
  if (map->npages == 0)
    return NULL;

  const u8 * const blk = sst_blk_acquire(map, 0);
  const u8 * ptr = sst_blk_get_kvptr(blk, 0);
  u32 klen, vlen;
  ptr = vi128_decode_u32(ptr, &klen);
  ptr = vi128_decode_u32(ptr, &vlen);
  struct kv * const ret = out ? out : malloc(sizeof(*ret) + klen);
  ret->klen = klen;
  ret->vlen = 0;
  memcpy(ret->kv, ptr, klen);
  sst_blk_release(map->rc, blk);
  return ret;
}

  struct kv *
sst_last_key(struct sst * const map, struct kv * const out)
{
  if (map->npages == 0)
    return NULL;

  u32 bmsi = map->npages-1;
  while (map->bms[bmsi].npages == 0)
    bmsi--;
  debug_assert(bmsi < map->npages);

  const u8 * const blk = sst_blk_acquire(map, bmsi);
  const u8 * ptr = sst_blk_get_kvptr(blk, map->bms[bmsi].nkeys-1);
  u32 klen, vlen;
  ptr = vi128_decode_u32(ptr, &klen);
  ptr = vi128_decode_u32(ptr, &vlen);
  struct kv * const ret = out ? out : malloc(sizeof(*ret) + klen);
  ret->klen = klen;
  ret->vlen = 0;
  memcpy(ret->kv, ptr, klen);
  sst_blk_release(map->rc, blk);
  return ret;
}

  static void
sst_deinit(struct sst * const map)
{
  if (map->refcnt == 1) {
    debug_assert(map->mem);
    munmap((void *)map->mem, map->fsize);
    if (map->rc)
      rcache_close(map->rc, map->fd);
    else
      close(map->fd);
  } else {
    map->refcnt--;
  }
}

  static void
sst_deinit_lazy(struct sst * const map)
{
  if (map->refcnt == 1) {
    debug_assert(map->mem);
    munmap((void *)map->mem, map->fsize);
    if (map->rc)
      rcache_close_lazy(map->rc, map->fd);
    else
      close(map->fd);
  } else {
    map->refcnt--;
  }
}

  void
sst_destroy(struct sst * const map)
{
  sst_deinit(map);
  free(map);
}

  void
sst_fprint(struct sst * const map, FILE * const out)
{
  fprintf(out, "%s totkv %u nblks %u npages %u filesz %u\n",
      __func__, map->totkv, map->nblks, map->npages, map->fsize);
}
// }}} sst

// kvenc {{{
// an append-only buffer for serialization
// 2MB * 127 = 254 MB
#define KVENC_BUFSZ ((1u << 21))
#define KVENC_BUFNR ((127))
struct kvenc {
  u32 idx;
  u32 off;
  u8 * bufs[KVENC_BUFNR];
};

  static struct kvenc *
kvenc_create(void)
{
  return calloc(1, sizeof(struct kvenc));
}

// append raw data of any size
  static void
kvenc_append_raw(struct kvenc * const enc, const void * const data, const u32 size)
{
  u32 off = 0;
  u32 rem = size;
  while (rem) {
    const u32 bufidx = enc->idx;
    debug_assert(bufidx < KVENC_BUFNR);
    if (enc->bufs[bufidx] == NULL)
      enc->bufs[bufidx] = malloc(KVENC_BUFSZ);

    const u32 cpsz = (rem <= (KVENC_BUFSZ - enc->off)) ? rem : (KVENC_BUFSZ - enc->off);
    if (data)
      memcpy(enc->bufs[bufidx] + enc->off, ((u8 *)data) + off, cpsz);
    else
      memset(enc->bufs[bufidx] + enc->off, 0, cpsz);

    rem -= cpsz;
    off += cpsz;
    enc->off += cpsz;
    if (enc->off == KVENC_BUFSZ) {
      enc->idx = bufidx + 1;
      enc->off = 0;
    }
  }
}

// append 1 byte
  static void
kvenc_append_u8(struct kvenc * const enc, const u8 v)
{
  kvenc_append_raw(enc, &v, sizeof(v));
}

// append a u32 (4 bytes in little-endian order)
  static inline void
kvenc_append_u32(struct kvenc * const enc, const u32 val)
{
  kvenc_append_raw(enc, &val, sizeof(val));
}

// append a u16 (2 bytes in little-endian order)
  static inline void
kvenc_append_u16(struct kvenc * const enc, const u16 val)
{
  kvenc_append_raw(enc, &val, sizeof(val));
}

// append a place holder for u32 and return the pointer to the location
  static inline u32 *
kvenc_append_u32_backref(struct kvenc * const enc)
{
  const u32 idx = enc->idx;
  const u32 off = enc->off;
  debug_assert((off + sizeof(u32)) <= KVENC_BUFSZ);
  debug_assert((off % sizeof(u32)) == 0);
  kvenc_append_raw(enc, NULL, sizeof(u32));
  return (u32 *)(enc->bufs[idx] + off);
}

// encode a u32 value with vi128
  static inline void
kvenc_append_vi128(struct kvenc * const enc, const u32 val)
{
  u8 buf[8];
  u8 * const end = vi128_encode_u32(buf, val);
  kvenc_append_raw(enc, buf, (u32)(end - buf));
}

// move the cursor forward to the nearest aligned offset
  static inline void
kvenc_append_padding(struct kvenc * const enc, const u32 power)
{
  debug_assert(power <= 12);
  const u32 p2 = 1u << power;
  const u32 off = enc->off & (p2 - 1);
  if (off)
    kvenc_append_raw(enc, NULL, p2 - off);
}

// return the total size of the bufferred data
  static u32
kvenc_size(struct kvenc * const enc)
{
  return KVENC_BUFSZ * enc->idx + enc->off;
}

// write everything to a file
  static ssize_t
kvenc_write(struct kvenc * const enc, const int fd)
{
  struct iovec vec[KVENC_BUFNR+1];
  const u32 nr = enc->idx;
  for (u32 i = 0; i < nr; i++) {
    vec[i].iov_base = enc->bufs[i];
    vec[i].iov_len = KVENC_BUFSZ;
  }
  vec[nr].iov_base = enc->bufs[nr];
  vec[nr].iov_len = enc->off;
  return writev(fd, vec, (int)(enc->off ? (nr + 1) : nr));
}

// reset the cursor to 0 and free all the buffers
  static void
kvenc_reset(struct kvenc * const enc)
{
  for (u32 i = 0; i < KVENC_BUFNR; i++) {
    if (enc->bufs[i])
      free(enc->bufs[i]);
    else
      break;
  }
  memset(enc, 0, sizeof(*enc));
}

  static void
kvenc_destroy(struct kvenc * const enc)
{
  const u32 nr = enc->idx;
  for (u32 i = 0; i < nr; i++)
    free(enc->bufs[i]);
  if (enc->off)
    free(enc->bufs[nr]);
  free(enc);
}
// }}} kvenc

// sst_build {{{
#define SST_BUILD_METASZ ((sizeof(u16) * 256 * 2))
#define SST_BUILD_BUFSZ ((SST_BUILD_METASZ + SST_MAX_BLKSZ))
#define SST_BUILD_NVEC ((16))

struct sst_build_cfg {
  u64 seq;
  u32 run;
  u32 max_pages; // <= 65520
  u32 tgt_pages; // <= 16
  bool del;
  bool ckeys;
};

// from k0 (inclusive) to kz (exclusive)
// warning: all iters in miter must handle the tombstone (vlen >= SST_VLEN_TS)
// return the output file size (in bytes)
  static u64
sst_build_at(const int dfd, struct miter * const miter,
    const struct sst_build_cfg * const cfg,
    const struct kv * const k0, const struct kv * const kz)
{
  char fn[24];
  const u64 magic = cfg->seq * 100lu + cfg->run;
  sprintf(fn, "%03lu.sstx", magic);
  const int fdout = openat(dfd, fn, O_WRONLY|O_CREAT|O_TRUNC, 00644);
  if (fdout < 0)
    return 0;

  struct kv * const tmp0 = malloc(SST_MAX_BLKSZ);
  kv_refill(tmp0, "", 0, "", 0);
  struct kv * const tmp1 = malloc(SST_MAX_BLKSZ);

  // kv encoding buffers
  struct iovec vecs[SST_BUILD_NVEC];
  u8 * const databufs = malloc(SST_BUILD_BUFSZ * SST_BUILD_NVEC); // kv blocks
  u16 * mbuf = (u16 *)databufs;
  u8 * kvbuf = ((u8 *)mbuf) + SST_BUILD_METASZ;
  u8 * kvcsr = kvbuf;
  u32 vi = 0;

  // max number of 4kB data blocks
  debug_assert(cfg->tgt_pages <= SST_MAX_PPB);
  const u32 tgtblksz = cfg->tgt_pages * PGSZ;
  debug_assert(cfg->max_pages && (cfg->max_pages <= SST_MAX_PAGEID));
  struct sst_blkmeta * const bms = calloc(1, sizeof(bms[0]) * (cfg->max_pages + SST_MAX_PPB));
  u32 keyid = 0; // number of keys in current block
  u32 pageid = 0;
  u32 totkv = 0;
  // at most 65536 ikeys
  u32 * const ioffs = malloc(sizeof(ioffs[0]) * (1lu << 16)); // offsets of ikeys

  struct kvenc * const aenc = kvenc_create();
  struct kvenc * const kenc = kvenc_create();
  u32 nblks = 0;

  if (k0)
    miter_kv_seek(miter, k0);

  do {
    // key in tmp1
    struct kv * curr = miter_peek(miter, tmp1);

    // if del is true then skip all tombstones and stale keys
    if (cfg->del) {
      while (curr && (curr->vlen == SST_VLEN_TS)) {
        miter_skip_unique(miter);
        curr = miter_peek(miter, tmp1);
      }
    }

    // check for termination
    if (curr && kz && (kv_compare(curr, kz) >= 0))
      curr = NULL;

    const size_t est1 = curr ? ((u32)sst_kv_vi128_estimate(curr) + sizeof(u16)) : 0;
    if (est1 > SST_MAX_KVSZ) {
      fprintf(stderr, "WARNING: skip very long kv: size=%zu\n", est1);
      miter_skip_unique(miter);
      continue;
    }
    // estimate the sizes if curr is added to the current block
    const u32 metasz = sizeof(u16) * (keyid + 1);
    const u32 datasz = (u32)(kvcsr - kvbuf);
    const u32 totsz = metasz + datasz;
    const u32 esttot = totsz + (u32)est1;
    // close and write current block if:
    //     no more new data: curr == NULL
    // or: the new keys is not the first key AND total size is more than one page
    // or: current block has too many keys (254)
    // NOTE: a huge key (need 2+ pages) exclusively ocuppy a multi-page block
    if (curr == NULL || (keyid && (esttot > tgtblksz)) || (keyid == 254)) {
      if (keyid == 0)
        break;

      // blksize: whole pages
      const u32 blksize = (u32)bits_round_up(metasz + datasz, PGBITS);
      const u8 npages = (u8)(blksize >> PGBITS); // 1 to 16
      debug_assert(npages && (npages <= SST_MAX_PPB));
      // encode the metadata right before the kvbuf
      u16 * const mbuf1 = (u16 *)(kvbuf - metasz);
      for (u32 i = 0; i < keyid; i++)
        mbuf1[i+1] = mbuf[i] + (u16)metasz;

      // write all metadata: metasz bytes
      // save in the buffer
      struct sst_blkmeta * const pblkmeta = &(bms[pageid]);
      pblkmeta->nkeys = (u8)keyid; // 1 byte # of keys
      pblkmeta->npages = (u8)npages; // 1byte # of 4kB blocks
      mbuf1[0] = *(u16 *)pblkmeta;
      memset(kvcsr, 0, blksize - metasz - datasz); // zero-padding

      struct iovec * const vec = &vecs[vi];
      vec->iov_base = mbuf1;
      vec->iov_len = blksize;
      vi++;
      if (vi == SST_BUILD_NVEC) {
        writev(fdout, vecs, (int)vi); // ignore I/O errors
        vi = 0;
      }
      mbuf = (u16 *)(databufs + (SST_BUILD_BUFSZ * vi));
      kvbuf = ((u8 *)mbuf) + SST_BUILD_METASZ;
      kvcsr = kvbuf;
      keyid = 0;
      pageid += npages;
      // stop processing the next block; break the do-while loop
      if ((curr == NULL) || (pageid >= cfg->max_pages))
        break;
    }

    // the beginning of a block: build anchor key for every head key of block
    if (keyid == 0) {
      ioffs[nblks] = kvenc_size(aenc);
      // block id
      debug_assert(pageid <= SST_MAX_PAGEID);
      kvenc_append_u16(aenc, (u16)pageid);
      // anchor key
      const u32 alen = tmp0->klen ? (kv_key_lcp(tmp0, curr)+1) : 0;
      debug_assert(alen <= curr->klen);
      // encode index key
      kvenc_append_vi128(aenc, alen);
      kvenc_append_raw(aenc, curr->kv, alen);
      nblks++;
    }

    // append kv to data block
    mbuf[keyid++] = (u16)(kvcsr - kvbuf);
    kvcsr = sst_kv_vi128_encode(kvcsr, curr);
    totkv++;
    // copy keys for faster remix building
    if (cfg->ckeys) {
      const u32 lcp = kv_key_lcp(curr, tmp0);
      const u32 slen = curr->klen - lcp;
      kvenc_append_vi128(kenc, lcp); // prefix length
      kvenc_append_vi128(kenc, slen); // suffix length
      kvenc_append_u8(kenc, curr->vlen == SST_VLEN_TS ? 1 : 0);
      kvenc_append_raw(kenc, curr->kv + lcp, slen);
    }
    // remember last key in tmp0
    kv_dup2_key(curr, tmp0);
    miter_skip_unique(miter);
  } while (true);

  if (vi)
    writev(fdout, vecs, (int)vi); // ignore I/O errors

  debug_assert(nblks <= pageid && pageid <= UINT16_MAX);
  // place bms immediately after data blocks
  const u32 bmsoff = PGSZ * pageid;
  const u32 bmssz = sizeof(bms[0]) * pageid;
  // now all data blocks have been written; write one big index block
  // calculate index-key offsets
  const u32 ikeysoff = bmsoff + bmssz; // index keys
  for (u64 i = 0; i < nblks; i++)
    ioffs[i] += ikeysoff;
  // write: index keys; all index-key offsets; # of index-keys
  kvenc_append_padding(aenc, 4);
  kvenc_append_padding(kenc, 4);
  const u32 ikeyssz = kvenc_size(aenc);
  const u32 ioffsoff = ikeysoff + ikeyssz;
  const u32 ioffssz = sizeof(ioffs[0]) * nblks;
  const u32 ckeysoff = ioffsoff + ioffssz;
  const u32 ckeyssz = kvenc_size(kenc);

  // metadata
  struct sst_meta endmeta = {.nblks = nblks, .npages = pageid, .seq = cfg->seq, .run = cfg->run, .totkv = totkv,
    .bmsoff = bmsoff, .ioffsoff = ioffsoff, .ckeysoff = ckeysoff, .ckeyssz = ckeyssz, };
  const u32 endsz = sizeof(endmeta);
  const u64 totsz = ckeysoff + ckeyssz + endsz;

  // sst file layout:
  // 0: data blocks 4kB x pageid     +bmsoff==blkssz
  // bmsoff: blockmetas (bms)        +bmssz[0]
  // ikeysoff: index keys (ikeys)    +ikeyssz[1]
  // ioffsoff: index offsets (ioffs) +ioffssz[2]
  // ?:      endmeta                 +endsz[3]
  // totsz is file size

  //fprintf(stderr, "%s totkv %u nblks %u npages %u bmssz %u ikeyssz %u ioffssz %u\n",
  //    __func__, totkv, nblks, pageid, bmssz, ikeyssz, ioffssz);

  const ssize_t nwbms = write(fdout, bms, bmssz);
  const ssize_t nwanc = kvenc_write(aenc, fdout);
  const ssize_t nwiof = write(fdout, ioffs, ioffssz);
  const ssize_t nwcpy = kvenc_write(kenc, fdout);
  const ssize_t nwmeta = write(fdout, &endmeta, endsz);
  const bool wok = (bmssz + ikeyssz + ioffssz + ckeyssz + endsz) == (nwbms + nwanc + nwiof + nwcpy + nwmeta);

  // done
  fsync(fdout);
  close(fdout);
  free(tmp0);
  free(tmp1);
  free(databufs);
  free(bms);
  free(ioffs);
  kvenc_destroy(aenc);
  kvenc_destroy(kenc);
  return wok ? totsz : 0;
}

  u64
sst_build(const char * const dirname, struct miter * const miter,
    const u64 seq, const u32 run, const u32 max_pages, const u32 tgt_pages,
    const bool del, const bool ckeys,
    const struct kv * const k0, const struct kv * const kz)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return 0;

  const struct sst_build_cfg cfg = {.seq = seq, .run = run,
    .max_pages = max_pages, .tgt_pages = tgt_pages, .del = del, .ckeys = ckeys};
  const u64 ret = sst_build_at(dfd, miter, &cfg, k0, kz);
  close(dfd);
  return ret;
}
// }}} sst_build

// sst_iter {{{
struct sst_ptr {
  u16 pageid; // xth 4kb-block in the table
  u16 keyid; // xth key in the block // MAX == invalid
};

struct sst_iter { // 32 bytes
  struct sst * sst;
  u32 rank; // pure rank value < nrun
  struct sst_ptr ptr;
  u32 klen;
  u32 vlen;
  const u8 * kvdata;
};

  static int
sst_iter_compare(struct sst_iter * const i1, struct sst_iter * const i2)
{
  debug_assert(i1->ptr.keyid != UINT16_MAX);
  debug_assert(i2->ptr.keyid != UINT16_MAX);
  const u32 len = i1->klen < i2->klen ? i1->klen : i2->klen;
  const int cmp = memcmp(i1->kvdata, i2->kvdata, len);
  return cmp ? cmp : (((int)i1->klen) - ((int)i2->klen));
}

// i1 must be valid
// key can be NULL
  static int
sst_iter_compare_kref(struct sst_iter * const iter, const struct kref * const key)
{
  debug_assert(iter->ptr.keyid != UINT16_MAX);
  debug_assert(key);
  const u32 len = (iter->klen < key->len) ? iter->klen : key->len;
  const int cmp = memcmp(iter->kvdata, key->ptr, len);
  if (cmp != 0) {
    return cmp;
  } else {
    return ((int)iter->klen) - ((int)key->len);
  }
}

  static inline bool
sst_iter_match_kref(const struct sst_iter * const i1, const struct kref * const key)
{
  debug_assert(i1->ptr.keyid != UINT16_MAX);
  return (i1->klen == key->len) && (!memcmp(i1->kvdata, key->ptr, i1->klen));
}

  static inline const u8 *
sst_iter_blk_addr(struct sst_iter * const iter)
{
  debug_assert(iter->kvdata);
  const struct sst * const sst = iter->sst;
  if (sst->bms[iter->ptr.pageid].npages == 1) { // single-page block
    const u64 addr = ((u64)iter->kvdata) >> PGBITS << PGBITS;
    return (const u8 *)addr;
  } else { // multi-page block: mmap address
    return sst->mem + (PGSZ * iter->ptr.pageid);
  }
}

  static inline void
sst_iter_blk_release(struct sst_iter * const iter)
{
  if (iter->kvdata) {
    // get page address
    const u8 * const blk = sst_iter_blk_addr(iter);
    sst_blk_release(iter->sst->rc, blk);
    iter->kvdata = NULL;
  }
}

// sst only
// call this function when iter has been moved
  static void
sst_iter_fix_ptr(struct sst_iter * const iter)
{
  const struct sst_blkmeta blkmeta = iter->sst->bms[iter->ptr.pageid];

  // will fix kvdata in peek()
  if (iter->ptr.keyid >= blkmeta.nkeys) { // beyond the current block
    iter->ptr.pageid += blkmeta.npages;
    if (iter->ptr.pageid >= iter->sst->npages) {
      iter->ptr.keyid = UINT16_MAX; // invalid
      return;
    }
    iter->ptr.keyid = 0;
  }
}

  static void
sst_iter_fix_kv_reuse(struct sst_iter * const iter)
{
  // reuse the kvdata and keyid
  debug_assert(iter->kvdata);
  const u8 * const blk = sst_iter_blk_addr(iter);
  const u8 * ptr = sst_blk_get_kvptr(blk, iter->ptr.keyid);
  ptr = vi128_decode_u32(ptr, &iter->klen);
  iter->kvdata = vi128_decode_u32(ptr, &iter->vlen);
}

// make kvdata current with the iter; acquire blk
// also used by mssty
  static void
sst_iter_fix_kv(struct sst_iter * const iter)
{
  // don't fix if invalid or already has the ->kvdata
  if ((!sst_iter_valid(iter)) || iter->kvdata)
    return;

  const u8 * blk = sst_blk_acquire(iter->sst, iter->ptr.pageid);
  const u8 * ptr = sst_blk_get_kvptr(blk, iter->ptr.keyid);
  ptr = vi128_decode_u32(ptr, &iter->klen);
  iter->kvdata = vi128_decode_u32(ptr, &iter->vlen);
}

// points to the first key; invalid for empty sst
  static inline void
sst_iter_init(struct sst_iter * const iter, struct sst * const sst, const u32 rank)
{
  debug_assert(rank < MSST_NRUN);
  iter->sst = sst;
  iter->rank = rank;
  iter->ptr.pageid = (u16)sst->npages;
  iter->ptr.keyid = UINT16_MAX;
  // klen, vlen are ignored
  iter->kvdata = NULL;
}

  struct sst_iter *
sst_iter_create(struct sst * const sst)
{
  struct sst_iter * const iter = calloc(1, sizeof(*iter));
  if (iter == NULL)
    return NULL;
  sst_iter_init(iter, sst, 0);
  return iter;
}

  void
sst_iter_seek(struct sst_iter * const iter, const struct kref * const key)
{
  sst_iter_blk_release(iter);
  // first, find the block
  iter->ptr.pageid = sst_search_pageid(iter->sst, key);
  if (iter->ptr.pageid < iter->sst->npages) {
    // second, find search in the block
    const u8 * const blk = sst_blk_acquire(iter->sst, iter->ptr.pageid);
    iter->ptr.keyid = (u16)sst_search_block_ge(blk, key); // ignoring the high bits
    sst_iter_fix_ptr(iter);
  } else {
    debug_assert(iter->ptr.keyid == UINT16_MAX);
  }
}

  inline void
sst_iter_seek_null(struct sst_iter * const iter)
{
  sst_iter_blk_release(iter);
  iter->ptr.pageid = 0;
  if (iter->sst->npages) {
    iter->ptr.keyid = 0;
    sst_iter_fix_ptr(iter);
    debug_assert(iter->ptr.keyid != UINT16_MAX);
  } else {
    iter->ptr.keyid = UINT16_MAX;
  }
}

  inline bool
sst_iter_valid(struct sst_iter * const iter)
{
  return iter->ptr.keyid != UINT16_MAX;
}

// test if iter points to a tombstone
  inline bool
sst_iter_ts(struct sst_iter * const iter)
{
  sst_iter_fix_kv(iter);
  return iter->vlen == SST_VLEN_TS;
}

  struct kv *
sst_iter_peek(struct sst_iter * const iter, struct kv * const out)
{
  if (!sst_iter_valid(iter))
    return NULL;

  sst_iter_fix_kv(iter);

  const u32 vlen1 = iter->vlen & SST_VLEN_MASK;
  const u32 kvlen = iter->klen + vlen1;
  struct kv * const ret = out ? out : malloc(sizeof(*ret) + kvlen);
  ret->klen = iter->klen;
  ret->vlen = iter->vlen;
  memcpy(ret->kv, iter->kvdata, kvlen);
  return ret;
}

  bool
sst_iter_kref(struct sst_iter * const iter, struct kref * const kref)
{
  if (!sst_iter_valid(iter))
    return false;

  sst_iter_fix_kv(iter);
  kref_ref_raw(kref, iter->kvdata, iter->klen); // no hash32
  return true;
}

  bool
sst_iter_kvref(struct sst_iter * const iter, struct kvref * const kvref)
{
  if (!sst_iter_valid(iter))
    return false;

  sst_iter_fix_kv(iter);
  kvref->hdr.klen = iter->klen;
  kvref->hdr.vlen = iter->vlen;
  kvref->hdr.hash = 0;
  kvref->kptr = iter->kvdata;
  kvref->vptr = iter->kvdata + iter->klen;
  return true;
}

  inline u64
sst_iter_retain(struct sst_iter * const iter)
{
  return sst_blk_retain(iter->sst->rc, sst_iter_blk_addr(iter));
}

  inline void
sst_iter_release(struct sst_iter * const iter, const u64 opaque)
{
  sst_blk_release(iter->sst->rc, (const u8 *)opaque);
}

  void
sst_iter_skip1(struct sst_iter * const iter)
{
  debug_assert(sst_iter_valid(iter));
  const struct sst_blkmeta * const bms = iter->sst->bms;
  struct sst_ptr * const pptr = &iter->ptr;

  pptr->keyid++;
  do {
    if (pptr->keyid < bms[pptr->pageid].nkeys) {
      if (iter->kvdata)
        sst_iter_fix_kv_reuse(iter);
      return; // done
    }
    sst_iter_park(iter); // discard iter->kvdata
    pptr->pageid += bms[pptr->pageid].npages;
    if (pptr->pageid >= iter->sst->npages) {
      pptr->keyid = UINT16_MAX;
      return; // invalid
    }
    pptr->keyid = 0;
  } while (true);
}

// skip using the given blkmeta[]
  void
sst_iter_skip(struct sst_iter * const iter, const u32 nr)
{
  if (!sst_iter_valid(iter))
    return;

  const struct sst_blkmeta * const bms = iter->sst->bms;
  struct sst_ptr * const pptr = &iter->ptr;
  u32 todo = nr;
  do {
    const u32 ncap = bms[pptr->pageid].nkeys - pptr->keyid;
    if (todo < ncap) {
      pptr->keyid += (u16)todo;
      if (iter->kvdata)
        sst_iter_fix_kv_reuse(iter);
      return; // done
    }
    sst_iter_park(iter); // discard iter->kvdata
    pptr->pageid += bms[pptr->pageid].npages;
    if (pptr->pageid >= iter->sst->npages) {
      pptr->keyid = UINT16_MAX;
      return; // invalid
    }
    pptr->keyid = 0;
    todo -= ncap;
  } while (todo);
}

  struct kv *
sst_iter_next(struct sst_iter * const iter, struct kv * const out)
{
  struct kv * const ret = sst_iter_peek(iter, out);
  if (sst_iter_valid(iter))
    sst_iter_skip1(iter);
  return ret;
}

  void
sst_iter_park(struct sst_iter * const iter)
{
  sst_iter_blk_release(iter);
}

  void
sst_iter_destroy(struct sst_iter * const iter)
{
  sst_iter_park(iter);
  free(iter);
}

  void
sst_dump(struct sst * const sst, const char * const fn)
{
  const int fd = open(fn, O_WRONLY|O_CREAT|O_TRUNC, 00644);
  debug_assert(fd >= 0);
  struct sst_iter iter;
  sst_iter_init(&iter, sst, 0);
  sst_iter_seek_null(&iter);
  struct kvref kvref;
  u32 n = 0;
  dprintf(fd, "npages %u totkv %u\n", sst->npages, sst->totkv);
  while (sst_iter_kvref(&iter, &kvref)) {
    dprintf(fd, "%6u b %6u k %6u %.*s (%u,%u)\n",
        n, iter.ptr.pageid, iter.ptr.keyid, iter.klen, iter.kvdata, iter.klen, iter.vlen);
    sst_iter_skip1(&iter);
    n++;
  }
  fsync(fd);
  close(fd);
}
// }}} sst_iter

// msstx {{{
struct msst {
  u64 seq;
  u32 nrun;
  u32 refcnt; // not atomic: -- in msstz_gc(); ++ in append-to-v; no race condition
  struct ssty * ssty; // ssty makes it mssty
  struct rcache * rc;
  struct sst ssts[MSST_NRUN];
};

struct msstx_iter {
  struct msst * msst;
  u32 nrun;
  // minheap
  struct sst_iter * mh[MSST_NRUN+1];
  struct sst_iter iters[MSST_NRUN];
};

  static struct msst *
msstx_open_at_reuse(const int dfd, const u64 seq, const u32 nrun, struct msst * const msst0, const u32 nrun0)
{
  if (nrun > MSST_NRUN)
    return NULL;
  struct msst * const msst = calloc(1, sizeof(*msst));
  if (msst == NULL)
    return NULL;

  debug_assert(nrun0 <= nrun);
  for (u32 i = 0; i < nrun0; i++) {
    debug_assert(msst0->ssts[i].refcnt == 1);
    msst->ssts[i] = msst0->ssts[i];
    // only increment the old's refcnt
    msst0->ssts[i].refcnt++;
  }

  for (u32 i = nrun0; i < nrun; i++) {
    if (!sst_init_at(dfd, seq, i, &(msst->ssts[i]))) {
      // error
      for (u64 j = 0; j < i; j++)
        sst_deinit(&(msst->ssts[j]));

      free(msst);
      return NULL;
    }
  }
  msst->seq = seq;
  msst->nrun = nrun;
  return msst;

}

  static struct msst *
msstx_open_at(const int dfd, const u64 seq, const u32 nrun)
{
  return msstx_open_at_reuse(dfd, seq, nrun, NULL, 0);
}

  inline struct msst *
msstx_open(const char * const dirname, const u64 seq, const u32 nrun)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return NULL;
  struct msst * const msst = msstx_open_at_reuse(dfd, seq, nrun, NULL, 0);
  close(dfd);
  return msst;
}

  inline void
msst_rcache(struct msst * const msst, struct rcache * const rc)
{
  msst->rc = rc;
  for (u32 i = 0; i < msst->nrun; i++)
    sst_rcache(&(msst->ssts[i]), rc);
}

  void
msstx_destroy(struct msst * const msst)
{
  debug_assert(msst->ssty == NULL);
  for (u32 i = 0; i < msst->nrun; i++)
    sst_deinit(&(msst->ssts[i]));
  free(msst);
}

  struct msstx_iter *
msstx_iter_create(struct msst * const msst)
{
  struct msstx_iter * const iter = calloc(1, sizeof(*iter));
  if (iter == NULL)
    return NULL;

  iter->msst = msst;
  iter->nrun = msst->nrun;
  for (u32 i = 0; i < msst->nrun; i++) {
    sst_iter_init(&(iter->iters[i]), &(msst->ssts[i]), i);
    iter->mh[i+1] = &(iter->iters[i]);
  }
  return iter;
}

  struct kv *
msstx_get(struct msst * const msst, const struct kref * const key, struct kv * const out)
{
  for (u32 i = msst->nrun-1; i < msst->nrun; i--) {
    struct kv * const ret = sst_get(&(msst->ssts[i]), key, out);
    if (ret)
      return ret;
  }
  return NULL;
}

  bool
msstx_probe(struct msst * const msst, const struct kref * const key)
{
  for (u32 i = msst->nrun-1; i < msst->nrun; i--)
    if (sst_probe(&(msst->ssts[i]), key))
      return true;
  return false;
}

// mh {{{
  static void
msstx_mh_swap(struct msstx_iter * const iter, const u32 cidx)
{
  debug_assert(cidx > 1);
  struct sst_iter * const tmp = iter->mh[cidx];
  iter->mh[cidx] = iter->mh[cidx>>1];
  iter->mh[cidx>>1] = tmp;
}

  static bool
msstx_mh_should_swap(struct sst_iter * const sp, struct sst_iter * const sc)
{
  debug_assert(sp != sc);
  debug_assert(sp->rank != sc->rank);
  if (!sst_iter_valid(sp))
    return true;
  if (!sst_iter_valid(sc))
    return false;

  const int c = sst_iter_compare(sp, sc);
  if (c > 0)
    return true;
  else if (c < 0)
    return false;
  return sp->rank < sc->rank; // high rank == high priority
}

  static void
msstx_mh_uphead(struct msstx_iter * const iter, u32 idx)
{
  while (idx > 1) {
    struct sst_iter * const sp = iter->mh[idx >> 1];
    struct sst_iter * const sc = iter->mh[idx];
    if (!sst_iter_valid(sc))
      return;
    if (msstx_mh_should_swap(sp, sc))
      msstx_mh_swap(iter, idx);
    else
      return;
    idx >>= 1;
  }
}

  static void
msstx_mh_downheap(struct msstx_iter * const iter, u32 idx)
{
  const u32 nrun = iter->nrun;
  while ((idx<<1) <= nrun) {
    struct sst_iter * sl = iter->mh[idx<<1];
    u32 idxs = idx << 1;
    if ((idx<<1) < nrun) { // has sr
      struct sst_iter * sr = iter->mh[(idx<<1) + 1];
      if (msstx_mh_should_swap(sl, sr))
        idxs++;
    }

    if (msstx_mh_should_swap(iter->mh[idx], iter->mh[idxs]))
      msstx_mh_swap(iter, idxs);
    else
      return;
    idx = idxs;
  }
}
// }}} mh

  bool
msstx_iter_valid(struct msstx_iter * const iter)
{
  return iter->nrun && sst_iter_valid(iter->mh[1]);
}

  static inline bool
msstx_iter_valid_1(struct msstx_iter * const iter)
{
  return iter->nrun != 0;
}

  void
msstx_iter_seek(struct msstx_iter * const iter, const struct kref * const key)
{
  const u32 nrun = iter->nrun;
  for (u32 i = 1; i <= nrun; i++) {
    struct sst_iter * const iter1 = iter->mh[i];
    sst_iter_seek(iter1, key);
    if (sst_iter_valid(iter1))
      sst_iter_fix_kv(iter1);
  }
  for (u32 i = 2; i <= nrun; i++)
    msstx_mh_uphead(iter, i);
}

  void
msstx_iter_seek_null(struct msstx_iter * const iter)
{
  const u32 nrun = iter->nrun;
  for (u32 i = 1; i <= nrun; i++) {
    struct sst_iter * const iter1 = iter->mh[i];
    sst_iter_seek_null(iter1);
    if (sst_iter_valid(iter1))
      sst_iter_fix_kv(iter1);
  }
  for (u32 i = 2; i <= nrun; i++)
    msstx_mh_uphead(iter, i);
}

  struct kv *
msstx_iter_peek(struct msstx_iter * const iter, struct kv * const out)
{
  if (!msstx_iter_valid_1(iter))
    return NULL;
  return sst_iter_peek(iter->mh[1], out);
}

  bool
msstx_iter_kref(struct msstx_iter * const iter, struct kref * const kref)
{
  if (!msstx_iter_valid_1(iter))
    return false;

  return sst_iter_kref(iter->mh[1], kref);
}

  bool
msstx_iter_kvref(struct msstx_iter * const iter, struct kvref * const kvref)
{
  if (!msstx_iter_valid_1(iter))
    return false;

  return sst_iter_kvref(iter->mh[1], kvref);
}

  inline u64
msstx_iter_retain(struct msstx_iter * const iter)
{
  return sst_iter_retain(iter->mh[1]);
}

  inline void
msstx_iter_release(struct msstx_iter * const iter, const u64 opaque)
{
  // all should use the same rcache
  sst_blk_release(iter->msst->rc, (const u8 *)opaque);
}

  void
msstx_iter_skip1(struct msstx_iter * const iter)
{
  if (!msstx_iter_valid(iter))
    return;
  struct sst_iter * const iter1 = iter->mh[1];
  sst_iter_skip1(iter1);
  if (sst_iter_valid(iter1))
    sst_iter_fix_kv(iter1);
  msstx_mh_downheap(iter, 1);
}

  void
msstx_iter_skip(struct msstx_iter * const iter, const u32 nr)
{
  for (u32 i = 0; i < nr; i++) {
    if (!msstx_iter_valid(iter))
      return;
    struct sst_iter * const iter1 = iter->mh[1];
    sst_iter_skip1(iter1);
    if (sst_iter_valid(iter1))
      sst_iter_fix_kv(iter1);
    msstx_mh_downheap(iter, 1);
  }
}

  struct kv *
msstx_iter_next(struct msstx_iter * const iter, struct kv * const out)
{
  struct kv * const ret = msstx_iter_peek(iter, out);
  msstx_iter_skip1(iter);
  return ret;
}

  void
msstx_iter_park(struct msstx_iter * const iter)
{
  const u32 nrun = iter->nrun;
  for (u32 i = 0; i < nrun; i++)
    sst_iter_park(&(iter->iters[i]));
}

  void
msstx_iter_destroy(struct msstx_iter * const iter)
{
  msstx_iter_park(iter);
  free(iter);
}
// }}} msstx

// ssty {{{
struct ssty_meta {
  u32 nrun; // 0 to MSST_NRUN
  u32 nkidx; // size of the ranks array
  u32 tagoff; // (optional) hash tag array if not zero
  u32 ptroff; // cursor offsets
  u32 inr1; // number of L1 index keys; number of blocks; <= 4KB blocks
  u32 ioff1;
  u32 inr2; // number of L2 index keys (much smaller)
  u32 ioff2;

  u32 totkv; // total number, including stale keys and tombstones
  u32 totsz; // sum of all sstx file's sizes (NOTE: totsz < 4GB)
  u32 valid; // number of valid keys (excluding stale keys and tombstones)
  u32 uniqx[MSST_NRUN+1];
  u64 magic;
};

struct ssty {
  union {
    u8 * mem; // and the array
    const u8 * ranks;
  };
  size_t size; // ssty file size
  u32 nrun;
  u32 nkidx; // number of entries (including placeholders)
  u32 inr1; // meta->inr1
  u32 inr2; // meta->inr2
  const struct sst_ptr * ptrs; // array of seek pointers
  u32 ioff1; // meta->ioff1
  u32 ioff2; // meta->ioff2
  const u8 * tags; // (optional) 16-bit non-zero hash tags
  const struct ssty_meta * meta;
  //const struct sst_blkmeta * bms[MSST_NRUN];
};

  static struct ssty *
ssty_open_at(const int dfd, const u64 seq, const u32 nrun)
{
  char fn[16];
  const u64 magic = seq * 100lu + nrun;
  sprintf(fn, "%03lu.ssty", magic);
  const int fd = openat(dfd, fn, O_RDONLY);
  if (fd < 0)
    return NULL;
  struct stat st;

  if (fstat(fd, &st) != 0) {
    close(fd);
    return NULL;
  }

  const u64 fsize = (u64)st.st_size;
  debug_assert(fsize < UINT32_MAX);
  u8 * const mem = mmap(NULL, fsize, PROT_READ, SSTY_MMAP_FLAGS, fd, 0);
  close(fd); // will only use the mapped memory
  if (mem == MAP_FAILED)
    return NULL;

  debug_assert(mem);
  struct ssty * const ssty = calloc(1, sizeof(*ssty));

  ssty->mem = mem;
  ssty->size = fsize;
  //pages_lock(mem, fsize);
  const struct ssty_meta * const meta = (typeof(meta))(mem + fsize - sizeof(*meta));
  ssty->nrun = meta->nrun;
  ssty->nkidx = meta->nkidx;
  ssty->ptrs = (struct sst_ptr *)(mem + meta->ptroff); // size0+size1
  ssty->inr1 = meta->inr1; // nsecs
  ssty->ioff1 = meta->ioff1; // ioffsoff
  ssty->inr2 = meta->inr2; // ipages
  ssty->ioff2 = meta->ioff2; // ioffsoff2
  ssty->tags = meta->tagoff ? (typeof(ssty->tags))(mem + meta->tagoff) : NULL;
  ssty->meta = meta;
  debug_assert(ssty->meta->magic == magic);
  return ssty;
}

  struct ssty *
ssty_open(const char * const dirname, const u64 seq, const u32 nrun)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return NULL;
  struct ssty * const ssty = ssty_open_at(dfd, seq, nrun);
  close(dfd);
  return ssty;
}

  void
ssty_destroy(struct ssty * const ssty)
{
  debug_assert(ssty);
  munmap((void *)ssty->mem, ssty->size);
  free(ssty);
}

  void
ssty_fprint(struct ssty * const ssty, FILE * const fout)
{
  const double bpk = (double)(ssty->size << 3) / (double)ssty->meta->totkv;
  fprintf(fout, "%s magic %lu nrun %u nkidx %u inr1 %u inr2 %u filesz %zu tags %c bits/key %.1lf\n",
      __func__, ssty->meta->magic, ssty->nrun, ssty->nkidx,
      ssty->inr1, ssty->inr2, ssty->size, ssty->tags ? 'y' : 'n', bpk);
}

  static inline u32
ssty_search_index(const struct ssty * const ssty, const struct kref * const key)
{
  // use two-level index
  // i2: the first 8 bytes are two u32 for [start, end) of i1 (plr)
  const u32 * const ioffs2 = (const u32 *)(ssty->mem + ssty->ioff2);
  const u32 sidx2 = k128_search_le(ssty->mem + sizeof(u32) + sizeof(u32), ioffs2, key, 0, ssty->inr2);
  const u32 * const plr = (typeof(plr))(ssty->mem + ioffs2[sidx2]);
  const u32 * const ioffs1 = (const u32 *)(ssty->mem + ssty->ioff1);
  const u32 sidx = k128_search_le(ssty->mem, ioffs1, key, plr[0], plr[1]);
  return sidx;
}

  static u32
ssty_ranks_match_mask(const u8 * const ranks, const u8 rank)
{
#if defined(__x86_64__)

#if SSTY_DBITS == 5
#if defined(__AVX2__)
  const m256 maskv = _mm256_set1_epi8(SSTY_RANK);
  const m256 rankv = _mm256_set1_epi8((char)rank);
  const m256 tmpv = _mm256_and_si256(_mm256_load_si256((const void *)ranks), maskv);
  return (u32)_mm256_movemask_epi8(_mm256_cmpeq_epi8(tmpv, rankv));
#else // No __AVX2__, use SSE 4.2
  const m128 maskv = _mm_set1_epi8(SSTY_RANK);
  const m128 rankv = _mm_set1_epi8((char)rank);
  const m128 tmpvlo = _mm_and_si128(_mm_load_si128((const void *)ranks), maskv);
  const m128 tmpvhi = _mm_and_si128(_mm_load_si128((const void *)(ranks + sizeof(m128))), maskv);
  const u32 masklo = m128_movemask_u8(_mm_cmpeq_epi8(tmpvlo, rankv));
  const u32 maskhi = m128_movemask_u8(_mm_cmpeq_epi8(tmpvhi, rankv));
  return (maskhi << sizeof(m128)) | masklo;
#endif // __AVX2__
#elif SSTY_DBITS == 4
  const m128 maskv = _mm_set1_epi8(SSTY_RANK);
  const m128 rankv = _mm_set1_epi8((char)rank);
  const m128 tmpv = _mm_and_si128(_mm_load_si128((const void *)ranks), maskv);
  return (u32)_mm_movemask_epi8(_mm_cmpeq_epi8(tmpv, rankv));
#endif // SSTY_DBITS

#elif defined(__aarch64__)
  const m128 maskv = vdupq_n_u8(SSTY_RANK);
  const m128 rankv = vdupq_n_u8(rank);
#if SSTY_DBITS == 5
  const m128 cmplo = vceqq_u8(vandq_u8(vld1q_u8(ranks), maskv), rankv); // cmpeq => 0xff or 0x00
  const m128 cmphi = vceqq_u8(vandq_u8(vld1q_u8(ranks + sizeof(m128)), maskv), rankv); // cmpeq => 0xff or 0x00
  const u32 masklo = m128_movemask_u8(cmplo);
  const u32 maskhi = m128_movemask_u8(cmphi);
  return (maskhi << sizeof(m128)) | masklo;
#elif SSTY_DBITS == 4
  const m128 cmp = vceqq_u8(vandq_u8(vld1q_u8(ranks), maskv), rankv); // cmpeq => 0xff or 0x00
  return m128_movemask_u8(cmp);
#endif // SSTY_DBITS

#endif // __x86_64__
}

  static u32
ssty_ranks_count(const u8 * const ranks, const u32 nr, const u8 rank)
{
  const u32 mask = ssty_ranks_match_mask(ranks, rank) & (((u32)(1lu << nr)) - 1u);
  return (u32)__builtin_popcount(mask);
}

  static u8
ssty_tag(const u32 hash32)
{
  return (u8)hash32;
}

// find the matching tags and filter out stale keys
  static u32
ssty_tags_match_mask(const u8 * const tags, const u8 * const ranks, const u8 tag)
{
#if defined(__x86_64__)

#if SSTY_DBITS == 5
#if defined(__AVX2__)
  // stale -> 0xFF
  const m256 maskv = _mm256_cmpgt_epi8(_mm256_setzero_si256(), _mm256_load_si256((const void *)ranks));
  // match -> 0xFF
  const m256 matchv = _mm256_cmpeq_epi8(_mm256_load_si256((const void *)tags), _mm256_set1_epi8((char)tag));
  // (not stale) && match -> 0xFF -> bit
  return (u32)_mm256_movemask_epi8(_mm256_andnot_si256(maskv, matchv));
#else // No __AVX2__, use SSE 4.2
  const m128 zerov = _mm_setzero_si128();
  const m128 maskvlo = _mm_cmpgt_epi8(zerov, _mm_load_si128((const void *)ranks));
  const m128 maskvhi = _mm_cmpgt_epi8(zerov, _mm_load_si128((const void *)(ranks + sizeof(m128))));
  const m128 tagv = _mm_set1_epi8((char)tag);
  const m128 matchvlo = _mm_cmpeq_epi8(_mm_load_si128((const void *)tags), tagv);
  const m128 matchvhi = _mm_cmpeq_epi8(_mm_load_si128((const void *)(tags + sizeof(m128))), tagv);
  const u32 masklo = m128_movemask_u8(_mm_andnot_si128(maskvlo, matchvlo));
  const u32 maskhi = m128_movemask_u8(_mm_andnot_si128(maskvhi, matchvhi));
  return (maskhi << sizeof(m128)) | masklo;
#endif // __AVX2__
#elif SSTY_DBITS == 4
  // stale -> 0xFF
  const m128 maskv = _mm_cmpgt_epi8(_mm_setzero_si128(), _mm_load_si128((const void *)ranks));
  const m128 matchv = _mm_cmpeq_epi8(_mm_load_si128((const void *)tags), _mm_set1_epi8((char)tag));
  return (u32)_mm_movemask_epi8(_mm_andnot_si128(maskv, matchv));
#endif // SSTY_DBITS

#elif defined(__aarch64__)
  const m128 stalev = vdupq_n_u8(SSTY_STALE);
  const m128 tagv = vdupq_n_u8(tag);
#if SSTY_DBITS == 5
  const m128 maskvlo = vcltq_u8(vld1q_u8(ranks), stalev);
  const m128 maskvhi = vcltq_u8(vld1q_u8(ranks + sizeof(m128)), stalev);
  const m128 matchvlo = vceqq_u8(vld1q_u8(tags), tagv);
  const m128 matchvhi = vceqq_u8(vld1q_u8(tags + sizeof(m128)), tagv);
  const u32 masklo = m128_movemask_u8(vandq_u8(maskvlo, matchvlo));
  const u32 maskhi = m128_movemask_u8(vandq_u8(maskvhi, matchvhi));
  return (maskhi << sizeof(m128)) | masklo;
#elif SSTY_DBITS == 4
  const m128 maskv = vcltq_u8(vld1q_u8(ranks), stalev);
  const m128 matchv = vceqq_u8(vld1q_u8(tags), tagv);
  return m128_movemask_u8(vandq_u8(maskv, matchv));
#endif // SSTY_DBITS

#endif // __x86_64__
}
// }}} ssty

// mssty {{{
struct mssty_iter {
  // ssty status
  u32 kidx; // invalid if >= ssty->nkidx
  u32 valid_bm;
  const struct sst_ptr * seek_ptrs;
  struct msst * msst;
  struct ssty * ssty;
  // iters
  struct sst_iter iters[MSST_NRUN];
};

// misc {{{
  static bool
mssty_open_y_at(const int dfd, struct msst * const msst)
{
  debug_assert(msst->ssty == NULL);
  struct ssty * const ssty = ssty_open_at(dfd, msst->seq, msst->nrun);
  msst->ssty = ssty;
  return ssty != NULL;
}

  bool
mssty_open_y(const char * const dirname, struct msst * const msst)
{
  debug_assert(msst->ssty == NULL);
  struct ssty * const ssty = ssty_open(dirname, msst->seq, msst->nrun);
  msst->ssty = ssty;
  return ssty != NULL;
}

// naming convention example: seq=123, nrun=8:
// dir/12300.sstx, dir/12301.sstx, ..., dir/12307.sstx, dir/12308.ssty
  static struct msst *
mssty_open_at(const int dfd, const u64 seq, const u32 nrun)
{
  struct msst * const msst = msstx_open_at(dfd, seq, nrun);
  if (msst == NULL)
    return NULL;

  if (!mssty_open_y_at(dfd, msst)) {
    msstx_destroy(msst);
    return NULL;
  }

  return msst;
}

  struct msst *
mssty_open(const char * const dirname, const u64 seq, const u32 nrun)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return NULL;

  struct msst * const msst = mssty_open_at(dfd, seq, nrun);
  close(dfd);
  return msst;
}

  static void
mssty_destroy_lazy(struct msst * const msst)
{
  ssty_destroy(msst->ssty);
  for (u32 i = 0; i < msst->nrun; i++)
    sst_deinit_lazy(&(msst->ssts[i]));
  free(msst);
}

  void
mssty_destroy(struct msst * const msst)
{
  ssty_destroy(msst->ssty);
  msst->ssty = NULL;
  msstx_destroy(msst);
}

  void
mssty_fprint(struct msst * const msst, FILE * const fout)
{
  const u32 nrun = msst->nrun;
  fprintf(fout, "%s nrun %u\n", __func__, nrun);
  ssty_fprint(msst->ssty, fout);
  for (u32 i = 0; i < nrun; i++)
    sst_fprint(&(msst->ssts[i]), fout);
}

  struct mssty_iter *
mssty_iter_create(struct mssty_ref * const ref)
{
  // ref is already an iter
  return (struct mssty_iter *)ref;
}
// }}} misc

// helpers {{{
  static void
mssty_iter_init(struct mssty_iter * const iter, struct msst * const msst)
{
  iter->msst = msst;
  iter->ssty = msst->ssty;
  iter->valid_bm = 0;
  const u32 nrun = iter->ssty->nrun;
  for (u32 i = 0; i < nrun; i++)
    sst_iter_init(&(iter->iters[i]), &(msst->ssts[i]), i);
}

  void
mssty_iter_park(struct mssty_iter * const iter)
{
  u32 bits = iter->valid_bm;
  while (bits) {
    const u32 i = (u32)__builtin_ctz(bits);
    sst_iter_park(&(iter->iters[i]));
    bits ^= (1u << i);
  }
}

// internal: invalidate the iter and set a new ptr
  static inline void
mssty_iter_set_ptr(struct sst_iter * const iter, struct sst_ptr ptr)
{
  sst_iter_park(iter);
  iter->ptr = ptr;
}

  static void
mssty_iter_fix_rank(struct mssty_iter * const iter, const u8 rank)
{
  debug_assert(rank < MSST_NRUN);
  if (((1u << rank) & iter->valid_bm) == 0) {
    mssty_iter_set_ptr(&(iter->iters[rank]), iter->seek_ptrs[rank]);
    iter->valid_bm |= (1u << rank);
  }
}

  void
mssty_iter_seek_null(struct mssty_iter * const iter)
{
  mssty_iter_park(iter);
  iter->valid_bm = 0;
  iter->kidx = 0;
  struct ssty * const ssty = iter->ssty;
  iter->seek_ptrs = ssty->ptrs;
  if (ssty->nkidx)
    mssty_iter_fix_rank(iter, ssty->ranks[0] & SSTY_RANK);
}

  struct mssty_ref *
mssty_ref(struct msst * const msst)
{
  struct mssty_iter * const iter = malloc(sizeof(*iter));
  if (iter == NULL)
    return NULL;

  mssty_iter_init(iter, msst);
  iter->kidx = iter->ssty->nkidx; // invalid
  return (struct mssty_ref *)iter;
}

  struct msst *
mssty_unref(struct mssty_ref * const ref)
{
  struct mssty_iter * const iter = (typeof(iter))ref;
  struct msst * const msst = iter->msst;
  mssty_iter_park(iter);
  free(iter);
  return msst;
}

// to randomly access a sst_iter in a segment
  static struct sst_iter *
mssty_iter_access(struct mssty_iter * const iter, const u8 * const ranks, const u32 i)
{
  const u8 rank = ranks[i] & SSTY_RANK;
  struct sst_iter * const iter1 = &(iter->iters[rank]);
  mssty_iter_set_ptr(iter1, iter->seek_ptrs[rank]);
  const u32 nskip = ssty_ranks_count(ranks, i, rank);
  sst_iter_skip(iter1, nskip);
  debug_assert(sst_iter_valid(iter1));
  sst_iter_fix_kv(iter1);
  return iter1;
}

  inline bool
mssty_iter_valid(struct mssty_iter * const iter)
{
  return iter->kidx < iter->ssty->nkidx;
}

// internal: skip the sst_iter at rank (< nrun)
  static inline void
mssty_iter_skip1_rank(struct mssty_iter * const iter, const u8 rank)
{
  debug_assert(rank < MSST_NRUN);
  sst_iter_skip1(&(iter->iters[rank]));
}

  static inline void
mssty_iter_skip_rank(struct mssty_iter * const iter, const u8 rank, const u32 nr)
{
  debug_assert(rank < MSST_NRUN);
  sst_iter_skip(&(iter->iters[rank]), nr);
}

// does not check iter_valid
  void
mssty_iter_skip1(struct mssty_iter * const iter)
{
  debug_assert(mssty_iter_valid(iter));
  struct ssty * const ssty = iter->ssty;
  const u8 * const ranks = ssty->ranks;
  u8 rank = ranks[iter->kidx] & SSTY_RANK;

  mssty_iter_skip1_rank(iter, rank);
  iter->kidx++;
  while (ranks[iter->kidx] & SSTY_STALE) { // stop when kidx >= nkidx
    rank = ranks[iter->kidx] & SSTY_RANK;
    iter->kidx++;
    if (rank < ssty->nrun) { // not gap
      mssty_iter_fix_rank(iter, rank);
      mssty_iter_skip1_rank(iter, rank);
    }
  }

  if (mssty_iter_valid(iter)) {
    rank = ranks[iter->kidx] & SSTY_RANK;
    debug_assert(rank < ssty->nrun);
    mssty_iter_fix_rank(iter, rank);
  }
}
// }}} helpers

// seek {{{
  static u32
mssty_iter_seek_bisect(struct mssty_iter * const iter, const struct kref * const key, const u32 aidx, u32 l, u32 r)
{
  // notes from ssty_build:
  // gaps are marked SSTY_INVALID; end of array is marked nrun
  debug_assert(r <= SSTY_DIST);
  const u32 r0 = r;
  struct ssty * const ssty = iter->ssty;
  const u8 * const ranks = ssty->ranks + aidx;

  // l may point to a stale key when performing forward searching
  // search-key > key-at-l; it's safe to move forward
  while (ranks[l] & SSTY_STALE)
    l++;

  // skip stale slots and placeholders
  // no underflow because the first key of each group is not stale
  while (r && (ranks[r-1] & SSTY_STALE))
    r--;

  while (l < r) {
#ifdef MSSTY_SEEK_BISECT_OPT
    const u8 rankx = ranks[(l+r)>>1] & SSTY_RANK; // pick up a rank randomly
    struct sst_iter * const iterx = &(iter->iters[rankx]);
    mssty_iter_set_ptr(iterx, iter->seek_ptrs[rankx]);
    debug_assert(sst_iter_valid(iterx));
    // scan from l to r; skip 0 to l
    // use 1lu << r to avoid undefined behavior of left-shift by 32
    const u32 mask0 = ssty_ranks_match_mask(ranks, rankx) & (((u32)(1lu << r)) - 1u);
    debug_assert(l < SSTY_DIST);
    const u32 low = (1u << l) - 1u;
    const u32 nskip0 = (u32)__builtin_popcount(mask0 & low);
    if (nskip0)
      sst_iter_skip(iterx, nskip0);
    u32 mask = mask0 & (~low); // bits between l and r
    debug_assert(mask); // have at least one bit
    do { // scan one by one
      sst_iter_fix_kv(iterx);
      const int cmp = sst_iter_compare_kref(iterx, key);
      const u32 m = (u32)__builtin_ctz(mask);
      debug_assert((ranks[m] & SSTY_RANK) == rankx);
      debug_assert(m < r);
      if (cmp < 0) { // shrink forward
        sst_iter_skip1(iterx);
        l = m + 1;
      } else if (cmp > 0) { // shrink backward
        r = m;
        break;
      } else { // match; must point to the non-stale version
        l = m;
        while (ranks[l] & SSTY_STALE)
          l--;
        r = m;
        break;
      }
      mask &= (mask - 1);
    } while (mask);
    sst_iter_park(iterx);
    // l may point to a stale key when performing forward searching
    // search-key > key-at-l; it's safe to move forward
    while ((l < r0) && (ranks[l] & SSTY_STALE))
      l++;
    // skip stale slots and placeholders
    // no underflow because the first key of each group is not stale
    while (r && (ranks[r-1] & SSTY_STALE))
      r--;
#else // MSSTY_SEEK_BISECT_OPT
    u32 m = (l + r) >> 1;
    // skip stale keys and move left; always compare with non-stale key
    while (ranks[m] & SSTY_STALE)
      m--;

    // compare
    debug_assert(l <= m && m < r);
    debug_assert((ranks[m] & SSTY_STALE) == 0);
    struct sst_iter * const iterm = mssty_iter_access(iter, ranks, m);
    const int cmp = sst_iter_compare_kref(iterm, key);
    sst_iter_park(iterm);

    if (cmp < 0) { // shrink forward
      l = m + 1;
      // skip stale keys
      while ((l < r0) && (ranks[l] & SSTY_STALE))
        l++;
    } else if (cmp > 0) { // shrink backward
      r = m;
    } else { // cmp == 0; done
      l = m;
      r = m;
    }
#endif // MSSTY_SEEK_BISECT_OPT
  }
  return l;
}

// perform seek in the group of kidx0 (group_id = kidx0 / dist)
// kidx0 may point to a stale key
// bisect between kidx0 and the last element
  static void
mssty_iter_seek_local(struct mssty_iter * const iter, const struct kref * const key, const u32 kidx0)
{
  debug_assert(iter->valid_bm == 0);
  // first key's index of the target group
  struct ssty * const ssty = iter->ssty;
  const u32 aidx = kidx0 >> SSTY_DBITS << SSTY_DBITS;
  // <= dist
  const u32 l0 = kidx0 - aidx;
  const u32 r0 = (aidx + SSTY_DIST) > ssty->nkidx ? (ssty->nkidx - aidx) : SSTY_DIST;
  const u32 goff = mssty_iter_seek_bisect(iter, key, aidx, l0, r0);

  debug_assert(iter->valid_bm == 0);
  // skip keys
  if (goff) {
    if (goff < SSTY_DIST) {
      const u8 * const ranks = ssty->ranks + aidx;
      for (u8 i = 0; i < ssty->nrun; i++) {
        const u32 nskip = ssty_ranks_count(ranks, goff, i);
        if (nskip) {
          mssty_iter_fix_rank(iter, i);
          mssty_iter_skip_rank(iter, i, nskip);
        }
      }
    } else { // shortcut to the next group
      debug_assert(goff == SSTY_DIST);
      iter->seek_ptrs += ssty->nrun;
    }
  } // else: goff == 0; do nothing
  iter->kidx = aidx + goff;
  // the current key must be the one unless >= nkidx
  if (unlikely(iter->kidx >= ssty->nkidx))
    return;

  const u8 rank = ssty->ranks[iter->kidx] & SSTY_RANK; // rank starts with 0
  debug_assert(rank < ssty->nrun);
  mssty_iter_fix_rank(iter, rank);
}

  void
mssty_iter_seek(struct mssty_iter * const iter, const struct kref * const key)
{
  struct ssty * const ssty = iter->ssty;
  mssty_iter_park(iter);
  iter->valid_bm = 0;
  const u32 sidx = ssty_search_index(ssty, key);
  if (unlikely(sidx >= ssty->inr1)) { // invalid
    iter->kidx = ssty->nkidx;
    return;
  }
  iter->seek_ptrs = &(ssty->ptrs[sidx * ssty->nrun]);
  const u32 kidx0 = sidx << SSTY_DBITS;
  mssty_iter_seek_local(iter, key, kidx0);
}
// }}} seek

// seek-near {{{
  static u32
mssty_iter_seek_index_near(struct ssty * const ssty, const struct kref * const key, u32 l)
{
  while ((l < ssty->nkidx) && (ssty->ranks[l] & SSTY_STALE))
    l++;
  if (l == ssty->nkidx)
    return l;

  // linear scan
  const u32 * const ioffs = (const u32 *)(ssty->mem + ssty->ioff1);
  u32 g = l >> SSTY_DBITS;
  u32 r = (g + 1) << SSTY_DBITS;
  while (r < ssty->nkidx) {
    const int cmp = kref_k128_compare(key, ssty->mem + ioffs[g + 1]);
    if (cmp < 0) {
      break;
    } else {
      g++;
      l = r;
      r += SSTY_DIST;
    }
  }
  debug_assert(l < ssty->nkidx);
  return l;
}

  static void
mssty_iter_seek_local_near(struct mssty_iter * const iter, const struct kref * const key, const u32 l)
{
  struct ssty * const ssty = iter->ssty;
  // now search in l's group; l must be a non-stale key
  const u32 g = l >> SSTY_DBITS;
  if (g == (iter->kidx >> SSTY_DBITS)) { // stay in the same group; reuse the valid iter
    while (iter->kidx < l)
      mssty_iter_skip1(iter);
    debug_assert(iter->kidx == l);
  } else { // switch group
    debug_assert((l & (SSTY_DIST-1)) == 0);
    mssty_iter_park(iter);
    iter->valid_bm = 0;
    iter->seek_ptrs = &(ssty->ptrs[(l >> SSTY_DBITS) * ssty->nrun]);
    iter->kidx = l;
  }

  do {
    const u8 rank = ssty->ranks[iter->kidx] & SSTY_RANK; // rank starts with 0
    debug_assert(rank < ssty->nrun);
    mssty_iter_fix_rank(iter, rank);
    struct sst_iter * const iter1 = &(iter->iters[rank]);
    sst_iter_fix_kv(iter1);
    if (sst_iter_compare_kref(iter1, key) >= 0)
      return;
    mssty_iter_skip1(iter);
  } while (mssty_iter_valid(iter));
}

// the target key must >= the current key under the iter
// the current iter can point to a stale key or deleted key
  void
mssty_iter_seek_near(struct mssty_iter * const iter, const struct kref * const key, const bool bsearch_keys)
{
  debug_assert(mssty_iter_valid(iter));
  struct ssty * const ssty = iter->ssty;

  // first test if key < iter
  const u8 * const ranks = ssty->ranks;
  const u8 rank0 = ranks[iter->kidx];
  struct sst_iter * const iter0 = &(iter->iters[rank0 & SSTY_RANK]);
  sst_iter_fix_kv(iter0);
  // return without any change
  if (sst_iter_compare_kref(iter0, key) >= 0)
    return;

  // seek_index does not affect the iter
  const u32 l = mssty_iter_seek_index_near(ssty, key, iter->kidx+1);
  if (l == ssty->nkidx) { // invalid
    mssty_iter_park(iter);
    iter->kidx = ssty->nkidx;
    return;
  }

  if (bsearch_keys) { // reset iter and use seek_local
    mssty_iter_park(iter);
    iter->valid_bm = 0;
    iter->seek_ptrs = &(ssty->ptrs[(l >> SSTY_DBITS) * ssty->nrun]);
    mssty_iter_seek_local(iter, key, l);
  } else { // linear scan using the valid iter
    mssty_iter_seek_local_near(iter, key, l);
  }
}
// }}} seek-near

// iter {{{
// peek non-stale keys
  struct kv *
mssty_iter_peek(struct mssty_iter * const iter, struct kv * const out)
{
  if (!mssty_iter_valid(iter))
    return NULL;
  const u8 rank = iter->ssty->ranks[iter->kidx]; // rank starts with 0
  debug_assert((rank & SSTY_STALE) == 0);
  return sst_iter_peek(&(iter->iters[rank & SSTY_RANK]), out);
}

// kvref non-stale keys
  bool
mssty_iter_kref(struct mssty_iter * const iter, struct kref * const kref)
{
  if (!mssty_iter_valid(iter))
    return false;

  const u8 rank = iter->ssty->ranks[iter->kidx]; // rank starts with 0
  debug_assert((rank & SSTY_STALE) == 0);
  return sst_iter_kref(&(iter->iters[rank & SSTY_RANK]), kref);
}

// kvref non-stale keys
  bool
mssty_iter_kvref(struct mssty_iter * const iter, struct kvref * const kvref)
{
  if (!mssty_iter_valid(iter))
    return false;

  const u8 rank = iter->ssty->ranks[iter->kidx]; // rank starts with 0
  debug_assert((rank & SSTY_STALE) == 0);
  return sst_iter_kvref(&(iter->iters[rank & SSTY_RANK]), kvref);
}

  inline u64
mssty_iter_retain(struct mssty_iter * const iter)
{
  const u8 rank = iter->ssty->ranks[iter->kidx] & SSTY_RANK; // rank starts with 0
  return sst_iter_retain(&(iter->iters[rank]));
}

  inline void
mssty_iter_release(struct mssty_iter * const iter, const u64 opaque)
{
  sst_blk_release(iter->msst->rc, (const u8 *)opaque);
}

// non-stale keys
  void
mssty_iter_skip(struct mssty_iter * const iter, const u32 nr)
{
  if (!mssty_iter_valid(iter))
    return;
  struct ssty * const ssty = iter->ssty;
  const u8 * const ranks = ssty->ranks;
  u8 rank = ranks[iter->kidx] & SSTY_RANK;
  debug_assert(rank < ssty->nrun);

  for (u32 i = 0; i < nr; i++) {
    mssty_iter_skip1_rank(iter, rank);
    iter->kidx++;

    // skip stale keys or gaps
    while (ranks[iter->kidx] & SSTY_STALE) { // stop when kidx >= nkidx
      rank = ranks[iter->kidx] & SSTY_RANK;
      iter->kidx++;
      if (rank < ssty->nrun) { // not gap
        mssty_iter_fix_rank(iter, rank);
        mssty_iter_skip1_rank(iter, rank);
      }
    }
    if (!mssty_iter_valid(iter))
      return;
    // still valid
    rank = ranks[iter->kidx] & SSTY_RANK;
    debug_assert(rank < ssty->nrun);
    mssty_iter_fix_rank(iter, rank);
  }
}

  struct kv *
mssty_iter_next(struct mssty_iter * const iter, struct kv * const out)
{
  struct kv * const ret = mssty_iter_peek(iter, out);
  if (mssty_iter_valid(iter))
    mssty_iter_skip1(iter);
  return ret;
}

  void
mssty_iter_destroy(struct mssty_iter * const iter)
{
  mssty_iter_park(iter);
}
// }}} iter

// ts {{{
// ts iter: ignore a key if its newest version is a tombstone
  bool
mssty_iter_ts(struct mssty_iter * const iter)
{
  return (iter->ssty->ranks[iter->kidx] & SSTY_TOMBSTONE) != 0;
}

// hide tomestones
  void
mssty_iter_seek_ts(struct mssty_iter * const iter, const struct kref * const key)
{
  mssty_iter_seek(iter, key);
  while (mssty_iter_valid(iter) && mssty_iter_ts(iter))
    mssty_iter_skip1(iter);
}

  void
mssty_iter_skip1_ts(struct mssty_iter * const iter)
{
  if (!mssty_iter_valid(iter))
    return;
  mssty_iter_skip1(iter);
  while (mssty_iter_valid(iter) && mssty_iter_ts(iter))
    mssty_iter_skip1(iter);
}

// skip nr valid keys (tomestones are transparent)
  void
mssty_iter_skip_ts(struct mssty_iter * const iter, const u32 nr)
{
  for (u32 i = 0; i < nr; i++) {
    if (!mssty_iter_valid(iter))
      return;
    mssty_iter_skip1(iter);
    while (mssty_iter_valid(iter) && mssty_iter_ts(iter))
      mssty_iter_skip1(iter);
  }
}

  struct kv *
mssty_iter_next_ts(struct mssty_iter * const iter, struct kv * const out)
{
  struct kv * const ret = mssty_iter_peek(iter, out);
  mssty_iter_skip1_ts(iter);
  return ret;
}
// }}} ts

// dup {{{
// _dup iterator: return all versions, including old keys and tombstones
  struct kv *
mssty_iter_peek_dup(struct mssty_iter * const iter, struct kv * const out)
{
  if (!mssty_iter_valid(iter))
    return NULL;
  const u8 rank = iter->ssty->ranks[iter->kidx] & SSTY_RANK; // rank starts with 0
  debug_assert(rank < iter->ssty->nrun);
  return sst_iter_peek(&(iter->iters[rank]), out);
}

  void
mssty_iter_skip1_dup(struct mssty_iter * const iter)
{
  if (!mssty_iter_valid(iter))
    return;
  struct ssty * const ssty = iter->ssty;
  const u8 * const ranks = ssty->ranks;
  u8 rank = ranks[iter->kidx] & SSTY_RANK;
  debug_assert(rank < ssty->nrun);

  mssty_iter_skip1_rank(iter, rank);
  iter->kidx++;

  // skip gaps
  while (ranks[iter->kidx] == SSTY_INVALID)
    iter->kidx++;

  if (!mssty_iter_valid(iter))
    return;

  // still valid
  rank = ranks[iter->kidx] & SSTY_RANK;
  mssty_iter_fix_rank(iter, rank);
}

  void
mssty_iter_skip_dup(struct mssty_iter * const iter, const u32 nr)
{
  if (!mssty_iter_valid(iter))
    return;
  struct ssty * const ssty = iter->ssty;
  const u8 * const ranks = ssty->ranks;
  u8 rank = ranks[iter->kidx] & SSTY_RANK;
  debug_assert(rank < ssty->nrun);

  for (u32 i = 0; i < nr; i++) {
    mssty_iter_skip1_rank(iter, rank);
    iter->kidx++;

    // skip gaps
    while (ranks[iter->kidx] == SSTY_INVALID)
      iter->kidx++;

    if (!mssty_iter_valid(iter))
      return;

    // still valid
    rank = ranks[iter->kidx] & SSTY_RANK;
    mssty_iter_fix_rank(iter, rank);
  }
}

  struct kv *
mssty_iter_next_dup(struct mssty_iter * const iter, struct kv * const out)
{
  struct kv * const ret = mssty_iter_peek_dup(iter, out);
  mssty_iter_skip1_dup(iter);
  return ret;
}

  bool
mssty_iter_kref_dup(struct mssty_iter * const iter, struct kref * const kref)
{
  if (!mssty_iter_valid(iter))
    return false;

  const u8 rank = iter->ssty->ranks[iter->kidx] & SSTY_RANK; // rank starts with 0
  return sst_iter_kref(&(iter->iters[rank]), kref);
}

  bool
mssty_iter_kvref_dup(struct mssty_iter * const iter, struct kvref * const kvref)
{
  if (!mssty_iter_valid(iter))
    return false;

  const u8 rank = iter->ssty->ranks[iter->kidx] & SSTY_RANK; // rank starts with 0
  return sst_iter_kvref(&(iter->iters[rank]), kvref);
}
// }}} dup

// point {{{
// return the internal sst_iter if there is a match
// when hide_ts == true, return NULL when the matching kv is a ts
// when a non-NULL sst_iter is returned, the caller must park it after use
  static struct sst_iter *
mssty_iter_match(struct mssty_iter * const iter, const struct kref * const key, const bool hide_ts)
{
  struct ssty * const ssty = iter->ssty;
  mssty_iter_park(iter);
  iter->valid_bm = 0;
  const u32 sidx = ssty_search_index(ssty, key);
  if (unlikely(sidx >= ssty->inr1)) { // invalid
    iter->kidx = ssty->nkidx;
    return NULL;
  }

  iter->seek_ptrs = &(ssty->ptrs[sidx * ssty->nrun]);

  // local
  const u32 aidx = sidx << SSTY_DBITS;
  const u8 * const ranks = ssty->ranks + aidx;
  const u32 rmax0 = ssty->nkidx - aidx;
  const u32 rmax = rmax0 < SSTY_DIST ? rmax0 : SSTY_DIST;

  if (ssty->tags) {
    const u8 * const tags = ssty->tags + aidx;
    u32 mask = ssty_tags_match_mask(tags, ranks, ssty_tag(key->hash32));
    while (mask) {
      const u32 i = (u32)__builtin_ctz(mask);
      debug_assert((ranks[i] & SSTY_STALE) == 0);

      // overflow
      if (i >= rmax)
        return NULL;

      if ((!hide_ts) || ((ranks[i] & SSTY_TOMBSTONE) == 0)) {
        struct sst_iter * const iter1 = mssty_iter_access(iter, ranks, i);
        if (sst_iter_match_kref(iter1, key)) // the caller must park the iterx later
          return iter1;

        sst_iter_park(iter1);
      }
      mask &= (mask - 1);
    }
  } else {
    const u32 r0 = (aidx + SSTY_DIST) > ssty->nkidx ? (ssty->nkidx - aidx) : SSTY_DIST;
    const u32 i = mssty_iter_seek_bisect(iter, key, aidx, 0, r0);

    // no match
    if (i >= rmax)
      return NULL;

    debug_assert((ranks[i] & SSTY_STALE) == 0);
    if ((!hide_ts) || ((ranks[i] & SSTY_TOMBSTONE) == 0)) {
      struct sst_iter * const iter1 = mssty_iter_access(iter, ranks, i);
      if (sst_iter_match_kref(iter1, key)) // the caller must park the iterx later
        return iter1;

      sst_iter_park(iter1);
    }
  }
  return NULL;
}

  static struct kv *
mssty_get_internal(struct mssty_ref * const ref, const struct kref * const key, struct kv * const out, const bool hide_ts)
{
  struct mssty_iter * const iter = (typeof(iter))ref;
  struct sst_iter * const iter1 = mssty_iter_match(iter, key, hide_ts);
  if (iter1) {
    struct kv * const ret = sst_iter_peek(iter1, out);
    sst_iter_park(iter1);
    return ret;
  } else {
    return NULL;
  }
}

  static bool
mssty_probe_internal(struct mssty_ref * const ref, const struct kref * const key, const bool hide_ts)
{
  struct mssty_iter * const iter = (typeof(iter))ref;
  struct sst_iter * const iter1 = mssty_iter_match(iter, key, hide_ts);
  if (iter1) {
    sst_iter_park(iter1);
    return true;
  } else {
    return false;
  }
}

// mssty_get can return tombstone
  struct kv *
mssty_get(struct mssty_ref * const ref, const struct kref * const key, struct kv * const out)
{
  return mssty_get_internal(ref, key, out, false);
}

// mssty_probe can return tombstone
  bool
mssty_probe(struct mssty_ref * const ref, const struct kref * const key)
{
  return mssty_probe_internal(ref, key, false);
}

// return NULL for tomestone
  struct kv *
mssty_get_ts(struct mssty_ref * const ref, const struct kref * const key, struct kv * const out)
{
  return mssty_get_internal(ref, key, out, true);
}

// return false for tomestone
  bool
mssty_probe_ts(struct mssty_ref * const ref, const struct kref * const key)
{
  return mssty_probe_internal(ref, key, true);
}

// return false for tomestone
  bool
mssty_get_value_ts(struct mssty_ref * const ref, const struct kref * const key,
    void * const vbuf_out, u32 * const vlen_out)
{
  struct mssty_iter * const iter = (typeof(iter))ref;
  struct sst_iter * const iter1 = mssty_iter_match(iter, key, true);
  if (iter1) {
    memcpy(vbuf_out, iter1->kvdata + iter1->klen, iter1->vlen);
    *vlen_out = iter1->vlen;
    sst_iter_park(iter1);
    return true;
  } else {
    return false;
  }
}

  struct kv *
mssty_first(struct msst * const msst, struct kv * const out)
{
  if (msst->ssty->nkidx == 0)
    return NULL;
  struct sst_iter iter1;
  iter1.rank = msst->ssty->ranks[0] & SSTY_RANK;
  iter1.sst = &(msst->ssts[iter1.rank]);
  iter1.ptr.pageid = 0;
  iter1.ptr.keyid = 0;
  iter1.kvdata = NULL; // init to NULL
  struct kv * const ret = sst_iter_peek(&iter1, out);
  sst_iter_park(&iter1);
  return ret;
}

  struct kv *
mssty_last(struct msst * const msst, struct kv * const out)
{
  const u32 nkidx = msst->ssty->nkidx;
  if (nkidx == 0)
    return NULL;
  struct sst_iter iter1;
  const u8 rank = msst->ssty->ranks[nkidx-1] & SSTY_RANK;
  struct sst * const sst = &(msst->ssts[rank]);
  iter1.sst = sst;
  debug_assert(sst->npages); // the sst at this rank must be non-empty
  u32 pageid = sst->npages-1;
  // seek to the head if it's a jumbo block
  while (sst->bms[pageid].npages == 0)
    pageid--;
  debug_assert((pageid + sst->bms[pageid].npages) == sst->npages);
  iter1.ptr.pageid = (u16)pageid;
  iter1.ptr.keyid = sst->bms[pageid].nkeys-1;
  iter1.kvdata = NULL; // init to NULL
  struct kv * const ret = sst_iter_peek(&iter1, out);
  sst_iter_park(&iter1);
  return ret;
}
// }}} point

// dump {{{
  void
mssty_dump(struct msst * const msst, const char * const fn)
{
  const int fd = open(fn, O_WRONLY|O_CREAT|O_TRUNC, 00644);
  debug_assert(fd >= 0);
  void * const ref = mssty_ref(msst);
  struct mssty_iter * const iter = mssty_iter_create(ref);
  struct ssty * const ssty = msst->ssty;
  const struct ssty_meta * const meta = ssty->meta;
  dprintf(fd, "mssty seq%lu nrun %u nkidx %u inr1 %u inr2 %u valid %u uniqx ",
      msst->seq, ssty->nrun, ssty->nkidx, ssty->inr1, ssty->inr2, meta->valid);
  for (u32 i = 0; i < ssty->nrun; i++)
    dprintf(fd, "%u%c", meta->uniqx[i], (i == (ssty->nrun-1)) ? '\n' : ' ');

  // print i2 keys
  const u32 n2 = ssty->inr2;
  const u32 * const ioffs2 = (const u32 *)(ssty->mem + ssty->ioff2);
  for (u32 i = 0; i < n2; i++) {
    const u32 ioff2 = ioffs2[i];
    const u32 * const e2 = (typeof(e2))(ssty->mem + ioff2);
    const u8 * const a2 = (typeof(a2))(ssty->mem + ioff2 + sizeof(u32) + sizeof(u32));
    u32 klen = 0;
    const u8 * const keyptr = vi128_decode_u32(a2, &klen);
    dprintf(fd, "i2 %6u  %6u %6u %.*s (%u)\n", i, e2[0], e2[1], klen, keyptr, klen);
  }

  struct kvref kvref;
  u32 n = 0;
  mssty_iter_seek_null(iter);
  while (mssty_iter_kvref_dup(iter, &kvref)) { // dump all the keys
    const u8 rank = ssty->ranks[iter->kidx];
    const bool stale = (rank & SSTY_STALE) != 0;
    const bool ts = (rank & SSTY_TOMBSTONE) != 0; // first X: ssty says it's a TS
    const bool ts2 = (kvref.hdr.vlen & SST_VLEN_TS) != 0; // second X: the KV is really a TS
    debug_assert(ts == ts2);
    // count kidx(anchor) !DD rank key
    dprintf(fd, "%7u %7u%c %c%c%c%x %.*s (%u,%u)\n", n, iter->kidx, (iter->kidx % SSTY_DIST) ? ' ' : '*',
        stale ? '!' : ' ', ts ? 'X' : ' ', ts2 ? 'X' : ' ', rank & SSTY_RANK,
        kvref.hdr.klen, kvref.kptr, kvref.hdr.klen, kvref.hdr.vlen & SST_VLEN_MASK);
    mssty_iter_skip1_dup(iter);
    n++;
  }
  mssty_iter_destroy(iter);
  mssty_unref(ref);
  fsync(fd);
  close(fd);
}
// }}} dump

// }}} mssty

// ssty_build {{{

// bi {{{
struct ssty_build_info {
  struct msst * x1; // input: target tables
  struct msst * y0; // input: the old mssty or NULL

  // allocated by the main function; filled by the sort function
  u8 * ranks; // output: run selectors
  struct sst_ptr * ptrs; // output: cursor positions
  struct kv ** anchors; // output: anchors
  u8 * tags; // output: hash tags

  int dfd; // input
  u32 run0;  // input: number of ssts to reuse in y0
  u32 nkidx; // output: maximum key index
  u32 nsecs; // output: number of groups
  u32 valid; // output: number of valid keys
  u32 uniqx[MSST_NRUN]; // output: uniq non-stale keys at each level
};

// number of bytes read
static __thread u64 ssty_build_ckeys_reads = 0;
// }}} bi

// sstc_iter {{{
struct sstc_iter {
  struct sst_iter iter;
  u8 * rawbuffer;
  u8 * buffer; // rawbuffer + 8
  size_t bufsz;
  const u8 * cptr;
  const u8 * ckeysptr;
};

  static void
sstc_sync_kv(struct sstc_iter * const iter)
{
  const u8 * ptr = iter->cptr;
  u32 plen = 0, slen = 0;
  ptr = vi128_decode_u32(ptr, &plen);
  ptr = vi128_decode_u32(ptr, &slen);
  const bool ts = *ptr++;
  if ((plen + slen + sizeof(u64)) > iter->bufsz) {
    iter->bufsz = bits_p2_up_u32(plen + slen + 256);
    iter->rawbuffer = realloc(iter->rawbuffer, iter->bufsz);
    iter->buffer = iter->rawbuffer + sizeof(u64);
    debug_assert(iter->rawbuffer);
  }
  memcpy(iter->buffer + plen, ptr, slen);

  iter->cptr = ptr + slen;

  struct sst_iter * const iter0 = &iter->iter;
  iter0->klen = plen + slen;
  iter0->vlen = ts ? SST_VLEN_TS : 0;
  iter0->kvdata = iter->buffer;
}

  static inline void
sstc_iter_init(struct sstc_iter * const iter, struct sst * const sst, const u32 rank)
{
  sst_iter_init(&iter->iter, sst, rank);
  const struct sst_meta * const meta = sst_meta(sst);

  if (sst->npages && meta->ckeyssz) {
    iter->bufsz = 256; // buffer size
    iter->rawbuffer = malloc(iter->bufsz);
    iter->buffer = iter->rawbuffer + sizeof(u64);

    const u8 * const ckeys = sst->mem + meta->ckeysoff;
    iter->ckeysptr = ckeys;
    posix_madvise((void *)ckeys, meta->ckeyssz, POSIX_MADV_WILLNEED);
    iter->cptr = ckeys;

    struct sst_iter * const iter0 = &iter->iter;
    iter0->ptr.pageid = 0;
    iter0->ptr.keyid = 0;

    sstc_sync_kv(iter);
  }
}

  static struct sstc_iter *
sstc_iter_create(struct sst * const sst)
{
  struct sstc_iter * const iter = calloc(1, sizeof(*iter));
  if (iter == NULL)
    return NULL;
  sstc_iter_init(iter, sst, 0);
  return iter;
}

  static inline bool
sstc_iter_valid(struct sstc_iter * const iter)
{
  return sst_iter_valid(&iter->iter);
}

  static inline void
sstc_iter_seek(struct sstc_iter * const iter, const struct kref * const key)
{
  debug_assert(key == NULL || key == kref_null());
  (void)iter;
  (void)key;
}

  static inline void
sstc_iter_skip1(struct sstc_iter * const iter)
{
  struct sst_iter * const iter0 = &iter->iter;
  iter0->kvdata = NULL; // it points to the sstc buffer; just set to NULL
  sst_iter_skip1(iter0);
  if (sst_iter_valid(iter0))
    sstc_sync_kv(iter);
}

  static void
sstc_iter_skip(struct sstc_iter * const iter, const u32 nr)
{
  for (u32 i = 0; i < nr; i++)
    sstc_iter_skip1(iter);
}

  static bool
sstc_iter_kref(struct sstc_iter * const iter, struct kref * const kref)
{
  if (!sstc_iter_valid(iter))
    return false;

  kref_ref_raw(kref, iter->buffer, iter->iter.klen); // no hash32
  return true;
}

  static bool
sstc_iter_kvref(struct sstc_iter * const iter, struct kvref * const kvref)
{
  if (!sstc_iter_valid(iter))
    return false;

  struct sst_iter * const iter0 = &iter->iter;
  kvref->hdr.klen = iter0->klen;
  kvref->hdr.vlen = iter0->vlen;
  kvref->hdr.hash = 0;
  kvref->kptr = iter->buffer;
  kvref->vptr = NULL; // no value
  return true;
}

  static bool
sstc_iter_ts(struct sstc_iter * const iter)
{
  return sstc_iter_valid(iter) && (iter->iter.vlen == SST_VLEN_TS);
}

  static int
sstc_iter_compare(struct sstc_iter * const iter1, struct sstc_iter * const iter2)
{
  debug_assert(sstc_iter_valid(iter1) && sstc_iter_valid(iter2));
  // both are valid
  struct kref kref1, kref2;
  kref_ref_raw(&kref1, iter1->buffer, iter1->iter.klen); // no hash32
  kref_ref_raw(&kref2, iter2->buffer, iter2->iter.klen); // no hash32
  return kref_compare(&kref1, &kref2);
}

  static void
sstc_iter_destroy(struct sstc_iter * const iter)
{
  if (iter->ckeysptr) {
    const u32 ckeyssz = sst_meta(iter->iter.sst)->ckeyssz;
    debug_assert(ckeyssz);
    posix_madvise((void *)iter->ckeysptr, ckeyssz, POSIX_MADV_DONTNEED);
  }

  free(iter->rawbuffer);
  free(iter);
}

static const struct kvmap_api kvmap_api_sstc = {
  .ordered = true,
  .readonly = true,
  .unique = true,
  .iter_create = (void *)sstc_iter_create,
  .iter_seek = (void *)sstc_iter_seek,
  .iter_valid = (void *)sstc_iter_valid,
  .iter_kref = (void *)sstc_iter_kref,
  .iter_kvref = (void *)sstc_iter_kvref,
  .iter_skip1 = (void *)sstc_iter_skip1,
  .iter_skip = (void *)sstc_iter_skip,
  .iter_destroy = (void *)sstc_iter_destroy,
};

  static bool
msstb_use_ckeys(struct msst * const msstx1)
{
  bool use_ckeys = true;
  u64 ckeys_size = 0;
  for (u32 i = 0; i < msstx1->nrun; i++) {
    const struct sst_meta * const meta = sst_meta(&(msstx1->ssts[i]));
    if ((meta->totkv != 0) && (meta->ckeyssz == 0)) {
      use_ckeys = false;
      break;
    }
    ckeys_size += meta->ckeyssz;
  }
  if (use_ckeys)
    ssty_build_ckeys_reads += ckeys_size;
  return use_ckeys;
}
// }}} sstc_iter

// msstb {{{

// msstb: a special kind of iterator structure for building ssty (REMIX)
// There are three instances of msstb
// A msstz-compaction always write new tables sequentially (a sorted view),
// and the existing data are already sorted (another sorted view).
// both msstb2 and msstbc perform a two-way merge between the old and new data.
// msstb2 uses the key-value data. It uses binary searches to find merge points.
// msstbc uses the ckeys to perform a regular two-way merge.
// msstbc is the most (I/O) efficient.

// If the input data does not qualify for a two-way merge, msstbm should be used.
// msstbm can be used to create ssty (REMIX) for any kind of inputs (e.g., overlapping).
// msstbm uses miter to perform multi-way merge.
// msstbm automatically uses ckeys if every input table contains ckeys.
// A rough comparison of speed: fast | msstbc > msstbm+ckeys <?> msstb2 > msstbm | slow

struct msstb {
  u32 rankenc; // the current rankenc (rank and flags)
  u32 idx; // index on the full sorted view
  u32 nrun; // the target nrun
  u32 run0; // <= y0->nrun, the tables to reuse in the new ssty
  u32 run1; // iters[run1] is the current sst iter; valid if run1 < nrun
  u32 nkidx; // a copy of y0->ssty->nkidx
  u32 kidx0; // index on ranks (of y0) for bc

  bool bsearch_anchors; // binary search anchors vs. linear scan anchors
  bool bsearch_keys; // binary search keys vs. linear scan keys in a group
  bool dup;   // the new key == old key; msstb2_sync1 can set dup to true
  bool stale;   // the old key should be set as stale

  const u8 * ranks; // shortcut to y0->ssty->ranks

  struct msst * x1; // the input msstx
  struct msst * y0; // the old mssty
  const u8 * tags0; // y0->ssty->tags
  struct miter * miter;

  struct kv * tmp0; // for bc
  struct kv * tmp1; // for bc
  struct sst_iter older;
  struct sst_iter newer;

  // iterb will be moved ahead of iter0 (iterb >= iter0)
  // then iter0 will be used to encode the old kvs until they meet again
  struct mssty_iter iterb; // for binary search only
  struct mssty_iter iter0; // for linear iteration and encoding
  union {
    struct sst_iter * iters[MSST_NRUN]; // for new tables
    struct sstc_iter * citers[MSST_NRUN]; // for bc
  };
};

struct msstb_api {
  struct msstb * (*create)  (struct msst * const msstx1, struct msst * const mssty0, const u32 run0);
  void (*ptrs)              (struct msstb * const b, struct sst_ptr * const ptrs_out); // dump cursor offsets
  struct kv * (*anchor)     (struct msstb * const b); // create anchor key
  u8 (*tag)                (struct msstb * const b); // calculate a 8-bit tag (the lowest 8-bits of crc32c)
  void (*skip1)             (struct msstb * const b);
  void (*destroy)           (struct msstb * const b);
};

  static inline bool
msstb_valid(struct msstb * const b)
{
  return b->rankenc != UINT32_MAX;
}

  static inline u32
msstb_rankenc(struct msstb * const b)
{
  return b->rankenc;
}
// }}} msstb

// msstbm {{{
  static void
msstbm_sync_rank(struct msstb * const b)
{
  if (!miter_valid(b->miter)) {
    b->rankenc = UINT32_MAX;
    return;
  }

  const u32 rank = miter_rank(b->miter);
  debug_assert(rank < MSST_NRUN);
  struct kvref cref;
  miter_kvref(b->miter, &cref);
  const bool stale = (cref.hdr.klen == b->tmp1->klen) && (!memcmp(cref.kptr, b->tmp1->kv, b->tmp1->klen));
  const bool ts = cref.hdr.vlen == SST_VLEN_TS;
  b->rankenc = rank | (stale ? SSTY_STALE : 0u) | (ts ? SSTY_TOMBSTONE : 0u);

  if (!stale) {
    struct kv * const xchg = b->tmp0;
    b->tmp0 = b->tmp1;
    b->tmp1 = xchg;
    kvref_dup2_key(&cref, b->tmp1);
  }
}

  static struct kv *
msstbm_anchor(struct msstb * const b)
{
  const u32 alen = b->idx ? (kv_key_lcp(b->tmp0, b->tmp1)+1) : 0;
  debug_assert(alen <= b->tmp1->klen);
  struct kv * const anchor = kv_create(b->tmp1->kv, alen, NULL, 0); // key only
  return anchor;
}

  static u8
msstbm_tag(struct msstb * const b)
{
  // no extra I/O for accessing the key
  struct kv * const curr = b->tmp1;
  return ssty_tag(kv_crc32c(curr->kv, curr->klen));
}

  static void
msstbm_ptrs(struct msstb * const b, struct sst_ptr * const ptrs)
{
  for (u32 i = 0; i < b->nrun; i++)
    ptrs[i] = b->iters[i]->ptr;
}

  static void
msstbm_skip1(struct msstb * const b)
{
  miter_skip1(b->miter);
  b->idx++;
  msstbm_sync_rank(b);
}

  static struct msstb *
msstbm_create(struct msst * const msstx1, struct msst * const mssty0, const u32 run0)
{
  (void)mssty0;
  (void)run0;
  struct msstb * const b = calloc(1, sizeof(*b));
  b->nrun = msstx1->nrun;
  b->idx = 0;
  b->tmp0 = malloc(sizeof(*b->tmp0) + SST_MAX_KVSZ);
  b->tmp1 = malloc(sizeof(*b->tmp1) + SST_MAX_KVSZ);
  const bool use_ckeys = msstb_use_ckeys(msstx1);
  const struct kvmap_api * const api_build = use_ckeys ? &kvmap_api_sstc : &kvmap_api_sst;
  struct miter * const miter = miter_create();
  b->miter = miter;
  for (u32 i = 0; i < b->nrun; i++)
    b->iters[i] = miter_add(miter, api_build, &msstx1->ssts[i]);

  miter_seek(miter, kref_null());
  if (miter_valid(miter)) {
    struct kvref kvref;
    miter_kvref(miter, &kvref);
    kvref_dup2_key(&kvref, b->tmp1);
    b->tmp1->klen = !b->tmp1->klen; // let the first stale == false
  }
  msstbm_sync_rank(b);
  return b;
}

  static void
msstbm_destroy(struct msstb * const b)
{
  free(b->tmp0);
  free(b->tmp1);
  miter_destroy(b->miter);
  free(b);
}

static const struct msstb_api msstb_api_miter = {
  .create = msstbm_create,
  .ptrs = msstbm_ptrs,
  .anchor = msstbm_anchor,
  .tag = msstbm_tag,
  .skip1 = msstbm_skip1,
  .destroy = msstbm_destroy,
};
// }}} msstbm

// msstb2 {{{
  static void
msstb2_iter(struct msstb * const b, const u8 rank, struct sst_iter * const out)
{
  debug_assert(rank < b->nrun);
  struct sst_iter * const iter = (rank < b->run0) ? &(b->iter0.iters[rank]) : b->iters[rank];
  debug_assert(iter->rank == rank);
  out->sst = iter->sst;
  out->rank = rank;
  out->ptr = iter->ptr;
  // klen and vlen are ignored
  out->kvdata = NULL;
}

  static void
msstb2_sync_rank(struct msstb * const b)
{
  const bool valid = (b->iter0.kidx < b->nkidx) || (b->run1 < b->nrun);
  if (!valid) {
    b->rankenc = UINT32_MAX;
    return;
  }

  if (b->iter0.kidx < b->iterb.kidx) { // use the old
    b->rankenc = b->ranks[b->iter0.kidx] | (b->stale ? SSTY_STALE : 0);
  } else { // use the new
    debug_assert(b->iter0.kidx == b->iterb.kidx);
    const u8 ts = (b->iters[b->run1]->vlen == SST_VLEN_TS) ? SSTY_TOMBSTONE : 0;
    b->rankenc = b->run1 | ts;
  }

  if ((b->rankenc & SSTY_STALE) == 0) {
    b->older = b->newer;
    msstb2_iter(b, b->rankenc & SSTY_RANK, &b->newer);
  }
}

// update iterb to mark the merge point for the current sst_iter
  static void
msstb2_sync_mp(struct msstb * const b)
{
  struct mssty_iter * const iterb = &(b->iterb);
  do {
    if (b->run1 == b->nrun) {
      mssty_iter_park(iterb); // no longer needed
      iterb->kidx = b->nkidx;
      return;
    } else if (sst_iter_valid(b->iters[b->run1])) {
      // use this run1 and iter1
      break;
    }
    b->run1++;
  } while (true);
  struct sst_iter * const iter1 = b->iters[b->run1];

  if (iterb->kidx == b->nkidx) { // !mssty_iter_valid(iterb)
    sst_iter_fix_kv(iter1); // msstb2_sync_rank needs the vlen
    return;
  }
  // seek on iterb; now iter1 is valid
  struct kref kref1;
  sst_iter_kref(iter1, &kref1);
  // let iterb point to the merge point
  mssty_iter_seek_near(iterb, &kref1, b->bsearch_keys);

  // skip placeholders and high-rank keys
  while ((iterb->kidx < b->nkidx) && ((b->ranks[iterb->kidx] & SSTY_RANK) >= b->run0))
    mssty_iter_skip1_dup(iterb);

  struct kref kref0; // the current
  if (mssty_iter_kref_dup(iterb, &kref0)) // mssty_iter is also valid
    b->dup = kref_match(&kref1, &kref0); // may find a dup
}

  static struct kv *
msstb2_anchor(struct msstb * const b)
{
  struct kref tmp0 = {};
  struct kref tmp1 = {};
  sst_iter_kref(&b->older, &tmp0);
  sst_iter_kref(&b->newer, &tmp1);
  const u32 alen = b->idx ? (kref_lcp(&tmp0, &tmp1)+1) : 0;
  debug_assert(alen <= tmp1.len);
  struct kv * const anchor = kv_create(tmp1.ptr, alen, NULL, 0); // key only
  sst_iter_park(&b->older);
  sst_iter_park(&b->newer);
  return anchor;
}

// used when (ckeys == false && tags == true)
// The current implementation will have to read every key
// This configuration is obsolete and should not be used
// TODO: can be improved by reading u8 tags from y0
  static u8
msstb2_tag(struct msstb * const b)
{
  if (b->tags0 && ((b->rankenc & SSTY_RANK) < b->run0)) // tags0 && lo
    return b->tags0[b->iter0.kidx];

  struct kref curr = {};
  sst_iter_kref(&b->newer, &curr);
  const u32 hash32 = kv_crc32c(curr.ptr, curr.len);
  sst_iter_park(&b->newer);
  return ssty_tag(hash32);
}

  static void
msstb2_ptrs(struct msstb * const b, struct sst_ptr * const ptrs)
{
  struct mssty_iter * const iter0 = &(b->iter0);
  for (u32 i = 0; i < b->run0; i++)
    ptrs[i] = iter0->iters[i].ptr;

  for (u32 i = b->run0; i < b->nrun; i++)
    ptrs[i] = b->iters[i]->ptr;
}

  static void
msstb2_skip1(struct msstb * const b)
{
  b->idx++;
  if (b->iter0.kidx < b->iterb.kidx) { // skip an old key
    b->stale = false; // stale is one shot only
    do {
      mssty_iter_skip1_dup(&(b->iter0));
    } while ((b->iter0.kidx < b->nkidx) && ((b->ranks[b->iter0.kidx] & SSTY_RANK) >= b->run0));
    debug_assert(b->iter0.kidx <= b->iterb.kidx);
  } else { // skip a new key
    debug_assert(b->iter0.kidx == b->iterb.kidx);
    sst_iter_skip1(b->iters[b->run1]);
    b->stale = b->dup; // force the next key to be stale
    b->dup = false; // dup is one shot only
    msstb2_sync_mp(b); // update iterb (and run1)
  }
  msstb2_sync_rank(b);
}

  static struct msstb *
msstb2_create_common(struct msst * const msstx1, struct msst * const mssty0, const u32 run0)
{
  debug_assert(msstx1);
  struct msstb * const b = calloc(1, sizeof(*b));
  if (!b)
    return NULL;

  b->x1 = msstx1;
  b->y0 = mssty0;
  b->tags0 = mssty0 ? mssty0->ssty->tags : NULL;
  b->run0 = run0;
  b->run1 = run0; // new tables start with run0
  b->nrun = msstx1->nrun; // the target nrun
  b->newer.ptr.keyid = UINT16_MAX;

  if (run0) {
    debug_assert(mssty0);
    b->nkidx = mssty0->ssty->nkidx; // shortcut
    b->ranks = mssty0->ssty->ranks; // shortcut
  }
  return b;
}

  static struct msstb *
msstb2_create(struct msst * const msstx1, struct msst * const mssty0, const u32 run0)
{
  struct msstb * const b = msstb2_create_common(msstx1, mssty0, run0);
  if (run0) {
    mssty_iter_init(&(b->iterb), mssty0);
    mssty_iter_seek_null(&(b->iterb));
    mssty_iter_init(&(b->iter0), mssty0);
    mssty_iter_seek_null(&(b->iter0));
    for (u8 i = 0; i < run0; i++)
      mssty_iter_fix_rank(&(b->iter0), i);

    // skip the first a few stale keys
    while ((b->iter0.kidx < b->nkidx) && ((b->ranks[b->iter0.kidx] & SSTY_RANK) >= b->run0))
      mssty_iter_skip1_dup(&(b->iter0));
  }

  u32 newcnt = 0;
  for (u32 i = run0; i < b->nrun; i++) {
    b->iters[i] = sst_iter_create(&(msstx1->ssts[i]));
    b->iters[i]->rank = i;
    sst_iter_seek_null(b->iters[i]);
    newcnt += msstx1->ssts[i].totkv;
  }

  // size ratio between the old and new sorted views; old:new, 1 <= ratio
  const u32 ratio = (newcnt && (newcnt < b->nkidx)) ? (b->nkidx / newcnt) : 1;
  // compensate the linear search for locality and efficiency
  b->bsearch_keys = ratio > (SSTY_DBITS + run0);

  msstb2_sync_mp(b);
  debug_assert(b->iter0.kidx <= b->iterb.kidx);
  msstb2_sync_rank(b);
  return b;
}

  static void
msstb2_destroy(struct msstb * const b)
{
  mssty_iter_park(&(b->iterb));
  mssty_iter_park(&(b->iter0));
  for (u32 i = b->run0; i < b->nrun; i++)
    sst_iter_destroy(b->iters[i]);
  free(b);
}

static const struct msstb_api msstb_api_b2 = {
  .create = msstb2_create,
  .ptrs = msstb2_ptrs,
  .anchor = msstb2_anchor,
  .tag = msstb2_tag,
  .skip1 = msstb2_skip1,
  .destroy = msstb2_destroy,
};
// }}} msstb2

// msstbc {{{
  static void
msstbc_ptrs(struct msstb * const b, struct sst_ptr * const ptrs)
{
  for (u32 i = 0; i < b->nrun; i++)
    ptrs[i] = b->citers[i]->iter.ptr;
}

  static void
msstbc_sync_lo(struct msstb * const b)
{
  while ((b->kidx0 < b->nkidx) && ((b->ranks[b->kidx0] & SSTY_RANK) >= b->run0))
    b->kidx0++;
}

  static void
msstbc_sync_hi(struct msstb * const b)
{
  while ((b->run1 < b->nrun) && (!sstc_iter_valid(b->citers[b->run1])))
    b->run1++;
}

  static void
msstbc_sync_rank(struct msstb * const b)
{
  const bool validlo = b->kidx0 < b->nkidx;
  const bool validhi = b->run1 < b->nrun;

  if (validlo) {
    const u8 loenc = b->ranks[b->kidx0];
    const u8 lorank = loenc & SSTY_RANK;
    debug_assert(lorank < b->run0);
    const int cmp = validhi ? sstc_iter_compare(b->citers[lorank], b->citers[b->run1]) : -1;
    if (cmp < 0) { // use lo
      b->rankenc = lorank;
      if (b->stale || (loenc & SSTY_STALE))
        b->rankenc |= SSTY_STALE;
    } else { // use hi
      b->rankenc = b->run1;
      debug_assert(b->stale == false);
    }
    b->stale = cmp == 0;
  } else { // validlo == false
    if (validhi) { // use hi
      b->rankenc = b->run1;
      //b->stale = false; // no need
    } else { // stop
      b->rankenc = UINT32_MAX;
      return;
    }
  }

  // check ts
  struct sstc_iter * const citer = b->citers[b->rankenc & SSTY_RANK];
  if (sstc_iter_ts(citer))
    b->rankenc |= SSTY_TOMBSTONE;

  if ((b->rankenc & SSTY_STALE) == 0) {
    struct kv * const xchg = b->tmp0;
    b->tmp0 = b->tmp1;
    b->tmp1 = xchg;
    struct kref cref;
    if (sstc_iter_kref(citer, &cref)) {
      b->tmp1->klen = cref.len;
      memcpy(b->tmp1->kv, cref.ptr, cref.len); // hash is ignored
    } else {
      debug_die();
    }
  }
}

  static u8
msstbc_tag(struct msstb * const b)
{
  if (b->tags0 && ((b->rankenc & SSTY_RANK) < b->run0)) // tags0 && lo
    return b->tags0[b->kidx0];

  // no extra I/O for accessing the key
  struct kv * const curr = b->tmp1;
  return ssty_tag(kv_crc32c(curr->kv, curr->klen));
}

  static void
msstbc_skip1(struct msstb * const b)
{
  const u8 rank = b->rankenc & SSTY_RANK;
  sstc_iter_skip1(b->citers[rank]);
  b->idx++;
  if (rank < b->run0) { // lo
    b->kidx0++;
    msstbc_sync_lo(b);
  } else {
    msstbc_sync_hi(b);
  }
  msstbc_sync_rank(b);
}

  static struct msstb *
msstbc_create(struct msst * const msstx1, struct msst * const mssty0, const u32 run0)
{
  struct msstb * const b = msstb2_create_common(msstx1, mssty0, run0);

  b->tmp0 = malloc(sizeof(*b->tmp0) + SST_MAX_KVSZ);
  b->tmp1 = malloc(sizeof(*b->tmp1) + SST_MAX_KVSZ);

  for (u32 i = 0; i < b->nrun; i++) {
    b->citers[i] = sstc_iter_create(&(msstx1->ssts[i]));
    b->citers[i]->iter.rank = i;
  }
  msstbc_sync_lo(b);
  msstbc_sync_hi(b);
  msstbc_sync_rank(b);
  return b;
}

  static void
msstbc_destroy(struct msstb * const b)
{
  free(b->tmp0);
  free(b->tmp1);
  for (u32 i = 0; i < b->nrun; i++)
    sstc_iter_destroy(b->citers[i]);
  free(b);
}

static const struct msstb_api msstb_api_bc = {
  .create = msstbc_create,
  .ptrs = msstbc_ptrs,
  .anchor = msstbm_anchor, // reuse msstbm_anchor
  .tag = msstbc_tag,
  .skip1 = msstbc_skip1,
  .destroy = msstbc_destroy,
};
// }}} msstbc

// sort {{{
// check if tables at run0 to nrun overlap
  static const struct msstb_api *
ssty_build_api(struct msst * const msstx1, const u32 run0)
{
  const u32 nrun = msstx1->nrun;
  struct kv * const last = malloc(sizeof(*last) + SST_MAX_KVSZ);
  struct kv * const tmp = malloc(sizeof(*last) + SST_MAX_KVSZ);
  last->klen = UINT32_MAX;
  bool overlap = false;
  for (u32 i = run0; i < nrun; i++) {
    if (msstx1->ssts[i].totkv == 0)
      continue;

    struct kv * const first = sst_first_key(&(msstx1->ssts[i]), tmp);
    if ((last->klen != UINT32_MAX) && (kv_compare(last, first) >= 0)) {
      overlap = true;
      break;
    }
    sst_last_key(&(msstx1->ssts[i]), last);
  }

  free(last);
  free(tmp);
  if (overlap) {
    return &msstb_api_miter;
  } else if (msstb_use_ckeys(msstx1)) {
    return &msstb_api_bc;
  } else {
    return &msstb_api_b2;
  }
}

  static void
ssty_build_sort_msstb(struct ssty_build_info * const bi)
{
  const struct msstb_api * const api = ssty_build_api(bi->x1, bi->run0);
  struct msstb * const b = api->create(bi->x1, bi->y0, bi->run0);
  if (!b)
    debug_die();

  const u32 nrun = bi->x1->nrun;
  u8 * const ranks = bi->ranks;
  u8 * const tags = bi->tags;

  struct sst_ptr * ptrs = bi->ptrs;

  u32 kidx0 = 0; // id of the first key of multiple identical keys (<= kidx1)
  u32 kidx1 = 0; // the current key
  u32 valid = 0; // number of unique and valid keys (unique_keys - tombstones)
  u32 aidx = 0; // the next anchor's index; generate anchor key when kidx0 == aidx

  while (msstb_valid(b)) {
    const u32 rankenc = msstb_rankenc(b);
    debug_assert(rankenc < SSTY_INVALID);
    debug_assert((rankenc & SSTY_RANK) < nrun);

    if ((rankenc & SSTY_STALE) == 0) { // not a stale key
      api->ptrs(b, ptrs); // save ptrs of every newest version
      kidx0 = kidx1;
      bi->uniqx[rankenc & SSTY_RANK]++;
      if ((rankenc & SSTY_TOMBSTONE) == 0)
        valid++;
    } else if ((kidx0 ^ kidx1) >> SSTY_DBITS) { // crossing boundary
      const u32 gap = kidx1 - kidx0;
      memmove(&(ranks[kidx1]), &(ranks[kidx0]), gap); // move forward
      memset(&(ranks[kidx0]), SSTY_INVALID, gap); // fill with INVALID
      if (tags)
        memmove(&(tags[kidx1]), &(tags[kidx0]), gap); // move forward

      kidx0 += gap;
      kidx1 += gap;
    }

    if (kidx0 == aidx) { // generate anchors
      bi->anchors[aidx >> SSTY_DBITS] = api->anchor(b);
      aidx += SSTY_DIST;
      ptrs += nrun; // ptrs accepted
    }
    ranks[kidx1] = (u8)rankenc;
    if (tags)
      tags[kidx1] = api->tag(b);
    api->skip1(b);
    kidx1++;
  }
  api->destroy(b);

  // metadata
  bi->nkidx = kidx1;
  bi->nsecs = (kidx1 + SSTY_DIST - 1) >> SSTY_DBITS;
  bi->valid = valid;
}
// }}} sort

// main {{{
// layout
// ranks: sz0
// tags: sz1 at tagoff
// ptrs: sz2  at ptroff [inr1]
// anchors: sz3
// ioffs: sz4   at ioff1 [inr1]
// ikeys2: sz5
// ioffs2: sz6  at ioff2 [inr2]
// meta
// y0 and run0 are optional
  static u32
ssty_build_at(const int dfd, struct msst * const msstx1,
    const u64 seq, const u32 nrun, struct msst * const mssty0, const u32 run0, const bool gen_tags)
{
  // open ssty file for output
  debug_assert(nrun == msstx1->nrun);
  char fn[24];
  const u64 magic = seq * 100lu + nrun;
  sprintf(fn, "%03lu.ssty", magic);
  const int fdout = openat(dfd, fn, O_WRONLY|O_CREAT|O_TRUNC, 00644);
  if (fdout < 0)
    return 0;

  u32 totkv = 0;
  size_t totsz = 0;
  for (u32 i = 0; i < nrun; i++) {
    totkv += msstx1->ssts[i].totkv;
    totsz += msstx1->ssts[i].fsize;
  }
  debug_assert(totsz <= UINT32_MAX);

  const u32 maxkidx = (totkv + SSTY_DIST) * 2; // large enough
  const u32 maxsecs = maxkidx >> SSTY_DBITS;
  u8 * const ranks = malloc(maxkidx + 128); // double size is enough
  struct sst_ptr * const ptrs = malloc(sizeof(*ptrs) * (maxsecs * nrun + MSST_NRUN + 8));
  struct kv ** const anchors = malloc(sizeof(*anchors) * maxsecs);
  debug_assert(ranks && ptrs && anchors);

  u8 * const tags = gen_tags ? malloc((maxkidx + 128) * sizeof(tags[0])) : NULL; // double size is enough
  debug_assert(tags || (!gen_tags));

  struct ssty_build_info bi = {.x1 = msstx1, .y0 = mssty0,
    .ranks = ranks, .ptrs = ptrs, .anchors = anchors, .tags = tags,
    .dfd = dfd, .run0 = run0};

  ssty_build_sort_msstb(&bi);
  debug_assert(bi.nkidx <= maxkidx);
  debug_assert(bi.nsecs <= maxsecs);
  const u32 nkidx = bi.nkidx;
  const u32 nsecs = bi.nsecs;
  // write ranks // x16 for simd in seek_local
  const u32 sz0 = (u32)bits_round_up(sizeof(u8) * nkidx + 1, SSTY_DBITS);
  memset(&ranks[nkidx], (int)nrun, sizeof(ranks[0]) * SSTY_DIST); // pad with nrun (at least 1 byte, up to 16)
  write(fdout, ranks, sz0);
  free(ranks);

  const u32 sz1 = tags ? (sz0 * sizeof(tags[0])) : 0;
  const u32 tagoff = tags ? sz0 : 0;
  if (tags) {
    memset(&tags[nkidx], 0, sizeof(tags[0]) * SSTY_DIST); // pad with 0
    write(fdout, tags, sz1);
    free(tags);
  }

  //write level indicators and seek ptrs
  const u32 ptroff = sz0 + sz1;
  const u32 sz2 = (u32)(sizeof(struct sst_ptr) * nsecs * nrun);
  write(fdout, ptrs, sz2);
  free(ptrs);

  // gen anchors
  const u32 baseoff1 = ptroff + sz2;
  u32 * const ioffs = malloc(sizeof(*ioffs) * nsecs);
  struct kvenc * const aenc = kvenc_create();
  for (u64 i = 0; i < nsecs; i++) {
    const u32 ioff = baseoff1 + kvenc_size(aenc);
    ioffs[i] = ioff;
    const u32 klen = anchors[i]->klen;
    const u32 est = vi128_estimate_u32(klen) + klen;
    const u32 rem = PGSZ - (ioff % PGSZ);
    if (est > rem) {
      kvenc_append_raw(aenc, NULL, rem);
      ioffs[i] += rem;
      debug_assert((ioffs[i] % PGSZ) == 0);
    }
    kvenc_append_vi128(aenc, klen);
    kvenc_append_raw(aenc, anchors[i]->kv, klen);
  }
  kvenc_append_padding(aenc, 2);
  const u32 sz3 = kvenc_size(aenc);
  kvenc_write(aenc, fdout);
  kvenc_reset(aenc);
  const u32 sz4 = sizeof(*ioffs) * nsecs;
  write(fdout, ioffs, sz4);
  const u32 ioffsoff = baseoff1 + sz3;

  // ikeys2
  const u32 baseoff2 = ioffsoff + sz4;
  const u32 pga = nsecs ? (ioffs[0] / PGSZ) : 0; // first pageno of index blocks
  const u32 pgz = nsecs ? (ioffs[nsecs-1] / PGSZ) : 0; // last pageno of index blocks
  const u32 ipages = nsecs ? (pgz - pga + 1) : 0; // totol number of pages of index blocks
  u32 * const ioffs2 = malloc(sizeof(*ioffs2) * ipages);

  u32 i1 = 0;
  u32 * pend2 = NULL;
  for (u32 i = 0; i < ipages; i++) {
    // search for the first anchor key in the block
    while ((ioffs[i1] / PGSZ) != (pga + i))
      i1++;
    if (pend2)
      *pend2 = i1;

    const u32 ioff2 = baseoff2 + kvenc_size(aenc);
    ioffs2[i] = ioff2; // offset of this entry
    kvenc_append_u32(aenc, i1);
    pend2 = kvenc_append_u32_backref(aenc);
    struct kv * const anchor = anchors[i1];
    kvenc_append_vi128(aenc, anchor->klen);
    kvenc_append_raw(aenc, anchor->kv, anchor->klen);
    kvenc_append_padding(aenc, 2); // x4
    i1++;
  }
  if (pend2)
    *pend2 = nsecs;

  free(ioffs);
  for (u64 i = 0; i < nsecs; i++)
    free(anchors[i]);
  free(anchors);

  const u32 sz5 = kvenc_size(aenc);
  kvenc_write(aenc, fdout);
  kvenc_destroy(aenc);
  const u32 sz6 = sizeof(*ioffs2) * ipages;
  write(fdout, ioffs2, sz6);
  free(ioffs2);
  const u32 ioffsoff2 = baseoff2 + sz5;
  const u32 metaoff = ioffsoff2 + sz6;

  // ssty metadata
  struct ssty_meta meta = {
    .nrun = nrun, .nkidx = nkidx, .tagoff = tagoff, .ptroff = ptroff, .inr1 = nsecs,
    .ioff1 = ioffsoff, .inr2 = ipages, .ioff2 = ioffsoff2, .totkv = totkv,
    .totsz = (u32)totsz, .valid = bi.valid, .magic = magic,};

  // in the ssty file, each uniqx[i] is the number of unique keys at [i:n] levels if they are merged
  u32 uniq = 0;
  for (u32 i = nrun-1; i < nrun; i--) {
    uniq += bi.uniqx[i];
    meta.uniqx[i] = uniq;
  }
  const bool succ = write(fdout, &meta, sizeof(meta)) == sizeof(meta);
  const u32 fsize = metaoff + (u32)sizeof(meta);
  debug_assert(fsize == (sz0+sz1+sz2+sz3+sz4+sz5+sz6+(u32)sizeof(meta)));
  debug_assert(fsize < UINT32_MAX);

  //fprintf(stderr, "%s ranks %u tags %u ptrs %u a1 %u o1 %u a2 %u o2 %u file %u\n",
  //    __func__, sz0, sz1, sz2, sz3, sz4, sz5, sz6, fsize);

  // done
  fsync(fdout);
  close(fdout);
  return succ ? fsize : 0;
}

  u32
ssty_build(const char * const dirname, struct msst * const msstx1,
    const u64 seq, const u32 nrun, struct msst * const mssty0, const u32 run0, const bool tags)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return 0;
  const u32 ret = ssty_build_at(dfd, msstx1, seq, nrun, mssty0, run0, tags);
  close(dfd);
  return ret;
}
// }}} main

// }}} ssty_build

// msstv {{{
struct msstv { // similar to a version in leveldb
  u64 version;
  u64 nr; // number of partitions
  u64 nslots; // assert(nr <= nslots)
  struct msstv * next; // to older msstvs
  au64 rdrcnt; // active readers; updated concurrently
  struct rcache * rc; // rcache

  struct msstv_part {
    struct kv * anchor; // magic in anchor->priv; anchor->vlen == 1 for rejected partition
    struct msst * msst; // mssty
  } es[0];
};

struct msstv_iter {
  struct msstv * v;
  u64 nr;
  u64 i; // select mssty
  struct mssty_iter iter;
};

struct msstv_ref { // ref is iter
  struct msstv_iter vi;
};

  inline void
msstv_rcache(struct msstv * const v, struct rcache * const rc)
{
  v->rc = rc;
  for (u64 i = 0; i < v->nr; i++)
    msst_rcache(v->es[i].msst, rc);
}

// for debugging now
  struct msstv *
msstv_create(const u64 nslots, const u64 version)
{
  // msstv does not record nslots
  // caller need to do it right
  struct msstv * const v = calloc(1, sizeof(*v) + (sizeof(v->es[0]) * nslots));
  v->version = version;
  v->nslots = nslots;
  // v->next is maintained externally
  return v;
}

// anchor can be NULL for auto detection; for new partitions only
// a private copy of the anchor will be duplicated if not NULL
  void
msstv_append(struct msstv * const v, struct msst * const msst, const struct kv * const anchor)
{
  debug_assert(msst && msst->ssty);
  debug_assert(v->nr < v->nslots);

  struct msstv_part * const e = &(v->es[v->nr]);
  if (v->nr && (anchor == NULL)) { // auto generate anchor
    struct kv * const first = mssty_first(msst, NULL); // malloced
    debug_assert(first); // major compaction should never generate an empty mssty
    struct kv * const plast = mssty_last(v->es[v->nr-1].msst, NULL); // malloced; might be NULL
    if (plast) {
      first->klen = kv_key_lcp(plast, first) + 1;
      //kv_update_hash(first);
      free(plast);
    }
    e->anchor = first;
  } else {
    debug_assert(anchor);
    e->anchor = kv_dup_key(anchor);
  }

  msst->refcnt++;
  // save magic in anchor->priv; anchor->hash is not saved
  e->anchor->priv = msst->ssty->meta->magic;
  e->msst = msst;
  v->nr++;
}

// save to a file
  static bool
msstv_save(struct msstv * const v, const int dfd)
{
  char fn[24];
  sprintf(fn, "%lu.ver", v->version);
  int fd = openat(dfd, fn, O_CREAT|O_WRONLY|O_TRUNC, 00644);
  if (fd < 0)
    return false;

  FILE * const fout = fdopen(fd, "w");
  if (fout == NULL)
    return false;
  setvbuf(fout, NULL, _IOFBF, 1lu << 16); // 64kB
  fwrite(&(v->version), sizeof(v->version), 1, fout);
  fwrite(&(v->nr), sizeof(v->nr), 1, fout);
  char bufz[8] = {};
  for (u64 i = 0; i < v->nr; i++) {
    const u64 keysize = key_size(v->es[i].anchor);
    fwrite(v->es[i].anchor, keysize, 1, fout);
    const u64 size = bits_round_up(keysize, 3);
    if (size > keysize)
      fwrite(bufz, size - keysize, 1, fout);
  }
  fclose(fout);
  return true;
}

// open version and open all msstys
  static struct msstv *
msstv_open_at(const int dfd, const char * const filename)
{
  const int fd = openat(dfd, filename, O_RDONLY);
  if (fd < 0)
    return NULL;

  const u64 filesz = fdsize(fd);
  if (filesz < (sizeof(u64) * 2 + sizeof(struct kv))) {
    close(fd);
    return NULL;
  }

  u8 * const buf = malloc(filesz);
  const ssize_t nread = pread(fd, buf, filesz, 0);
  if (filesz != (u64)nread) {
    free(buf);
    close(fd);
    return NULL;
  }
  close(fd);

  const u64 v1 = ((const u64 *)buf)[0];
  const u64 nr = ((const u64 *)buf)[1];
  debug_assert(nr);

  // open msstys
  struct msstv * const v = msstv_create(nr, v1);
  u8 * cursor = buf + (sizeof(u64) * 2);
  for (u64 i = 0; i < nr; i++) {
    struct kv * const anchor = (typeof(anchor))cursor;
    const u64 magic = anchor->priv;
    // rc: msstz_open sets rc later; compaction sets rc manually
    struct msst * const mssty = mssty_open_at(dfd, magic / 100lu, (u32)(magic % 100lu));
    if (!mssty) {
      msstv_destroy(v);
      free(buf);
      return NULL;
    }
    msstv_append(v, mssty, anchor);
    cursor += (bits_round_up(key_size(anchor), 3));
  }

  debug_assert((u64)(cursor - buf) == filesz);
  free(buf);
  return v;
}

  struct msstv *
msstv_open(const char * const dirname, const char * const filename)
{
  const int dfd = open(dirname, O_RDONLY|O_DIRECTORY);
  if (dfd < 0)
    return NULL;
  struct msstv * const v = msstv_open_at(dfd, filename);
  close(dfd);
  return v;
}

  struct msstv *
msstv_open_version(const char * const dirname, const u64 version)
{
  char filename[32];
  sprintf(filename, "%lu.ver", version);
  return msstv_open(dirname, filename);
}

  static void
msstv_destroy_lazy(struct msstv * const v)
{
  for (u64 i = 0; i < v->nr; i++) {
    // mssty can be shared by mutliple versions
    struct msst * const msst = v->es[i].msst;
    if (msst->refcnt == 1)
      mssty_destroy_lazy(msst);
    else
      msst->refcnt--;

    free(v->es[i].anchor);
  }
  free(v);
}

// it does not free the msstys
  void
msstv_destroy(struct msstv * const v)
{
  for (u64 i = 0; i < v->nr; i++) {
    // mssty can be shared by mutliple versions
    struct msst * const msst = v->es[i].msst;
    if (msst->refcnt == 1)
      mssty_destroy(msst);
    else
      msst->refcnt--;

    free(v->es[i].anchor);
  }
  free(v);
}

  struct msstv_ref *
msstv_ref(struct msstv * const v)
{
  struct msstv_iter * const vi = calloc(1, sizeof(*vi));
  vi->v = v;
  vi->nr = v->nr;
  vi->i = v->nr; // invalid
  return (struct msstv_ref *)vi;
}

  struct msstv *
msstv_unref(struct msstv_ref * const ref)
{
  struct msstv * const v = ref->vi.v;
  free(ref);
  return v;
}

  static u64
msstv_search_le(struct msstv * const v, const struct kref * const key)
{
  u64 l = 0;
  u64 r = v->nr;
  while ((l + 1) < r) {
    const u64 m = (l + r) >> 1;
    const int cmp = kref_kv_compare(key, v->es[m].anchor);
    if (cmp < 0)
      r = m; // m always > 0
    else if (cmp > 0)
      l = m;
    else
      return m;
  }
  return l;
}

  struct kv *
msstv_get(struct msstv_ref * const ref, const struct kref * const key, struct kv * const out)
{
  struct msstv_iter * const vi = (typeof(vi))ref;
  const u64 i = msstv_search_le(vi->v, key);
  debug_assert(i < vi->nr);
  if (i != vi->i)
    mssty_iter_init(&(vi->iter), vi->v->es[i].msst);
  struct kv * const ret = mssty_get((struct mssty_ref *)&(vi->iter), key, out);
  mssty_iter_park(&(vi->iter));
  return ret;
}

  bool
msstv_probe(struct msstv_ref * const ref, const struct kref * const key)
{
  struct msstv_iter * const vi = (typeof(vi))ref;
  const u64 i = msstv_search_le(vi->v, key);
  debug_assert(i < vi->nr);
  if (i != vi->i)
    mssty_iter_init(&(vi->iter), vi->v->es[i].msst);
  const bool r = mssty_probe((struct mssty_ref *)&(vi->iter), key);
  mssty_iter_park(&(vi->iter));
  return r;
}

  struct kv *
msstv_get_ts(struct msstv_ref * const ref, const struct kref * const key, struct kv * const out)
{
  struct msstv_iter * const vi = (typeof(vi))ref;
  const u64 i = msstv_search_le(vi->v, key);
  debug_assert(i < vi->nr);
  if (i != vi->i)
    mssty_iter_init(&(vi->iter), vi->v->es[i].msst);
  struct kv * const ret = mssty_get_ts((struct mssty_ref *)&(vi->iter), key, out);
  mssty_iter_park(&(vi->iter));
  return ret;
}

  bool
msstv_probe_ts(struct msstv_ref * const ref, const struct kref * const key)
{
  struct msstv_iter * const vi = (typeof(vi))ref;
  const u64 i = msstv_search_le(vi->v, key);
  debug_assert(i < vi->nr);
  if (i != vi->i)
    mssty_iter_init(&(vi->iter), vi->v->es[i].msst);
  const bool r = mssty_probe_ts((struct mssty_ref *)&(vi->iter), key);
  mssty_iter_park(&(vi->iter));
  return r;
}

  bool
msstv_get_value_ts(struct msstv_ref * const ref, const struct kref * const key,
    void * const vbuf_out, u32 * const vlen_out)
{
  struct msstv_iter * const vi = (typeof(vi))ref;
  const u64 i = msstv_search_le(vi->v, key);
  debug_assert(i < vi->nr);
  if (i != vi->i)
    mssty_iter_init(&(vi->iter), vi->v->es[i].msst);
  const bool r = mssty_get_value_ts((struct mssty_ref *)&(vi->iter), key, vbuf_out, vlen_out);
  mssty_iter_park(&(vi->iter));
  return r;
}

  static bool
msstv_iter_valid_i(struct msstv_iter * const vi)
{
  return (vi->i < vi->nr);
}

  bool
msstv_iter_valid(struct msstv_iter * const vi)
{
  debug_assert((vi->i < vi->nr) ? mssty_iter_valid(&(vi->iter)) : true);
  return msstv_iter_valid_i(vi);
}

  struct msstv_iter *
msstv_iter_create(struct msstv_ref * const ref)
{
  // reuse ref
  return (struct msstv_iter *)ref;
}

  void
msstv_iter_destroy(struct msstv_iter * const vi)
{
  if (msstv_iter_valid(vi))
    mssty_iter_park(&(vi->iter));
}

  void
msstv_iter_seek(struct msstv_iter * const vi, const struct kref * const key)
{
  struct msstv * const v = vi->v;
  const u64 i0 = msstv_search_le(v, key);
  debug_assert(i0 < vi->nr);
  if (i0 != vi->i) {
    if (msstv_iter_valid(vi))
      mssty_iter_park(&(vi->iter));
    vi->i = i0;
    mssty_iter_init(&(vi->iter), v->es[i0].msst);
  }

  do {
    mssty_iter_seek(&(vi->iter), key);
    if (mssty_iter_valid(&(vi->iter)))
      return;

    mssty_iter_park(&(vi->iter));
    vi->i++;
    if (!msstv_iter_valid_i(vi))
      return;
    mssty_iter_init(&(vi->iter), v->es[vi->i].msst);
  } while (true);
}

  struct kv *
msstv_iter_peek(struct msstv_iter * const vi, struct kv * const out)
{
  if (!msstv_iter_valid(vi))
    return NULL;
  return mssty_iter_peek(&(vi->iter), out);
}

  bool
msstv_iter_kref(struct msstv_iter * const vi, struct kref * const kref)
{
  if (!msstv_iter_valid(vi))
    return false;
  return mssty_iter_kref(&(vi->iter), kref);
}

  bool
msstv_iter_kvref(struct msstv_iter * const vi, struct kvref * const kvref)
{
  if (!msstv_iter_valid(vi))
    return false;
  return mssty_iter_kvref(&(vi->iter), kvref);
}

  inline u64
msstv_iter_retain(struct msstv_iter * const vi)
{
  debug_assert(msstv_iter_valid(vi));
  return mssty_iter_retain(&vi->iter);
}

  inline void
msstv_iter_release(struct msstv_iter * const vi, const u64 opaque)
{
  sst_blk_release(vi->v->rc, (const u8 *)opaque);
}

  void
msstv_iter_skip1(struct msstv_iter * const vi)
{
  debug_assert(msstv_iter_valid(vi));
  struct mssty_iter * const iter = &(vi->iter);
  mssty_iter_skip1(iter);
  while (unlikely(!mssty_iter_valid(iter))) { // next partition
    mssty_iter_park(iter);
    vi->i++;
    if (msstv_iter_valid_i(vi)) {
      mssty_iter_init(iter, vi->v->es[vi->i].msst);
      mssty_iter_seek_null(iter);
    } else { // invalid
      return;
    }
  }
}

  void
msstv_iter_skip(struct msstv_iter * const vi, const u32 nr)
{
  if (!msstv_iter_valid(vi))
    return;
  struct mssty_iter * const iter = &(vi->iter);
  for (u32 i = 0; i < nr; i++) {
    mssty_iter_skip1(iter);
    while (unlikely(!mssty_iter_valid(iter))) { // next partition
      mssty_iter_park(iter);
      vi->i++;
      if (msstv_iter_valid_i(vi)) {
        mssty_iter_init(iter, vi->v->es[vi->i].msst);
        mssty_iter_seek_null(iter);
      } else { // invalid
        return;
      }
    }
  }
}

  struct kv *
msstv_iter_next(struct msstv_iter * const vi, struct kv * const out)
{
  struct kv * const ret = msstv_iter_peek(vi, out);
  if (msstv_iter_valid(vi))
    msstv_iter_skip1(vi);
  return ret;
}

  inline bool
msstv_iter_ts(struct msstv_iter * const vi)
{
  // assume vi is valid
  debug_assert(msstv_iter_valid(vi));
  return mssty_iter_ts(&(vi->iter));
}

  void
msstv_iter_seek_ts(struct msstv_iter * const vi, const struct kref * const key)
{
  msstv_iter_seek(vi, key);
  while (msstv_iter_valid(vi) && msstv_iter_ts(vi))
    msstv_iter_skip1(vi);
}

  void
msstv_iter_skip1_ts(struct msstv_iter * const vi)
{
  if (!msstv_iter_valid(vi))
    return;

  msstv_iter_skip1(vi);
  while (msstv_iter_valid(vi) && msstv_iter_ts(vi))
    msstv_iter_skip1(vi);
}

  void
msstv_iter_skip_ts(struct msstv_iter * const vi, const u32 nr)
{
  for (u32 i = 0; i < nr; i++) {
    if (!msstv_iter_valid(vi))
      return;

    msstv_iter_skip1(vi);
    while (msstv_iter_valid(vi) && msstv_iter_ts(vi))
      msstv_iter_skip1(vi);
  }
}

  struct kv *
msstv_iter_next_ts(struct msstv_iter * const vi, struct kv * const out)
{
  struct kv * const ret = msstv_iter_peek(vi, out);
  msstv_iter_skip1_ts(vi);
  return ret;
}

  void
msstv_iter_park(struct msstv_iter * const vi)
{
  if (msstv_iter_valid(vi))
    mssty_iter_park(&(vi->iter));
}

  void
msstv_fprint(struct msstv * const v, FILE * const out)
{
  fprintf(out, "%s v %lu nr %lu refcnt %lu rcache %s\n",
      __func__, v->version, v->nr, atomic_load_explicit(&v->rdrcnt, MO_CONSUME), v->rc ? "ON" : "OFF");

  for (u64 i = 0; i < v->nr; i++) {
    struct msst * const msst = v->es[i].msst;
    debug_assert(v->rc == msst->rc);
    struct ssty * const ssty = msst->ssty;
    ssty_fprint(ssty, out);
    const u64 magic = v->es[i].anchor->priv;
    fprintf(out, "%s [%3lu %6.3lu]", __func__, i, magic);
    for (u32 j = 0; j < ssty->nrun; j++) {
      struct sst * const sst = &(msst->ssts[j]);
      debug_assert(v->rc == sst->rc);
      fprintf(out, " %7u:%9u ", sst->totkv, sst->fsize);
    }
    kv_print(v->es[i].anchor, "sn", out);
  }
}

  struct kv **
msstv_anchors(struct msstv * const v)
{
  struct kv ** const ret = malloc(sizeof(ret[0]) * (v->nr+1));
  for (u64 i = 0; i < v->nr; i++)
    ret[i] = v->es[i].anchor;
  ret[v->nr] = NULL;
  return ret;
}
// }}} msstv

// msstz {{{

// msst zoo
struct msstz {
  struct msstv * volatile hv; // the newest version
  au64 seq; // next available seq
  char * dirname;

  // compaction parameters
  struct msstz_cfg cfg;
  struct rcache * rc; // read-only cache

  double t0;
  int logfd;
  int dfd;
  u64 stat_time; // time spent in comp()
  u64 stat_writes; // total bytes written to sstx&ssty
  u64 stat_reads; // total bytes read through rcache

  u64 padding1[7];
  rwlock head_lock; // writer: compaction, gc
};

  int
msstz_logfd(struct msstz * const z)
{
  return z->logfd;
}

  static void
msstz_head_sync(const int dfd, const u64 version)
{
  char basefn[24];
  sprintf(basefn, "./%lu.ver", version);

  unlinkat(dfd, "HEAD", 0);
  symlinkat(basefn, dfd, "HEAD");

  unlinkat(dfd, "HEAD1", 0);
  symlinkat(basefn, dfd, "HEAD1");
  return;
}

// create empty store
  static struct msstv *
msstz_create_v0(const int dfd)
{
  // msstx nrun = 0
  struct msst * const msst = msstx_open_at(dfd, 0, 0);
  if (!msst)
    return NULL;

  if (!ssty_build_at(dfd, msst, 0, 0, NULL, 0, false)) {
    msstx_destroy(msst);
    return NULL;
  }

  if (!mssty_open_y_at(dfd, msst)) {
    msstx_destroy(msst);
    return NULL;
  }

  // msstv
  struct msstv * const v = msstv_create(1, 1); // version = 1
  if (v == NULL) {
    mssty_destroy(msst);
    return NULL;
  }
  msstv_append(v, msst, kv_null());
  return v;
}

  struct msstz *
msstz_open(const char * const dirname, const u64 cache_size_mb, const struct msstz_cfg * const cfg)
{
  // get the dir
  int dfd = open(dirname, O_RDONLY | O_DIRECTORY);
  if (dfd < 0) {
    mkdir(dirname, 00777);
    dfd = open(dirname, O_RDONLY | O_DIRECTORY);
  }
  if (dfd < 0)
    return NULL;

  // get a version
  struct msstv * hv = msstv_open_at(dfd, "HEAD");
  if (hv == NULL)
    hv = msstv_open_at(dfd, "HEAD1");
  if (hv == NULL)
    hv = msstz_create_v0(dfd);
  if (hv == NULL) {
    close(dfd);
    return NULL;
  }

  atomic_store_explicit(&hv->rdrcnt, 0, MO_RELAXED);
  u64 seq = 0;
  for (u64 i = 0; i < hv->nr; i++) {
    const u64 magic = hv->es[i].anchor->priv;
    const u64 seq1 = magic / 100;
    if (seq < seq1)
      seq = seq1;
  }
  const u64 seq0 = seq + 1; // use a new seq

  struct msstz * const z = yalloc(sizeof(*z));
  debug_assert(z);
  memset(z, 0, sizeof(*z));
  if (cache_size_mb)
    z->rc = rcache_create(cache_size_mb, 16);

  z->seq = seq0;
  z->hv = hv;
  msstv_rcache(hv, z->rc);

  z->dirname = strdup(dirname);
  debug_assert(z->dirname);

  z->cfg = cfg ? *cfg : msstz_cfg_default;
  z->dfd = dfd;

  char logfn[80];
  char buf[64];
  time_stamp2(buf, 64);

  sprintf(logfn, "log-%s", buf);
  z->logfd = openat(dfd, logfn, O_CREAT|O_WRONLY|O_TRUNC, 00644);
  debug_assert(z->logfd >= 0);

  unlinkat(dfd, "LOG", 0);
  symlinkat(logfn, dfd, "LOG");

  z->t0 = time_sec();

  rwlock_init(&(z->head_lock));
  char ts[64];
  time_stamp(ts, 64);
  logger_printf(z->logfd, "%s time %s v %lu seq %lu cache %lu\n", __func__, ts, msstz_version(z), seq0, cache_size_mb);

  for (u64 i = 0; i < hv->nr; i++) {
    const u64 magic = hv->es[i].anchor->priv;
    logger_printf(z->logfd, "%s [%3lu] %5lu\n", __func__, i, magic);
  }
  return z;
}

  inline u64
msstz_stat_writes(struct msstz * const z)
{
  return z->stat_writes;
}

  inline u64
msstz_stat_reads(struct msstz * const z)
{
  return z->stat_reads;
}

  inline u64
msstz_version(struct msstz * const z)
{
  return z->hv->version;
}

// free unused versions and delete unused files
// currently the first analyze worker do the gc
  static void
msstz_gc(struct msstz * const z)
{
  const u64 t0 = time_nsec();
  //const double t0 = time_sec();
  struct msstv * const hv = z->hv;
  debug_assert(hv);
  u64 nv = 0;
  // gc the tail, one at a time
  while (hv->next) {
    struct msstv ** pv = &(hv->next);
    // seek to &plast
    while ((*pv)->next)
      pv = &((*pv)->next);
    // stop if there is no next version, or the oldest version has version > 0

    struct msstv * const last = *pv;
    debug_assert(last->next == NULL);
    if (atomic_load_explicit(&last->rdrcnt, MO_CONSUME)) {
      break;
    } else { // do gc
      *pv = NULL; // remove from the list
      msstv_destroy_lazy(last);
      nv++;
    }
  }

  const u64 nc = z->rc ? rcache_close_flush(z->rc) : 0;

  // count nr
  u64 nr = 0;
  struct msstv * v = hv;
  while (v) {
    nr += v->nr;
    v = v->next;
  }
  // collect live seq numbers
  cpu_cfence();
  // array of all live seqs
  u64 * const vseq = malloc(sizeof(*vseq) * nr);
  // array of all live magics (live ssty)
  u64 * const vall = malloc(sizeof(*vall) * nr);
  u64 nr1 = 0;
  v = hv; // start over to collect seqs
  debug_assert(v);
  do {
    for (u64 i = 0; i < v->nr; i++) {
      const u64 magic = v->es[i].anchor->priv;
      vseq[nr1] = magic / 100;
      vall[nr1] = magic;
      nr1++;
    }
    v = v->next;
  } while (v);
  debug_assert(nr1 == nr);
  // it's ok to have duplicates
  qsort_u64(vseq, nr);
  qsort_u64(vall, nr);
  const u64 maxseq = vseq[nr-1];
  // search file in dir
  DIR * const dir = opendir(z->dirname); // don't directly use the dfd
  if (!dir) {
    logger_printf(z->logfd, "%s opendir() failed\n", __func__);
    exit(0);
  }

  u64 nu = 0;
  do {
    struct dirent * const ent = readdir(dir);
    if (!ent)
      break;
    char * dot = strchr(ent->d_name, '.');
    // has dot and is .sst*
    if (!dot)
      continue;

    if (!memcmp(dot, ".ver", 4)) {
      if (a2u64(ent->d_name) < hv->version)
        unlinkat(z->dfd, ent->d_name, 0);
      continue;
    }

    if (memcmp(dot, ".sst", 4))
      continue;
    const u64 magic = a2u64(ent->d_name);
    const u64 seq = magic / 100;

    if (seq > maxseq)
      continue;

    if (dot[4] == 'x') {
      if (bsearch_u64(seq, vseq, nr))
        continue;
    } else if (dot[4] == 'y') {
      if (bsearch_u64(magic, vall, nr))
        continue;
    } else {
      debug_die();
    }
    // now delete
    unlinkat(z->dfd, ent->d_name, 0);
    //logger_printf(z->logfd, "%s unlink %s\n", __func__, ent->d_name);
    nu++;
  } while (true);

  free(vseq);
  free(vall);
  closedir(dir);
  logger_printf(z->logfd, "%s gc dt-ms %lu free-v %lu close %lu unlink %lu\n", __func__, time_diff_nsec(t0)/1000000, nv, nc, nu);
}

  inline struct msstv *
msstz_getv(struct msstz * const z)
{
  rwlock_lock_read(&(z->head_lock));
  struct msstv * const v = z->hv;
  atomic_fetch_add_explicit(&(v->rdrcnt), 1, MO_ACQUIRE);
  rwlock_unlock_read(&(z->head_lock));
  return v;
}

  inline void
msstz_putv(struct msstz * const z, struct msstv * const v)
{
  (void)z;
  debug_assert(v->rdrcnt);
  atomic_fetch_sub_explicit(&(v->rdrcnt), 1, MO_RELEASE);
}

  void
msstz_destroy(struct msstz * const z)
{
  struct msstv * iter = z->hv;
  debug_assert(iter);
  msstz_gc(z);
  logger_printf(z->logfd, "%s hv %lu comp_time %lu writes %lu reads %lu\n", __func__,
      iter->version, z->stat_time, z->stat_writes, z->stat_reads);
  while (iter) {
    struct msstv * next = iter->next;
    msstv_destroy(iter);
    iter = next;
  }
  if (z->rc)
    rcache_destroy(z->rc);

  close(z->logfd);
  free(z->dirname);
  close(z->dfd);
  free(z);
}
// }}} msstz

// msstz-comp {{{

// struct {{{
// yq allows different threads to build sstys; very useful for sequential loading
struct msstz_yq {
  au64 pseq; // producer seq
  au64 cseq; // consumer seq
  spinlock lock;
  u32 padding;
  struct msstz_ytask {
    struct msst * y1; // output; the new mssty; NULL when not done
    struct msst * y0; // old mssty
    u64 seq1; // new seq // 0 if rejected
    u32 run1; // new run
    u32 run0; // how many tables to reuse from y0
    u64 ipart;
    u64 isub; // index of new partitions generated from an old partition
    const struct kv * anchor; // provide anchor key (isub == 0) or NULL (isub > 0)
  } tasks[0];
};

// global compaction information
struct msstz_comp_info {
  struct msstz * z;
  struct msstv * v0; // the old version
  au64 seqx; // to assign analysis/compaction tasks to threads
  struct msstz_yq * yq;
  u64 n0; // total number of v0 partitions
  au64 nx; // when nx == n0, the yq has all the tasks
  u32 nr_workers;
  u32 co_per_worker;
  const struct kvmap_api * api1; // memtable api
  void * map1; // memtable map
  au64 totsz;
  au64 stat_writes; // total bytes written to sstx&ssty
  au64 stat_reads; // total bytes read through rcache
  au64 stat_minor;
  au64 stat_partial;
  au64 stat_major;
  au64 stat_append;
  au64 time_analyze;
  au64 time_comp_x;
  au64 time_comp_y;
  u64 t0;
  u64 dta;
  u64 dtc;
  struct msstz_comp_part {
    u64 idx;
    u64 newsz; // size of new data in the memtable
    u32 bestrun; // how many existing (can be linked) tables to keep in the old partition
    float ratio; // write_size over read_size; newsz / totsz; the higher the better
  } parts[0];
};
// }}} struct

// y {{{
  static struct msstz_yq *
msstz_yq_create(const u64 nslots)
{
  struct msstz_yq * const yq = malloc(sizeof(*yq) + (sizeof(yq->tasks[0]) * nslots));
  yq->pseq = 0;
  yq->cseq = 0;
  spinlock_init(&yq->lock);
  return yq;
}

  static struct msstz_ytask *
msstz_yq_append(struct msstz_yq * const yq, struct msst * const mssty1, const u64 seq1, const u32 run1,
    struct msst * const mssty0, const u32 run0, const u64 ipart, const u64 isub, const struct kv * const anchor)
{
  spinlock_lock(&yq->lock);
  const u64 i = yq->pseq++;
  struct msstz_ytask * const task = &yq->tasks[i];
  task->y1 = mssty1; // not NULL only when append
  task->y0 = mssty0;
  task->seq1 = seq1;
  task->run1 = run1;
  task->run0 = run0;
  task->ipart = ipart;
  task->isub = isub;
  task->anchor = anchor;
  spinlock_unlock(&yq->lock);
  return &yq->tasks[i];
}

// return true when a task is found and executed
  static bool
msstz_yq_consume(struct msstz_comp_info * const ci)
{
  struct msstz_yq * const yq = ci->yq;
  struct msstz * const z = ci->z;
  spinlock_lock(&yq->lock);
  if (yq->cseq == yq->pseq) {
    // no task
    spinlock_unlock(&yq->lock);
    return false;
  }
  // claim a task with lock
  const u64 id = yq->cseq++;
  spinlock_unlock(&yq->lock);

  struct msstz_ytask * const task = &(yq->tasks[id]);
  // already done (only once for the store-append)
  if (task->y1)
    return true;

  // open a msstx and call ssty_build
  //const u64 t0 = time_nsec();
  struct msst * const msst = msstx_open_at_reuse(z->dfd, task->seq1, task->run1, task->y0, task->run0);
  msst_rcache(msst, z->rc);
  const u32 ysz = ssty_build_at(z->dfd, msst, task->seq1, task->run1, task->y0, task->run0, z->cfg.tags);
  if (!ysz)
    debug_die();
  ci->stat_writes += ysz;

  // convert msstx to mssty
  const bool ry = mssty_open_y_at(z->dfd, msst);
  if (!ry)
    debug_die();
  task->y1 = msst; // done; the new partition is now loaded and ready to use

  //const u64 dt = time_diff_nsec(t0);
  //const struct ssty_meta * const ym = msst->ssty->meta;
  //logger_printf(z->logfd, "%s dt-ms %lu ssty-build %lu %02u nkidx %u ysz %u xkv %u xsz %u uniq %u valid %u\n",
  //    __func__, dt / 1000000, task->seq1, task->run1, msst->ssty->nkidx,
  //    ysz, ym->totkv, ym->totsz, ym->uniqx[0], ym->valid);
  return true;
}
// }}} y

// x {{{
// compaction driver on one partition; it may create multiple partitions
// create ssts synchronously; queue build-ssty tasks in yq
// seq0 and run0 indicate the existing (can be linked) tables in the target partition
  static void
msstz_comp_ssts(struct msstz_comp_info * const ci, const u64 ipart, struct miter * const miter,
    const struct kv * const k0, const struct kv * const kz, const u64 seq0, const u32 run0, const bool split,
    struct msst * const mssty0, const bool is_append)
{
  struct msstz * const z = ci->z;
  // tmp anchor
  struct kv * const tmp = malloc(sizeof(struct kv) + SST_MAX_KVSZ); // just a buffer
  u64 seq = seq0;
  u32 run = run0;
  u32 np = 0;
  debug_assert(run < MSST_NRUN);

  if (is_append) {
    msstz_yq_append(ci->yq, mssty0, seq, run, NULL, 0, ipart, 0, k0); // only mssty0, ipart, and k0 will be used
    np++;
    seq = z->seq++;
    run = 0;
  }
  // a compaction may create new partitions, each with several new tables
  do {
    //const u64 t0 = time_nsec();

    const struct sst_build_cfg cfg = {.seq = seq, .run = run,
      .max_pages = z->cfg.max_pages, .tgt_pages = z->cfg.tgt_pages, .del = split, .ckeys = z->cfg.ckeys};
    const u64 sizex = sst_build_at(z->dfd, miter, &cfg, NULL, kz);
    //const u64 dt = time_diff_nsec(t0);
    ci->stat_writes += sizex;
    //logger_printf(z->logfd, "%s dt-ms %lu sst-build %lu-%02u %lu\n", __func__, dt / 1000000, seq, run, sizex);
    run++;

    struct kv * const tmpz = miter_peek(miter, tmp); // the current unconsumed key

    // the entire partition is done
    const bool donez = (tmpz == NULL) || (kz && (kv_compare(tmpz, kz) >= 0));

    // the current partition must close; will switch to a new partition
    const bool done1 = split ? (run >= z->cfg.major_switch) : false;
    if (donez || done1) { // close current mssty
      if (np == 0) { // on the original partition; use y0, run0, and k0
        msstz_yq_append(ci->yq, NULL, seq, run, mssty0, run0, ipart, np, k0);
      } else { // a new partition: reuse nothing, generate anchor
        msstz_yq_append(ci->yq, NULL, seq, run, NULL, 0, ipart, np, NULL);
      }
      np++;

      if (donez) { // the end
        break;
      } else if (split) { // done1: next partition
        seq = z->seq++;
        run = 0;
      } else if (run >= MSST_NRUN) {
        // it is acceptable to have tables above major_trigger; the actual threshold is MSST_NRUN
        debug_die();
      }
    }
  } while (true);
  debug_assert(split || (np == 1)); // only split can return more than one partition
  free(tmp);
  //logger_printf(z->logfd, "%s np %u seq0 %lu run0 %u seq %lu run %u\n", __func__, np, seq0, run0, seq, run);
}

  static void
msstz_comp_link(const int dfd, const u64 seq0, const u64 seq1, const u32 nrun)
{
  debug_assert(seq0 < seq1);
  char fn0[24];
  char fn1[24];

  for (u32 i = 0; i < nrun; i++) {
    sprintf(fn0, "%03lu.sstx", seq0 * 100lu + i);
    sprintf(fn1, "%03lu.sstx", seq1 * 100lu + i);
    unlinkat(dfd, fn1, 0);
    linkat(dfd, fn0, dfd, fn1, 0);
  }
}
// }}} x

// v {{{
  static int
msstz_cmp_ytask(const void * p1, const void * p2)
{
  const struct msstz_ytask * const t1 = p1;
  const struct msstz_ytask * const t2 = p2;
  if (t1->ipart < t2->ipart) {
    return -1;
  } else if (t1->ipart > t2->ipart) {
    return 1;
  } else if (t1->isub < t2->isub) {
    return -1;
  } else if (t1->isub > t2->isub) {
    return 1;
  } else {
    debug_die();
    return 0;
  }
}

  static void
msstz_comp_harvest(struct msstz_comp_info * const ci)
{
  struct msstz_yq * const yq = ci->yq;
  debug_assert(yq->pseq == yq->cseq);
  debug_assert(yq->pseq >= ci->n0);
  const u64 nr = yq->pseq;
  struct msstv * const v0 = ci->v0;

  // sort yq
  qsort(yq->tasks, nr, sizeof(yq->tasks[0]), msstz_cmp_ytask);

  u64 ssty_sz = 0; // total ssty file size
  u64 meta_sz = 0; // total sst metadata size
  u64 data_sz = 0; // total data size (pages)
  struct msstv * const v1 = msstv_create(nr, ci->v0->version + 1); // no resizing

  // collect new partitions and create v1
  for (u64 i = 0; i < nr; i++) {
    struct msstz_ytask * const t = &yq->tasks[i];
    msstv_append(v1, t->y1, t->anchor);
    v0->es[t->ipart].anchor->vlen = (t->seq1 == UINT64_MAX) ? 1 : 0; // 1: rej; 0: ok;

    // count sizes
    struct msst * const msst = t->y1;
    struct ssty * const ssty = msst->ssty;
    ssty_sz += ssty->size;
    for (u32 j = 0; j < ssty->nrun; j++) {
      struct sst * const sst = &(msst->ssts[j]);
      data_sz += (PGSZ * sst->npages);
      const u32 msz = sst->fsize - (PGSZ * sst->npages);
      debug_assert(msz <= sst->fsize);
      meta_sz += msz;
    }
  }

  struct msstz * const z = ci->z;
  logger_printf(z->logfd, "%s v %lu nr %lu ssty-size %lu sst-meta-size %lu sst-data-size %lu\n",
      __func__, v1->version, nr, ssty_sz, meta_sz, data_sz);

  v1->rc = z->rc;
  v1->next = v0;
  // finalize the new version v1
  msstv_save(v1, z->dfd);
  msstz_head_sync(z->dfd, v1->version);

  // add the new version to z
  rwlock_lock_write(&(z->head_lock));
  z->hv = v1;
  rwlock_unlock_write(&(z->head_lock));
}
// }}} v

// analyze {{{
  static u64
msstz_comp_estimate_ssty(const u64 nkeys, const float run)
{
  const u64 nsecs = nkeys / SSTY_DIST;
  return (sizeof(struct sst_ptr) * (u64)ceilf(run) + 16) * nsecs + nkeys;
}

  static inline const struct kv *
msstz_comp_get_kz(struct msstv * const v, const u64 ipart)
{
  return ((ipart + 1) < v->nr) ? v->es[ipart + 1].anchor : NULL;
}

  static inline u64
msstz_comp_runsz(struct msstz * const z)
{
  return (PGSZ - 256) * z->cfg.max_pages;
}

  static inline u32
msstz_comp_bestrun(struct msstz_comp_part * const cpart, struct msst * const msst,
    struct msstz * const z, const u64 newnr)
{
  // calculate wa[i] if start from run[i]
  // 0: fully rewrite with real deletion
  // 1 to nrun-1: partial merge
  // nrun: minor compaction (wa == 1)

  const struct ssty_meta * const meta = msst->ssty->meta;
  const u32 nrun = msst->nrun;
  debug_assert(meta->uniqx[nrun] == 0);

  struct { u64 wx, wy; float nrun1, wa, bonus, score; } f[MSST_NRUN+1];

  // wx: data write size
  f[0].wx = cpart->newsz + ((u64)meta->valid * meta->totsz / meta->totkv); // major
  for (u32 i = 1; i <= nrun; i++)
    f[i].wx = cpart->newsz + ((u64)meta->uniqx[i] * meta->totsz / meta->totkv); // major

  // nrun1: final run
  // penalty = nrun1
  const float runsz = (float)msstz_comp_runsz(z);
  for (u32 i = 0; i <= nrun; i++)
    f[i].nrun1 = ((float)f[i].wx / runsz) + (float)i;

  // wy: ysize
  f[0].wy = msstz_comp_estimate_ssty(newnr + meta->valid, fminf(f[0].nrun1, (float)z->cfg.major_trigger));
  u64 totkvi = 0;
  for (u32 i = 1; i <= nrun; i++) {
    totkvi += msst->ssts[i-1].totkv;
    f[i].wy = msstz_comp_estimate_ssty(newnr + totkvi + meta->uniqx[i], f[i].nrun1);
  }
  // wa: write amp.
  for (u32 i = 0; i <= nrun; i++)
    f[i].wa = (float)(f[i].wx + f[i].wy) / (float)cpart->newsz;

  // bonus: based on how effective can it reduce runs
  for (u32 i = 0; i <= nrun; i++)
    f[i].bonus = f[nrun].nrun1 - f[i].nrun1;

  // adjust major bonus: large bonus when split is necessary
  if (f[nrun].nrun1 > (float)z->cfg.major_trigger)
    f[0].bonus += (f[0].nrun1 - (float)z->cfg.major_switch);

  // adjust minor bonus; +1
  f[nrun].bonus += 1.0f;

  u32 bestrun = 0; // major by default
  for (u32 i = 0; i <= nrun; i++) {
    // score: lower is better
    //const float score = (f[i].wa + sqrtf(f[i].nrun1 + 1.0f)) / (f[i].bonus + 4.0f);
    const float score = (f[i].wa + (f[i].nrun1 * 0.75f)) / (f[i].bonus + 4.0f);
    f[i].score = score;
    if ((i < z->cfg.major_trigger) && (f[i].nrun1 < (float)z->cfg.estimate_safe) && (f[i].score < f[bestrun].score))
      bestrun = i;
  }
  debug_assert(bestrun < z->cfg.major_trigger);

  // log some details of the compaction
  //for (u32 i = 0; i <= nrun; i++) {
  //  const u64 sz = (i < nrun) ? (msst->ssts[i].npages * PGSZ) : newsz;
  //  const float pct = ((float)sz) * 100.0f / (float)(z->npages * PGSZ);
  //  logger_printf(z->logfd, "%c[%c%x] sz %9lu %6.2f%% wx %6lu wy %6lu nrun1 %4.1f wa %4.1f bonus %4.1f score %5.2f\n",
  //      (i == bestrun ? '>':' '), (i == nrun ? '*' : ' '), i,
  //      sz, pct, f[i].wx, f[i].wy, f[i].nrun1, f[i].wa, f[i].bonus, f[i].score);
  //}

  logger_printf(z->logfd, "%s magic0 %lu totkv0 %u valid0 %u newnr %lu newsz %lu minor %.1f major %.1f bestrun %u ratio %.3f\n",
      __func__, meta->magic, meta->totkv, meta->valid,
      newnr, cpart->newsz, f[nrun].nrun1, f[0].nrun1, bestrun, cpart->ratio);
  return bestrun;
}

// bestrun:
// 0: major, rewrite everything
// < nrun0 (nrun0 < MSST_NRUN): partial, rewrite a few tables
// == nrun0: minor, no rewritting
// == MSST_NRUN: store-append: the last partition and new data > existing keys
  static u64
msstz_comp_analyze(struct msstz_comp_info * const ci, const u64 ipart)
{
  const u64 t0 = time_nsec();
  struct msstv_part * const part = &(ci->v0->es[ipart]);
  struct msst * const msst = part->msst;
  // k0 kz
  const struct kv * const k0 = part->anchor;
  debug_assert(k0);
  // kz == NULL for the last partition
  const struct kv * const kz = msstz_comp_get_kz(ci->v0, ipart);
  const struct kvmap_api * const api = ci->api1;
  void * const map = ci->map1;

  void * const ref = kvmap_ref(api, map);
  void * const iter = api->iter_create(ref);
  u64 newsz = 0;
  u64 newnr = 0;
  struct kv * kz_inp = NULL;
  if (kz) { // not the last partition; search where to stop
    struct kref krefz = {.ptr = kz->kv, .len = kz->klen, .hash32 = kv_crc32c(kz->kv, kz->klen)};
    api->iter_seek(iter, &krefz);
    api->iter_inp(iter, kvmap_inp_steal_kv, &kz_inp);
  }

  // check if new key has no overlap
  struct kv * const sst_kz = mssty_last(msst, NULL); // free soon
  struct kv * map_k0 = NULL;

  // start from k0
  struct kref kref0 = {.ptr = k0->kv, .len = k0->klen, .hash32 = kv_crc32c(k0->kv, k0->klen)};
  api->iter_seek(iter, &kref0);
  api->iter_inp(iter, kvmap_inp_steal_kv, &map_k0);
  const bool overlap = sst_kz && map_k0 && (kv_compare(sst_kz, map_k0) >= 0);
  free(sst_kz);

  const u32 sample_skip = 8;
  const u32 sample_mask = sample_skip-1;
  while (api->iter_valid(iter)) {
    struct kv * kv_inp = NULL;
    api->iter_inp(iter, kvmap_inp_steal_kv, &kv_inp);
    if (kv_inp == kz_inp)
      break;
    if ((newnr & sample_mask) == 0)
      newsz += (sst_kv_vi128_estimate(kv_inp) + sizeof(u16));

    newnr++;
    api->iter_skip1(iter);
  }
  newsz *= sample_skip;
  api->iter_destroy(iter);
  kvmap_unref(api, ref);

  struct msstz_comp_part * const cpart = &(ci->parts[ipart]);
  cpart->newsz = newsz;

  const struct ssty_meta * const meta = msst->ssty->meta;
  const u32 nrun = msst->nrun;

  struct msstz * const z = ci->z;
  // MAJOR: no existing data at all
  if (meta->valid == 0) { // this also avoids divide-by-zero below
    cpart->ratio = (float)newsz;
    cpart->bestrun = 0;
    logger_printf(z->logfd, "%s newsz %lu direct-major\n", __func__, newsz);
    return time_diff_nsec(t0);
  }

  // REJECT empty input
  if (newnr == 0) {
    cpart->ratio = 0.0f;
    cpart->bestrun = z->cfg.major_trigger; // reject
    return time_diff_nsec(t0);
  }

  const u64 runsz = msstz_comp_runsz(z);
  // APPEND: not too small, store-wide append, partition has some data
  if (!overlap && !kz && nrun > 1 && newsz > runsz) {
    cpart->ratio = (float)newsz; // worth doing
    cpart->bestrun = MSST_NRUN;
    logger_printf(z->logfd, "%s newsz %lu store-append\n", __func__, newsz);
    return time_diff_nsec(t0);
  }

  debug_assert(meta->totsz && newsz);
  cpart->ratio = (float)newsz / (float)(meta->totsz);
  cpart->bestrun = msstz_comp_bestrun(cpart, msst, z, newnr); // bestrun is determined

  return time_diff_nsec(t0);
}

  static void *
msstz_analyze_worker(void * const ptr)
{
  struct msstz_comp_info * const ci = (typeof(ci))ptr;
  const u64 n = ci->v0->nr;
  do {
    const u64 i = ci->seqx++;
    if (i == 0)
      msstz_gc(ci->z);
    if (i >= n)
      return NULL;

    ci->parts[i].idx = i; // assign idx
    ci->time_analyze += msstz_comp_analyze(ci, i);
    ci->totsz += ci->parts[i].newsz;
  } while (true);
}
// }}} analyze

// part {{{
// do compaction in a partition; bestrun decides what to do
  static u64
msstz_comp_part(struct msstz_comp_info * const ci, const u64 ipart)
{
  const u64 t0 = time_nsec();
  struct msstz * const z = ci->z;
  struct msstv_part * const part = &(ci->v0->es[ipart]);
  struct msst * const mssty0 = part->msst;
  struct msstz_comp_part * const cpart = &(ci->parts[ipart]);
  const struct kv * const k0 = part->anchor;
  debug_assert(k0);

  const u64 magic0 = part->anchor->priv;
  const u64 seq0 = magic0 / 100lu; // seq of the old partition
  const u32 nrun0 = mssty0->nrun; // seq of the old partition

  if (cpart->bestrun == z->cfg.major_trigger) {
    // marked as rejected by msstz_comp()
    // reject: send to yqueue as completed; use seq = UINT64_MAX for real rejections or seq0 for newsz == 0
    msstz_yq_append(ci->yq, mssty0, cpart->newsz ? UINT64_MAX : seq0, nrun0, NULL, 0, ipart, 0, k0);
    // {y0, seq, ipart, k0} will be used later
    ci->nx++;
    return 0;
  }

  const u32 bestrun = cpart->bestrun;
  const bool is_append = (bestrun == MSST_NRUN);
  const bool is_minor = (bestrun == nrun0);
  const bool is_major = (bestrun == 0);

  debug_assert(bestrun <= z->cfg.major_trigger || is_append);
  // k0 kz
  // kz == NULL for the last partition
  const struct kv * const kz = msstz_comp_get_kz(ci->v0, ipart);

  // need a new seq unless it's a minor compaction
  // start with a different seq unless it's a minor compaction
  const u64 seq1 = (is_minor || is_append) ? seq0 : (z->seq++);

  struct miter * const miter = miter_create();
  if (bestrun < nrun0) { // major or partial
    debug_assert(seq1 != seq0);
    // hard link unchanged files for major and partial
    msstz_comp_link(z->dfd, seq0, seq1, bestrun);

    if (bestrun) { // partial
      for (u32 w = bestrun; w < nrun0; w++)
        miter_add(miter, &kvmap_api_sst, &(mssty0->ssts[w]));
      ci->stat_partial++;
    } else { // major
      miter_add(miter, &kvmap_api_mssty, mssty0);
      ci->stat_major++;
    }
  } else if (bestrun == nrun0) { // minor: add nothing
    debug_assert(seq1 == seq0);
    debug_assert(is_minor);
    ci->stat_minor++;
  } else { // append
    debug_assert(seq1 == seq0);
    debug_assert(is_append);
    ci->stat_append++;
  }

  // add the memtable
  miter_add(miter, ci->api1, ci->map1);
  struct kref kref0 = {.ptr = k0->kv, .len = k0->klen, .hash32 = kv_crc32c(k0->kv, k0->klen)};
  miter_seek(miter, &kref0);

  const u32 comprun = is_append ? nrun0 : bestrun;
  // allow split (and gc tombstones) when: major or append
  const bool split = is_major || is_append;
  msstz_comp_ssts(ci, ipart, miter, k0, kz, seq1, comprun, split, mssty0, is_append);
  miter_destroy(miter);
  ci->nx++; // done with one partition's x
  return time_diff_nsec(t0);
}
// }}} part

// driver {{{
  static void
msstz_comp_worker_func(struct msstz_comp_info * const ci)
{
  const u64 n = ci->v0->nr;
  struct coq * const coq = coq_current();
  // x loop
  do {
    const u64 i = ci->seqx++;
    if (i >= n)
      break;

    ci->time_comp_x += msstz_comp_part(ci, i);

    const u64 t0 = time_nsec();
    if (msstz_yq_consume(ci))
      ci->time_comp_y += time_diff_nsec(t0);
  } while (true);

  // process all ssty_build
  struct msstz_yq * const yq = ci->yq;
  while ((ci->nx < ci->n0) || (yq->cseq < yq->pseq)) {
    const u64 t0 = time_nsec();
    const bool r = msstz_yq_consume(ci);
    if (r) {
      ci->time_comp_y += time_diff_nsec(t0);
    } else if (coq) {
      coq_idle(coq);
    } else {
      usleep(1);
    }
  }
}

  static void
msstz_comp_worker_coq_co(void)
{
  void * const priv = co_priv();
  debug_assert(priv);
  msstz_comp_worker_func(priv);
}

// thread
  static void *
msstz_comp_worker(void * const ptr)
{
  ssty_build_ckeys_reads = 0;
  rcache_thread_stat_reset();
  struct msstz_comp_info * const ci = (typeof(ci))ptr;
  const u32 nco = ci->co_per_worker;
  if (nco > 1) {
    struct coq * const coq = coq_create_auto(nco << 1);
    coq_install(coq);
    u64 hostrsp = 0;
    for (u64 i = 0; i < nco; i++) {
      struct co * const co = co_create(PGSZ * 7, msstz_comp_worker_coq_co, ptr, &hostrsp);
      corq_enqueue(coq, co);
    }
    coq_run(coq);
    coq_uninstall();
    coq_destroy_auto(coq);
  } else {
    msstz_comp_worker_func(ci);
  }
  ci->stat_reads += ssty_build_ckeys_reads;
  ci->stat_reads += (rcache_thread_stat_reads() * PGSZ);
  return NULL;
}

// for sorting tasks based on their sizes
  static int
msstz_cmp_ratio(const void * p1, const void * p2)
{
  const float r1 = ((const struct msstz_comp_part *)p1)->ratio;
  const float r2 = ((const struct msstz_comp_part *)p2)->ratio;
  debug_assert(isfinite(r1));
  debug_assert(isfinite(r2));
  if (r1 < r2) {
    return -1;
  } else if (r1 > r2) {
    return 1;
  } else {
    return 0;
  }
}

  static int
msstz_cmp_idx(const void * p1, const void * p2)
{
  const u64 * const v1 = &(((const struct msstz_comp_part *)p1)->idx);
  const u64 * const v2 = &(((const struct msstz_comp_part *)p2)->idx);
  return compare_u64(v1, v2);
}

  static u64
msstz_comp_reject(struct msstz_comp_info * const ci, const u64 max_reject)
{
  const u64 nr = ci->v0->nr;
  qsort(ci->parts, nr, sizeof(ci->parts[0]), msstz_cmp_ratio);

  // reject keys
  u64 rejsz = 0;
  u64 nrej = 0;
  struct msstz * const z = ci->z;
  const u64 size_accept = PGSZ * z->cfg.pages_accept;
  logger_printf(z->logfd, "%s ratio min %.3f max %.3f\n", __func__, ci->parts[0].ratio, ci->parts[nr-1].ratio);
  for (u64 i = 0; i < nr; i++) {
    struct msstz_comp_part * const cp = &(ci->parts[i]);
    // no more rejections
    if ((cp->newsz > size_accept) || ((rejsz + cp->newsz) > max_reject) || cp->ratio > 0.1f) {
      logger_printf(z->logfd, "%s i %lu/%lu (newsz %lu > %lu) || ((rejsz %lu + newsz %lu) > max_reject %lu) || (ratio %.3f > 0.1)\n",
          __func__, i, nr, cp->newsz, size_accept, rejsz, cp->newsz, max_reject, cp->ratio);
      break;
    }
    rejsz += cp->newsz;
    nrej++;
    cp->bestrun = z->cfg.major_trigger; // reject
  }
  logger_printf(z->logfd, "%s reject size %lu/%lu np %lu/%lu\n", __func__, rejsz, ci->totsz, nrej, nr);

  // resume idx order
  qsort(ci->parts, nr, sizeof(ci->parts[0]), msstz_cmp_idx);
  return nrej;
}

  static void
msstz_comp_stat(struct msstz_comp_info * const ci)
{
  struct msstz * const z = ci->z;
  const u64 dt = time_diff_nsec(ci->t0);
  const u64 dw = ci->stat_writes;
  const u64 dr = ci->stat_reads;
  z->stat_time += dt;
  z->stat_writes += dw;
  z->stat_reads += dr;
  const u64 ta = ci->time_analyze;
  const u64 tx = ci->time_comp_x;
  const u64 ty = ci->time_comp_y;
  logger_printf(z->logfd, "%s dt-s %.6lf dw-mb %lu mbps %lu dr-mb %lu mbps %lu append %lu major %lu partial %lu minor %lu\n",
      __func__, ((double)dt) * 1e-9, dw>>20, (dw<<10)/dt, dr>>20, (dr<<10)/dt,
      ci->stat_append, ci->stat_major, ci->stat_partial, ci->stat_minor);
  logger_printf(z->logfd, "%s  dta %lu/%lu %3lu%% dtc (%lu+%lu)/%lu %3lu%%\n",
      __func__, ta/1000000, ci->dta/1000000, ta*100lu/ci->dta,
      tx/1000000, ty/1000000, ci->dtc/1000000, (tx+ty)*100lu/ci->dtc);
}

// comp is not thread safe
// p_min_write: 0 to 100, minimum percentage of data that must be written down
  void
msstz_comp(struct msstz * const z, const struct kvmap_api * const api1, void * const map1,
    const u32 nr_workers, const u32 co_per_worker, const u64 max_reject)
{
  struct msstv * const v0 = msstz_getv(z);
  const u64 nr = v0->nr;
  struct msstz_comp_info * const ci = calloc(1, sizeof(*ci) + (nr * sizeof(ci->parts[0])));
  ci->t0 = time_nsec();
  ci->z = z;
  ci->v0 = v0;
  ci->n0 = nr;
  ci->api1 = api1;
  ci->map1 = map1;
  ci->nr_workers = nr_workers;
  ci->co_per_worker = co_per_worker;

  // concurrent analysis + GC by seq==0
  ci->dta = thread_fork_join(nr_workers, msstz_analyze_worker, false, ci);
  const u64 nrej = msstz_comp_reject(ci, max_reject);
  if (nrej < nr) {
    ci->seqx = 0; // restart from 0
    ci->yq = msstz_yq_create((nr + 64) << 3); // large enough for adding new partitions
    ci->dtc = thread_fork_join(nr_workers, msstz_comp_worker, false, ci);
    msstz_comp_harvest(ci);
    free(ci->yq);
  } else {
    ci->dtc = 1; // avoid divide by zero
  }

  msstz_putv(z, v0);
  msstz_comp_stat(ci);
  free(ci);
}
// }}} driver

// }}} msstz-comp

// api {{{
//
// *sst* TOMESTONES
// regular functions: iter/get/probe
//   tompstones are treated as regular keys
//   they are always visible with iterators
//   this behavior is required by a few internal functions
//
// _ts functions:
//   tomestones are not visible to caller
const struct kvmap_api kvmap_api_sst = {
  .ordered = true,
  .readonly = true,
  .unique = true,
  .get = (void *)sst_get,
  .probe = (void *)sst_probe,
  .iter_create = (void *)sst_iter_create,
  .iter_seek = (void *)sst_iter_seek,
  .iter_valid = (void *)sst_iter_valid,
  .iter_peek = (void *)sst_iter_peek,
  .iter_kref = (void *)sst_iter_kref,
  .iter_kvref = (void *)sst_iter_kvref,
  .iter_retain = (void *)sst_iter_retain,
  .iter_release = (void *)sst_iter_release,
  .iter_skip1 = (void *)sst_iter_skip1,
  .iter_skip = (void *)sst_iter_skip,
  .iter_next = (void *)sst_iter_next,
  .iter_park = (void *)sst_iter_park,
  .iter_destroy = (void *)sst_iter_destroy,
  .destroy = (void *)sst_destroy,
  .fprint = (void *)sst_fprint,
};

const struct kvmap_api kvmap_api_msstx = {
  .ordered = true,
  .readonly = true,
  .unique = false,
  .get = (void *)msstx_get,
  .probe = (void *)msstx_probe,
  .iter_create = (void *)msstx_iter_create,
  .iter_seek = (void *)msstx_iter_seek,
  .iter_valid = (void *)msstx_iter_valid,
  .iter_peek = (void *)msstx_iter_peek,
  .iter_kref = (void *)msstx_iter_kref,
  .iter_kvref = (void *)msstx_iter_kvref,
  .iter_retain = (void *)msstx_iter_retain,
  .iter_release = (void *)msstx_iter_release,
  .iter_skip1 = (void *)msstx_iter_skip1,
  .iter_skip = (void *)msstx_iter_skip,
  .iter_next = (void *)msstx_iter_next,
  .iter_park = (void *)msstx_iter_park,
  .iter_destroy = (void *)msstx_iter_destroy,
  .destroy = (void *)msstx_destroy,
};

const struct kvmap_api kvmap_api_mssty = {
  .ordered = true,
  .readonly = true,
  .unique = true,
  .get = (void *)mssty_get,
  .probe = (void *)mssty_probe,
  .iter_create = (void *)mssty_iter_create,
  .iter_seek = (void *)mssty_iter_seek,
  .iter_valid = (void *)mssty_iter_valid,
  .iter_peek = (void *)mssty_iter_peek,
  .iter_kref = (void *)mssty_iter_kref,
  .iter_kvref = (void *)mssty_iter_kvref,
  .iter_retain = (void *)mssty_iter_retain,
  .iter_release = (void *)mssty_iter_release,
  .iter_skip1 = (void *)mssty_iter_skip1,
  .iter_skip = (void *)mssty_iter_skip,
  .iter_next = (void *)mssty_iter_next,
  .iter_park = (void *)mssty_iter_park,
  .iter_destroy = (void *)mssty_iter_destroy,
  .ref = (void *)mssty_ref,
  .unref = (void *)mssty_unref,
  .destroy = (void *)mssty_destroy,
  .fprint = (void *)mssty_fprint,
};

const struct kvmap_api kvmap_api_mssty_ts = {
  .ordered = true,
  .readonly = true,
  .unique = true,
  .get = (void *)mssty_get_ts,
  .probe = (void *)mssty_probe_ts,
  .iter_create = (void *)mssty_iter_create,
  .iter_seek = (void *)mssty_iter_seek_ts,
  .iter_valid = (void *)mssty_iter_valid,
  .iter_peek = (void *)mssty_iter_peek,
  .iter_kref = (void *)mssty_iter_kref,
  .iter_kvref = (void *)mssty_iter_kvref,
  .iter_retain = (void *)mssty_iter_retain,
  .iter_release = (void *)mssty_iter_release,
  .iter_skip1 = (void *)mssty_iter_skip1_ts,
  .iter_skip = (void *)mssty_iter_skip_ts,
  .iter_next = (void *)mssty_iter_next_ts,
  .iter_park = (void *)mssty_iter_park,
  .iter_destroy = (void *)mssty_iter_destroy,
  .ref = (void *)mssty_ref,
  .unref = (void *)mssty_unref,
  .destroy = (void *)mssty_destroy,
  .fprint = (void *)mssty_fprint,
};

const struct kvmap_api kvmap_api_mssty_dup = {
  .ordered = true,
  .readonly = true,
  .get = (void *)mssty_get,
  .probe = (void *)mssty_probe,
  .iter_create = (void *)mssty_iter_create,
  .iter_seek = (void *)mssty_iter_seek,
  .iter_valid = (void *)mssty_iter_valid,
  .iter_peek = (void *)mssty_iter_peek_dup,
  .iter_kref = (void *)mssty_iter_kref_dup,
  .iter_kvref = (void *)mssty_iter_kvref_dup,
  .iter_retain = (void *)mssty_iter_retain,
  .iter_release = (void *)mssty_iter_release,
  .iter_skip1 = (void *)mssty_iter_skip1_dup,
  .iter_skip = (void *)mssty_iter_skip_dup,
  .iter_next = (void *)mssty_iter_next_dup,
  .iter_park = (void *)mssty_iter_park,
  .iter_destroy = (void *)mssty_iter_destroy,
  .ref = (void *)mssty_ref,
  .unref = (void *)mssty_unref,
  .destroy = (void *)mssty_destroy,
  .fprint = (void *)mssty_fprint,
};
const struct kvmap_api kvmap_api_msstv = {
  .ordered = true,
  .readonly = true,
  .unique = true,
  .get = (void *)msstv_get,
  .probe = (void *)msstv_probe,
  .iter_create = (void *)msstv_iter_create,
  .iter_seek = (void *)msstv_iter_seek,
  .iter_valid = (void *)msstv_iter_valid,
  .iter_peek = (void *)msstv_iter_peek,
  .iter_kref = (void *)msstv_iter_kref,
  .iter_kvref = (void *)msstv_iter_kvref,
  .iter_retain = (void *)msstv_iter_retain,
  .iter_release = (void *)msstv_iter_release,
  .iter_skip1 = (void *)msstv_iter_skip1,
  .iter_skip = (void *)msstv_iter_skip,
  .iter_next = (void *)msstv_iter_next,
  .iter_park = (void *)msstv_iter_park,
  .iter_destroy = (void *)msstv_iter_destroy,
  .ref = (void *)msstv_ref,
  .unref = (void *)msstv_unref,
  .destroy = (void *)msstv_destroy,
  .fprint = (void *)msstv_fprint,
};

const struct kvmap_api kvmap_api_msstv_ts = { // xdb
  .ordered = true,
  .readonly = true,
  .unique = true,
  .get = (void *)msstv_get_ts,
  .probe = (void *)msstv_probe_ts,
  .iter_create = (void *)msstv_iter_create,
  .iter_seek = (void *)msstv_iter_seek_ts,
  .iter_valid = (void *)msstv_iter_valid,
  .iter_peek = (void *)msstv_iter_peek,
  .iter_kref = (void *)msstv_iter_kref,
  .iter_kvref = (void *)msstv_iter_kvref,
  .iter_retain = (void *)msstv_iter_retain,
  .iter_release = (void *)msstv_iter_release,
  .iter_skip1 = (void *)msstv_iter_skip1_ts,
  .iter_skip = (void *)msstv_iter_skip_ts,
  .iter_next = (void *)msstv_iter_next_ts,
  .iter_park = (void *)msstv_iter_park,
  .iter_destroy = (void *)msstv_iter_destroy,
  .ref = (void *)msstv_ref,
  .unref = (void *)msstv_unref,
  .destroy = (void *)msstv_destroy,
  .fprint = (void *)msstv_fprint,
};

  static void *
sst_kvmap_api_create(const char * const name, const struct kvmap_mm * const mm, char ** args)
{
  (void)mm;
  if (!strcmp(name, "sst")) {
    return sst_open(args[0], a2u64(args[1]), a2u32(args[2]));
  } else if (!strcmp(name, "msstx")) {
    return msstx_open(args[0], a2u64(args[1]), a2u32(args[2]));
  } else if ((!strcmp(name, "mssty")) || (!strcmp(name, "mssty_ts")) || (!strcmp(name, "mssty_dup"))) {
    return mssty_open(args[0], a2u64(args[1]), a2u32(args[2]));
  } else if ((!strcmp(name, "msstv")) || (!strcmp(name, "msstv_ts"))) {
    return msstv_open(args[0], args[1]);
  } else {
    return NULL;
  }
}

// alternatively, call the register function from main()
__attribute__((constructor))
  static void
sst_kvmap_api_init(void)
{
  kvmap_api_register(3, "sst", "<dirname> <seq> <run>", sst_kvmap_api_create, &kvmap_api_sst);
  kvmap_api_register(3, "msstx", "<dirname> <seq> <nrun>", sst_kvmap_api_create, &kvmap_api_msstx);
  kvmap_api_register(3, "mssty", "<dirname> <seq> <nrun>", sst_kvmap_api_create, &kvmap_api_mssty);
  kvmap_api_register(3, "mssty_ts", "<dirname> <seq> <nrun>", sst_kvmap_api_create, &kvmap_api_mssty_ts);
  kvmap_api_register(3, "mssty_dup", "<dirname> <seq> <nrun>", sst_kvmap_api_create, &kvmap_api_mssty_dup);
  kvmap_api_register(2, "msstv", "<dirname> <filename>", sst_kvmap_api_create, &kvmap_api_msstv);
  kvmap_api_register(2, "msstv_ts", "<dirname> <filename>", sst_kvmap_api_create, &kvmap_api_msstv_ts);
}
// }}} api

// vim:fdm=marker