msstztest.c

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

#include "lib.h"
#include "kv.h"
#include "wh.h"
#include "sst.h"

  static bool
memtable_del(const struct kvmap_api * const api, void * const ref, struct kv * const key)
{
  const u32 vlen0 = key->vlen;
  key->vlen = SST_VLEN_TS;
  const bool r = api->put(ref, key);
  key->vlen = vlen0;
  return r;
}

#define KEYS_PER_TABLE ((1000000))
  int
main(int argc, char ** argv)
{
  if (argc < 5) {
    printf("usage: <vlen> <nthreads> <ckeys(0/1)> <tags(0/1)>\n");
    return 0;
  }
  const u32 vlen = a2u32(argv[1]);
  const u32 nthreads = a2u32(argv[2]);
  const bool ckeys = argv[3][0] != '0';
  const bool tags = argv[4][0] != '0';
  char dirname[50];
  sprintf(dirname, "/tmp/db-%s", getlogin());

  struct msstz_cfg cfg = msstz_cfg_default;
  cfg.ckeys = ckeys;
  cfg.tags = tags;
  cfg.pages_accept = 0;
  struct msstz * const z = msstz_open(dirname, 128, &cfg);
  debug_assert(z);
  //msstz_set_nblks_accept(z, 0);
  struct msstv * const v0 = msstz_getv(z);
  debug_assert(v0);
  msstv_fprint(v0, stdout);
  msstz_putv(z, v0);

  const struct kvmap_api * const apim = &kvmap_api_wormhole;
  void * const mapm = wormhole_create(&kvmap_mm_ts);
  const struct kvmap_api * const apiv = &kvmap_api_msstv_ts;
  struct kv * const tmp = calloc(1,10000);
  struct kv * const out = calloc(1,10000);

  // five rounds: [1,3,5,...,1999999], [2000001,...,3999999], ..., [8000001,...,9999999]
  // do one compaction after each round
  for (u64 i = 0; i < 5; i++) {
    debug_perf_switch();
    const u64 start = i * KEYS_PER_TABLE;
    const u64 end = start + KEYS_PER_TABLE;
    void * const ref = kvmap_ref(apim, mapm);
    for (u64 x = start; x < end; x++) {
      kv_refill_hex64(tmp, (x<<1)+1, &i, sizeof(i)); // 1, 3, 5, ...
      tmp->vlen = vlen;
      apim->put(ref, tmp);
    }
    kvmap_unref(apim, ref);
    struct msstv * const oldv = msstz_getv(z);
    msstz_comp(z, apim, mapm, nthreads, 1, 0);
    struct kv ** const anchors = msstv_anchors(oldv);
    for (u64 a = 0; anchors[a]; a++)
      if (anchors[a]->vlen)
        printf("[%lu %lu] rejected\n", i, a);
    free(anchors);
    msstz_putv(z, oldv);
    apim->clean(mapm);

    // check the newest version
    struct msstv * const v = msstz_getv(z);
    debug_assert(v);
    msstv_fprint(v, stdout);
    msstz_putv(z, v);
  }

  debug_perf_switch();
  // verify keys
  const u64 nkeys = 5 * KEYS_PER_TABLE;
  struct msstv * const vz = msstz_getv(z);
  struct msstv_ref * const refz = msstv_ref(vz);
  debug_assert(refz);
  u64 err1 = 0;
  u64 err2 = 0;
  // all keys should be present
  for (u64 i = 0; i < nkeys; i++) {
    kv_refill_hex64(tmp, (i<<1)+1, NULL, 0);
    if (kvmap_kv_get(apiv, refz, tmp, out) == NULL)
      err1++;

    if (kvmap_kv_probe(apiv, refz, tmp) == false)
      err1++;

    // check data
    if (*(const u64 *)kv_vptr_c(out) != (i / KEYS_PER_TABLE))
      err2++;
  }
  printf("probe err1 %lu err2 %lu\n", err1, err2);
  msstv_unref(refz);
  msstz_putv(z, vz);

  debug_perf_switch();
  // delete [1,9,17,...] and do compaction
  void * const refd = kvmap_ref(apim, mapm);
  u64 ndel = 0;
  for (u64 i = 0; i < nkeys; i+=4) {
    kv_refill_hex64(tmp, (i<<1)+1, NULL, 0);
    memtable_del(apim, refd, tmp);
    ndel++;
  }
  kvmap_unref(apim, refd);

  // print if any partitions are rejected
  struct msstv * const oldv = msstz_getv(z);
  msstz_comp(z, apim, mapm, nthreads, 1, 0);
  struct kv ** const anchors = msstv_anchors(oldv);
  for (u64 a = 0; anchors[a]; a++)
    if (anchors[a]->vlen)
      printf("[del %lu] rejected (this can result in errors in the following execution)\n", a);
  free(anchors);
  msstz_putv(z, oldv);
  apim->clean(mapm);

  // check the newest version
  struct msstv * const v = msstz_getv(z);
  debug_assert(v);
  msstv_fprint(v, stdout);
  msstz_putv(z, v);

  debug_perf_switch();
  // verify deleted keys in msstv
  struct msstv * const vd = msstz_getv(z);
  struct msstv_ref * const refv = msstv_ref(vd);
  debug_assert(refv);
  err1 = 0;
  for (u64 i = 0; i < nkeys; i++) {
    kv_refill_hex64(tmp, (i<<1)+1, NULL, 0);
    const bool r = kvmap_kv_probe(apiv, refv, tmp);
    if (i % 4) {
      if (!r) {
        err1++;
        kvmap_kv_probe(apiv, refv, tmp); // for debugging
      }
    } else {
      if (r) {
        err1++;
        kvmap_kv_probe(apiv, refv, tmp); // for debugging
      }
    }
  }
  printf("total %lu deleted %lu  err1 %lu\n", nkeys, ndel, err1);

  // scan keys and count valid/deleted
  struct msstv_iter * const vi = msstv_iter_create(refv);
  msstv_iter_seek_ts(vi, kref_null());
  u64 nreg = 0;
  u64 nts = 0;
  while (msstv_iter_valid(vi)) {
    struct kv * const kv = msstv_iter_peek(vi, tmp);
    if (kv->vlen == SST_VLEN_TS)
      nts++;
    else
      nreg++;
    msstv_iter_skip1_ts(vi);
  }
  printf("nreg %lu nts %lu\n", nreg, nts);

  u64 serr = 0;
  struct kref kref;
  for (u64 i = 0; i <= nkeys; i++) {
    kv_refill_hex64(tmp, i<<1, NULL, 0); // 0,2,4,...
    kref_ref_kv(&kref, tmp);
    msstv_iter_seek_ts(vi, &kref);
    struct kv * const kv = msstv_iter_peek(vi, tmp);
    if (kv) { // > key
      const u64 key = strtoull((const char *)tmp->kv, NULL, 16);
      if (key < (i<<1))
        serr++;
    } else {
      printf("peeked NULL at %lu/%lu; this should only happen once at the end\n", i, nkeys);
    }
  }

  msstv_iter_destroy(vi);
  msstv_unref(refv);
  msstz_putv(z, vd);

  kvloader_destroy();
  apim->destroy(mapm);
  msstz_destroy(z);
  //sst_blk_debug("the end");
  free(tmp);
  free(out);
  return 0;
}