sweep_block.c

Go to the documentation of this file.

#include "sweep_block.h"
 
#include <string.h>
 
#include "miniz.h"
#include "ra8_err.h"
#include "ra8_vmem.h"
#include "ra8_vsource.h"
#include "sweep_block_internal.h"

typedef enum : uint32_t {
  k_cbs_block_512b   = 512U,    
  k_cbs_block_1kib   = 1024U,   
  k_cbs_block_4kib   = 4096U,   
  k_cbs_block_16kib  = 16384U,  
  k_cbs_block_64kib  = 65536U,  
  k_cbs_block_256kib = 262144U, 
} cbs_block_size_t;

void ra8_log_emit_error(const char* tag, const char* message)
{
  (void)tag;
  (void)message;
}

void ra8_log_emit_error_val(const char* tag, const char* message, uint32_t value)
{
  (void)tag;
  (void)message;
  (void)value;
}

RA8_INTERNAL
static uint32_t internal_pow2_ceil(uint32_t v)
{
  uint32_t p = 1U;
  while (p < v) {
    p <<= 1U;
  }
  return p;
}
 
/* ----------------------------------------------------------- the sweep -- */

typedef struct {
  ra8_vmem_t        vm;        
  ra8_vsource_t     vs;        
  ra8_vsource_obj_t objs[1];   
  cbs_meter_t       meter;     
  uint8_t*          frame_mem; 
  ra8_vmem_frame_t* meta;      
  ra8_vmem_key_t*   keys;      
  int32_t*          buckets;   
  uint32_t          frames;    
  uint32_t          object_id; 
} cbs_cache_t;

RA8_INTERNAL
static void internal_cache_close(cbs_cache_t* c)
{
  if (c == nullptr) {
    return;
  }
  *c = (cbs_cache_t){};
}

RA8_INTERNAL
static size_t internal_align_size(size_t value)
{
  const size_t alignment = alignof(max_align_t);
  return (value + alignment - 1U) & ~(alignment - 1U);
}

RA8_INTERNAL
static void* internal_workspace_take(cb_sweep_config_t* config, size_t* used, size_t bytes)
{
  const size_t span = internal_align_size(bytes);
  if ((*used > config->workspace_capacity) || (span > (config->workspace_capacity - *used))) {
    return nullptr;
  }
  void* result = &config->workspace[*used];
  *used += span;
  return result;
}

RA8_INTERNAL
static int internal_cache_storage(cbs_cache_t*       c,
                                  uint32_t           block_bytes,
                                  cb_sweep_config_t* config,
                                  uint32_t*          out_buckets)
{
  c->frames = (uint32_t)k_cbs_cache_bytes / block_bytes;
  if (c->frames == 0U) {
    c->frames = 1U;
  }
  uint32_t buckets = internal_pow2_ceil(c->frames * 2U);
  if (buckets < (uint32_t)k_cbs_bucket_min) {
    buckets = (uint32_t)k_cbs_bucket_min;
  }
  size_t used  = config->workspace_used;
  c->frame_mem = config->cache_backing;
  c->meta =
    (ra8_vmem_frame_t*)internal_workspace_take(config,
                                               &used,
                                               (size_t)c->frames * sizeof(ra8_vmem_frame_t));
  c->keys    = (ra8_vmem_key_t*)internal_workspace_take(config,
                                                        &used,
                                                        (size_t)c->frames * sizeof(ra8_vmem_key_t));
  c->buckets = (int32_t*)internal_workspace_take(config, &used, (size_t)buckets * sizeof(int32_t));
  config->workspace_required = used;
  if ((c->meta == nullptr) || (c->keys == nullptr) || (c->buckets == nullptr)) {
    return 1;
  }
  memset(config->cache_backing, 0, config->cache_capacity);
  memset(c->meta, 0, (size_t)c->frames * sizeof(ra8_vmem_frame_t));
  memset(c->keys, 0, (size_t)c->frames * sizeof(ra8_vmem_key_t));
  *out_buckets = buckets;
  return 0;
}

RA8_INTERNAL
static int internal_cache_open(cbs_cache_t*         c,
                               const cbs_backend_t* be,
                               uint32_t             blob_bytes,
                               uint32_t             block_bytes,
                               cb_sweep_config_t*   config)
{
  if ((c == nullptr) || (be == nullptr) || (be->read == nullptr) || (block_bytes == 0U) ||
      (config == nullptr) || (config->cache_backing == nullptr) ||
      (config->cache_capacity != (size_t)k_cbs_cache_bytes)) {
    return 1;
  }
  *c                = (cbs_cache_t){};
  uint32_t nbuckets = 0U;
  if (internal_cache_storage(c, block_bytes, config, &nbuckets) != 0) {
    internal_cache_close(c);
    return 1;
  }
  c->meter = (cbs_meter_t){.inner = be->read, .inner_ctx = be->read_ctx};
  if (ra8_vsource_init(&c->vs, c->objs, 1U) != k_ra8_ok) {
    internal_cache_close(c);
    return 1;
  }
  if (ra8_vsource_add_paged(&c->vs, priv_meter_read, &c->meter, 0U, blob_bytes, &c->object_id) !=
      k_ra8_ok) {
    internal_cache_close(c);
    return 1;
  }
  const ra8_vmem_cfg_t cfg = {.frame_mem    = c->frame_mem,
                              .frame_bytes  = block_bytes,
                              .frame_count  = c->frames,
                              .meta         = c->meta,
                              .keys         = c->keys,
                              .buckets      = c->buckets,
                              .bucket_count = nbuckets,
                              .loader       = ra8_vsource_loader,
                              .loader_ctx   = &c->vs};
  if (ra8_vmem_init(&c->vm, &cfg) != k_ra8_ok) {
    internal_cache_close(c);
    return 1;
  }
  return 0;
}

