ycsbcdf.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"

struct priv {
  u32 klen;
  u32 vlen;
  u32 nscan;
  u32 cget;
  u32 cset;
  u32 cupd;
  u32 cscn;
  void * ref;
  void * iter;
  struct kv * tmp;
  struct kv * out;
  struct kv ** kvs;
};

// up to 10ms
#define VCTRSZ ((10000))

  static bool
kvmap_analyze(void * const passdata[2], const u64 dt, const struct vctr * const va, struct damp * const d, char * const out)
{
  (void)dt;
  (void)d;
  const struct kvmap_api * const api = passdata[0];
  if (api->async) {
    api->fprint(passdata[1], stdout);
    strcpy(out, "\n");
    return true;
  }

  u64 sum = 0;
  for (u64 i = 0; i < VCTRSZ; i++)
    sum += vctr_get(va, i);

  const u64 tot = sum;
  const double totd = (double)tot;
  sum = 0;
  u64 last = 0;
  printf("time_us count delta cdf\n0 0 0 0.000\n");
  for (u64 i = 1; i < VCTRSZ; i++) {
    const u64 tmp = vctr_get(va, i);
    if (tmp) {
      if ((i-1) != last)
        printf("%lu %lu %lu %.3lf\n", i-1, sum, 0lu, (double)sum * 100.0 / totd);
      sum += tmp;
      printf("%lu %lu %lu %.3lf\n", i, sum, tmp, (double)sum * 100.0 / totd);
      last = i;
    }
  }

  sprintf(out, "total %lu\n", tot);
  return true;
}

  static struct kv *
kvmap_merge_dummy(struct kv * const key0, void * const priv)
{
  (void)key0;
  return (struct kv *)priv;
}

  static void
latency_add(struct vctr * const vctr, const u64 dt)
{
  debug_assert(dt);
  const u64 us = (dt + 999) / 1000;
  if (us < VCTRSZ) {
    vctr_add1_atomic(vctr, us);
  } else {
    vctr_add1_atomic(vctr, VCTRSZ-1);
    printf("%s micro-second %lu\n", __func__, us);
  }
}

  static void
kvmap_batch(const struct forker_worker_info * const info,
    const struct priv * const priv, const u64 nr)
{
  const struct kvmap_api * const api = (typeof(api))info->passdata[0];
  void * const ref = priv->ref;
  struct rgen * const gen = info->gen;
  rgen_next_func next = info->rgen_next;
  rgen_next_func next_write = info->rgen_next_write;
  struct kv * const tmp = priv->tmp;
  u64 t0;

  for (u64 i = 0; i < nr; i++) {
    const u32 p = random_u64() & 0xffffu;
    if (p < priv->cget) { // GET
      kv_refill_hex64_klen(tmp, next(gen), priv->klen, NULL, 0);
      t0 = time_nsec();
      kvmap_kv_get(api, ref, tmp, priv->out);
    } else if (p < priv->cscn) { // SCAN
      kv_refill_hex64_klen(tmp, next(gen), priv->klen, NULL, 0);

      void * const iter = priv->iter;
      debug_assert(iter);

      t0 = time_nsec();
      kvmap_kv_iter_seek(api, iter, tmp);
      for (u32 k = 0; k < priv->nscan; k++)
        api->iter_next(iter, priv->out);

      // may need to park
      if (api->iter_park)
        api->iter_park(iter);

    } else if (p < priv->cset) { // SET
      kv_refill_hex64_klen(tmp, next_write(gen), priv->klen, NULL, 0);
      tmp->vlen = priv->vlen;
      t0 = time_nsec();
      kvmap_kv_put(api, ref, tmp);

    } else { // UPDATE (RMW)
      kv_refill_hex64_klen(tmp, next_write(gen), priv->klen, NULL, 0);
      tmp->vlen = priv->vlen;
      if (api->merge) { // use merge()
        t0 = time_nsec();
        kvmap_kv_merge(api, ref, tmp, kvmap_merge_dummy, tmp);
      } else { // GET & PUT
        t0 = time_nsec();
        (void)kvmap_kv_get(api, ref, tmp, priv->out); // read
        kvmap_kv_put(api, ref, tmp); // write
      }
    }
    const u64 dt = time_diff_nsec(t0);
    latency_add(info->vctr, dt);
  }
}

  static void *