RA8_INTERNAL
static int internal_drive(cbs_cache_t*        c,
                          ra8_vsource_read_fn payload_read,
                          void*               payload_ctx,
                          uint64_t            n_reads,
                          uint64_t            wrap_bytes,
                          cbs_row_t*          row)
{
  if ((c == nullptr) || (payload_read == nullptr) || (row == nullptr) || (wrap_bytes == 0U)) {
    return 1;
  }
  const uint64_t req   = (uint64_t)k_cbs_req_bytes;
  const uint64_t block = (uint64_t)c->vm.cfg.frame_bytes;
  uint64_t       bad   = 0U;
  const uint64_t t0    = priv_now_ns();
  for (uint64_t i = 0U; i < n_reads; ++i) {
    const uint64_t off  = (i * req) % wrap_bytes;
    void*          page = nullptr;
    if (ra8_vmem_get(&c->vm, c->object_id, off, &page) != k_ra8_ok) {
      bad++;
      continue;
    }
    const uint8_t* piece = &((const uint8_t*)page)[off % block];
    uint8_t        expected[k_cbs_req_bytes];
    if ((payload_read(payload_ctx, off, expected, (uint32_t)req) != k_ra8_ok) ||
        (memcmp(piece, expected, (size_t)req) != 0)) {
      bad++;
    }
    if (ra8_vmem_put(&c->vm, page) != k_ra8_ok) {
      bad++;
    }
  }
  row->wall_ns  = priv_now_ns() - t0;
  row->reads    = n_reads;
  uint32_t hits = 0U;
  uint32_t miss = 0U;
  uint32_t evic = 0U;
  if (ra8_vmem_stats(&c->vm, &hits, &miss, &evic) != k_ra8_ok) {
    return 1;
  }
  row->hits      = (uint64_t)hits;
  row->misses    = (uint64_t)miss;
  row->evictions = (uint64_t)evic;
  if (bad != 0U) {
    return 1;
  }
  return 0;
}

typedef enum : uint8_t {
  k_cbs_leg_seq   = 0U, 
  k_cbs_leg_hot   = 1U, 
  k_cbs_leg_count = 2U, 
} cbs_leg_t;

static const char* const s_cbs_leg_names[k_cbs_leg_count] = {"seq", "hot"};

RA8_INTERNAL
static int internal_run_block(cbs_backend_t*      be,
                              ra8_vsource_read_fn payload_read,
                              void*               payload_ctx,
                              uint32_t            block_bytes,
                              cbs_row_t*          rows,
                              uint32_t*           nrows,
                              cb_sweep_config_t*  config)
{
  if ((be == nullptr) || (payload_read == nullptr) || (rows == nullptr) || (nrows == nullptr) ||
      (config == nullptr)) {
    return 1;
  }
  config->workspace_used = config->workspace_floor;
  if (be->setup(be, payload_read, payload_ctx, (uint32_t)k_cbs_blob_bytes, block_bytes, config) !=
      0) {
    return 1;
  }
  int rc = 0;
  for (uint8_t leg = 0U; leg < (uint8_t)k_cbs_leg_count; ++leg) {
    cbs_cache_t c = {};
    if (internal_cache_open(&c, be, (uint32_t)k_cbs_blob_bytes, block_bytes, config) != 0) {
      rc = 1;
      break;
    }
    const uint64_t src0 = (be->src_bytes != nullptr) ? *be->src_bytes : 0U;
    const bool     seq  = (leg == (uint8_t)k_cbs_leg_seq);
    const uint64_t n =
      seq ? ((uint64_t)k_cbs_seq_passes * (uint64_t)k_cbs_blob_bytes / (uint64_t)k_cbs_req_bytes)
          : (uint64_t)k_cbs_hot_reads;
    const uint64_t wrap = seq ? (uint64_t)k_cbs_blob_bytes : (uint64_t)block_bytes;
    cbs_row_t*     row  = &rows[*nrows];
    *row                = (cbs_row_t){.backend       = be->name,
                                      .leg           = s_cbs_leg_names[leg],
                                      .block_bytes   = block_bytes,
                                      .frames        = c.frames,
                                      .backing_bytes = be->backing_bytes};
    const int drc       = internal_drive(&c, payload_read, payload_ctx, n, wrap, row);
    row->be_calls       = c.meter.calls;
    row->be_bytes       = c.meter.bytes;
    row->src_bytes      = (be->src_bytes != nullptr) ? (*be->src_bytes - src0) : c.meter.bytes;
    internal_cache_close(&c);
    if (drc != 0) {
      rc = 1;
      break;
    }
    (*nrows)++;
    if (priv_print_row(config->output, row) != 0) {
      rc = 1;
      break;
    }
  }
  be->teardown(be);
  return rc;
}