kvmap_worker(void * const ptr)
{
  struct forker_worker_info * const info = (typeof(info))ptr;
  srandom_u64(info->seed);
  const struct kvmap_api * const api = (typeof(api))info->passdata[0];

  struct priv p = {};
  const u32 pset = a2u32(info->argv[0]);
  const u32 pupd = a2u32(info->argv[1]);
  const u32 pget = a2u32(info->argv[2]);
  const u32 pscn = a2u32(info->argv[3]);
  // scaled to 65536
  p.cget = pget * 65536 / 100; // 1st
  p.cscn = (pget + pscn) * 65536 / 100; // 2nd
  p.cset = (pget + pscn + pset) * 65536 / 100; // 3rd
  p.cupd = 65536; // not used
  (void)pupd;

  p.klen = a2u32(info->argv[4]);
  p.vlen = a2u32(info->argv[5]);
  p.nscan = a2u32(info->argv[6]);
  p.ref = kvmap_ref(api, info->passdata[1]);

  if (pscn) {
    p.iter = api->iter_create(p.ref);
    if (api->iter_park)
      api->iter_park(p.iter);
  }

  const u64 outlen = sizeof(struct kv) + p.klen + p.vlen + 4096;
  p.tmp = yalloc(outlen);
  debug_assert(p.tmp);
  memset(p.tmp, 0, outlen);
  p.out = yalloc(outlen);
  debug_assert(p.out);

  if (info->end_type == FORKER_END_TIME) {
    do {
      kvmap_batch(info, &p, 1lu << 14);
    } while (time_nsec() < info->end_magic);
  } else if (info->end_type == FORKER_END_COUNT) {
    kvmap_batch(info, &p, info->end_magic);
  }

  if (pscn)
    api->iter_destroy(p.iter);

  kvmap_unref(api, p.ref);
  free(p.tmp);
  free(p.out);
  return NULL;
}

#define NARGS ((7))
  static void
maptest_help_message(void)
{
  fprintf(stderr, "%s Usage: {api ... {rgen ... {pass ...}}}\n", __func__);
  kvmap_api_helper_message();
  forker_passes_message();
  fprintf(stderr, "%s wargs[%d]: <pset> <pupd> <pget> <pscn> <klen> <vlen> <nscan>\n", __func__, NARGS);
  fprintf(stderr, "%s load kv samples at cpu: MAPTEST_KVLOAD_CPU=<cpu>; default:1\n", __func__);
}

  static int
test_kvmap(const int argc, char ** const argv)
{
  const struct kvmap_api * api = NULL;
  void * map = NULL;
  const int n1 = kvmap_api_helper(argc, argv, NULL, &api, &map);
  if (n1 < 0)
    return n1;

  char *pref[64] = {};
  memcpy(pref, argv, sizeof(pref[0]) * (size_t)n1);

  struct pass_info pi = {};
  pi.passdata[0] = (void *)api;
  pi.passdata[1] = map;
  pi.vctr_size = VCTRSZ;
  pi.wf = kvmap_worker;
  pi.af = kvmap_analyze;
  const int n2 = forker_passes(argc - n1, argv + n1, pref, &pi, NARGS);

  if (api->fprint)
    api->fprint(map, stderr);

  api->destroy(map);
  if (n2 < 0) {
    return n2;
  } else {
    return n1 + n2;
  }
}

  int
main(int argc, char ** argv)
{
  if (argc < 3) {
    maptest_help_message();
    exit(0);
  }

  const bool r = forker_main(argc - 1, argv + 1, test_kvmap);
  if (r == false)
    maptest_help_message();
  return 0;
}