static const uint32_t s_cbs_blocks[] = {
  (uint32_t)k_cbs_block_512b,
  (uint32_t)k_cbs_block_1kib,
  (uint32_t)k_cbs_block_4kib,
  (uint32_t)k_cbs_block_16kib,
  (uint32_t)k_cbs_block_64kib,
  (uint32_t)k_cbs_block_256kib,
};

RA8_INTERNAL
static int internal_execute(cb_sweep_config_t* config,
                            cbs_payload_t*     payload,
                            cbs_backend_t*     backends,
                            uint32_t           backend_count,
                            cbs_row_t*         rows,
                            uint32_t*          row_count)
{
  const uint32_t block_count = (uint32_t)(sizeof(s_cbs_blocks) / sizeof(s_cbs_blocks[0]));
  int            result      = 0;
  for (uint32_t backend = 0U; (backend < backend_count) && (result == 0); ++backend) {
    for (uint32_t size = 0U; (size < block_count) && (result == 0); ++size) {
      result = internal_run_block(&backends[backend],
                                  priv_payload_read,
                                  payload,
                                  s_cbs_blocks[size],
                                  rows,
                                  row_count,
                                  config);
    }
  }
  return result;
}

RA8_INTERNAL
static int internal_summary(cb_sweep_config_t*   config,
                            const cbs_backend_t* backends,
                            uint32_t             backend_count,
                            const cbs_row_t*     rows,
                            uint32_t             row_count)
{
  const uint32_t block_count = (uint32_t)(sizeof(s_cbs_blocks) / sizeof(s_cbs_blocks[0]));
  if (cb_sink_format(config->output,
                     "\n## Summary (payload 8 MiB, cache budget 1 MiB, %u B reader requests)\n",
                     (unsigned)k_cbs_req_bytes) != k_cb_io_ok) {
    return 1;
  }
  for (uint32_t backend = 0U; backend < backend_count; ++backend) {
    if ((priv_print_seq_table(config->output,
                              rows,
                              row_count,
                              backends[backend].name,
                              s_cbs_blocks,
                              block_count) != 0) ||
        (priv_print_hot_table(config->output,
                              rows,
                              row_count,
                              backends[backend].name,
                              s_cbs_blocks,
                              block_count) != 0)) {
      return 1;
    }
  }
  return priv_print_crossover(config->output, rows, row_count, s_cbs_blocks, block_count);
}
 
int cb_sweep_block(cb_sweep_config_t* config)
{
  if ((config == nullptr) || (config->workspace == nullptr) || (config->output == nullptr) ||
      (config->error == nullptr) || (config->scratch == nullptr)) {
    return 1;
  }
  config->cache_required     = (size_t)k_cbs_cache_bytes;
  const size_t payload_bytes = priv_payload_workspace_required();
  config->workspace_required = payload_bytes;
  cbs_payload_t payload      = {};
  if ((payload_bytes > config->workspace_capacity) ||
      (priv_payload_init(&payload, config->workspace, config->workspace_capacity) != 0)) {
    return 1;
  }
  config->workspace_floor = internal_align_size(payload_bytes);
  config->workspace_used  = config->workspace_floor;
  if (cb_sink_format(config->output, "# #208 block/frame-size sweep\n\n") != k_cb_io_ok ||
      cb_sink_format(config->output,
                     "payload=%u cache_budget=%u req=%u seq_passes=%u hot_reads=%u zlib_level=%d "
                     "cache=ra8_vmem(SLRU)\n\n",
                     (unsigned)k_cbs_blob_bytes,
                     (unsigned)k_cbs_cache_bytes,
                     (unsigned)k_cbs_req_bytes,
                     (unsigned)k_cbs_seq_passes,
                     (unsigned)k_cbs_hot_reads,
                     MZ_BEST_COMPRESSION) != k_cb_io_ok) {
    return 1;
  }
 
  cbs_row_t      rows[k_cbs_max_rows] = {};
  uint32_t       row_count            = 0U;
  uint32_t       backend_count        = 0U;
  cbs_backend_t* backends             = priv_backends(&backend_count);
  int            rc = internal_execute(config, &payload, backends, backend_count, rows, &row_count);
  if (rc == 0) {
    rc = internal_summary(config, backends, backend_count, rows, row_count);
  }
  if (rc != 0) {
    (void)cb_sink_format(config->error,
                         "sweep-block: sweep aborted (cache required=%zu supplied=%zu; "
                         "workspace required=%zu supplied=%zu)\n",
                         config->cache_required,
                         config->cache_capacity,
                         config->workspace_required,
                         config->workspace_capacity);
  }
  return rc;
}