sweep_block_backends.c

Go to the documentation of this file.

#include <string.h>
#include <time.h>
 
#include "book_chunked.h"
#include "miniz.h"
#include "ra8_err.h"
#include "sweep_block_internal.h"
 
typedef enum : uint32_t {
  k_cbs_codec_input_bytes = 4096U, 
  k_cbs_codec_window_bits = 15U,   
} cbs_codec_limit_t;
 
static_assert(sizeof(tdefl_compressor) <= (size_t)k_cbs_cache_bytes,
              "tdefl overlay must fit the semantic cache backing");
static_assert(alignof(max_align_t) >= alignof(tdefl_compressor),
              "composition cache alignment must satisfy tdefl");
 
typedef struct {
  cb_scratch_t* scratch; 
  uint64_t      offset;  
  bool          failed;  
} cbs_pack_sink_t;
 
typedef struct {
  cb_scratch_t*       scratch;        
  tinfl_decompressor* inflater;       
  uint8_t*            input;          
  uint32_t            input_capacity; 
  uint32_t            chunk_bytes;    
  uint32_t            chunk_count;    
  uint64_t            total_bytes;    
  uint64_t            payload_offset; 
  uint64_t            source_bytes;   
} cbs_rbkc_t;
 
RA8_PRIV
uint64_t priv_now_ns(void)
{
  struct timespec ts = {};
  (void)clock_gettime(CLOCK_MONOTONIC, &ts);
  return ((uint64_t)ts.tv_sec * (uint64_t)k_cbs_ns_per_s) + (uint64_t)ts.tv_nsec;
}
 
RA8_PRIV
ra8_err_t priv_meter_read(void* ctx, uint64_t offset, uint8_t* buffer, uint32_t length)
{
  cbs_meter_t* meter = (cbs_meter_t*)ctx;
  if ((meter == nullptr) || (meter->inner == nullptr)) {
    return k_ra8_err_null_ptr;
  }
  meter->calls++;
  meter->bytes += length;
  return meter->inner(meter->inner_ctx, offset, buffer, length);
}

RA8_INTERNAL
static int internal_mem_setup(cbs_backend_t*      be,
                              ra8_vsource_read_fn payload_read,
                              void*               payload_ctx,
                              uint32_t            blob_bytes,
                              uint32_t            block_bytes,
                              cb_sweep_config_t*  config)
{
  (void)block_bytes;
  (void)config;
  if ((be == nullptr) || (payload_read == nullptr) || (payload_ctx == nullptr) ||
      (blob_bytes == 0U)) {
    return 1;
  }
  be->read          = payload_read;
  be->read_ctx      = payload_ctx;
  be->backing_bytes = blob_bytes;
  be->src_bytes     = nullptr;
  return 0;
}

RA8_INTERNAL
static void internal_backend_teardown(cbs_backend_t* be)
{
  if (be != nullptr) {
    be->read      = nullptr;
    be->read_ctx  = nullptr;
    be->src_bytes = nullptr;
  }
}

RA8_INTERNAL
static void
internal_rbkc_header(uint8_t* output, uint32_t block_bytes, uint64_t total, uint32_t count)
{
  size_t position = 0U;
  memcpy(&output[position], "RBKC", sizeof("RBKC") - 1U);
  position += sizeof("RBKC") - 1U;
  memcpy(&output[position], &block_bytes, sizeof(block_bytes));
  position += sizeof(block_bytes);
  memcpy(&output[position], &total, sizeof(total));
  position += sizeof(total);
  memcpy(&output[position], &count, sizeof(count));
  position += sizeof(count);
  const uint32_t reserved = 0U;
  memcpy(&output[position], &reserved, sizeof(reserved));
}

RA8_INTERNAL
static mz_bool internal_pack_write(const void* data, int length, void* user)
{
  cbs_pack_sink_t* sink = (cbs_pack_sink_t*)user;
  if ((sink == nullptr) || (data == nullptr) || (length < 0) || sink->failed) {
    return MZ_FALSE;
  }
  const cb_io_status_t status =
    sink->scratch->write(sink->scratch->ctx, sink->offset, (void*)data, (size_t)length);
  if (status != k_cb_io_ok) {
    sink->failed = true;
    return MZ_FALSE;
  }
  sink->offset += (uint64_t)length;
  return MZ_TRUE;
}

RA8_INTERNAL
static int internal_rbkc_pack_chunk(tdefl_compressor*   compressor,
                                    cbs_pack_sink_t*    sink,
                                    ra8_vsource_read_fn payload_read,
                                    void*               payload_ctx,
                                    uint8_t*            input,
                                    uint32_t            input_capacity,
                                    uint32_t            offset,
                                    uint32_t            length)
{
  const mz_uint flags = tdefl_create_comp_flags_from_zip_params(MZ_BEST_COMPRESSION,
                                                                (int)k_cbs_codec_window_bits,
                                                                MZ_DEFAULT_STRATEGY);
  if (tdefl_init(compressor, internal_pack_write, sink, (int)flags) != TDEFL_STATUS_OKAY) {
    return 1;
  }
  uint32_t complete = 0U;
  while (complete < length) {
    uint32_t span = length - complete;
    if (span > input_capacity) {
      span = input_capacity;
    }
    if (payload_read(payload_ctx, (uint64_t)offset + complete, input, span) != k_ra8_ok ||
        tdefl_compress_buffer(compressor, input, span, TDEFL_NO_FLUSH) < TDEFL_STATUS_OKAY) {
      return 1;
    }
    complete += span;
  }
  const tdefl_status status = tdefl_compress_buffer(compressor, nullptr, 0U, TDEFL_FINISH);
  return ((status == TDEFL_STATUS_DONE) && !sink->failed) ? 0 : 1;
}

RA8_INTERNAL
static int internal_rbkc_pack(ra8_vsource_read_fn payload_read,
                              void*               payload_ctx,
                              uint32_t            blob_bytes,
                              uint32_t            block_bytes,
                              cb_sweep_config_t*  config,
                              uint8_t*            input,
                              uint32_t            input_capacity,
                              uint64_t*           output_length)
{
  if ((sizeof(tdefl_compressor) > config->cache_capacity) ||
      (((uintptr_t)config->cache_backing % alignof(tdefl_compressor)) != 0U)) {
    config->workspace_required = sizeof(tdefl_compressor);
    return 1;
  }
  const uint32_t count          = (blob_bytes + block_bytes - 1U) / block_bytes;
  const uint64_t table_bytes    = ((uint64_t)count + 1U) * k_book_container_entry_len;
  const uint64_t payload_offset = (uint64_t)k_book_container_header_len + table_bytes;
  uint8_t        header[k_book_container_header_len] = {};
  internal_rbkc_header(header, block_bytes, blob_bytes, count);
  if (config->scratch->write(config->scratch->ctx, 0U, header, sizeof(header)) != k_cb_io_ok) {
    return 1;
  }
  uint64_t relative = 0U;
  if (config->scratch->write(config->scratch->ctx,
                             k_book_container_header_len,
                             &relative,
                             sizeof(relative)) != k_cb_io_ok) {
    return 1;
  }
  tdefl_compressor* compressor = (tdefl_compressor*)config->cache_backing;
  cbs_pack_sink_t   sink       = {.scratch = config->scratch, .offset = payload_offset};
  for (uint32_t chunk = 0U; chunk < count; ++chunk) {
    const uint32_t offset = chunk * block_bytes;
    uint32_t       length = blob_bytes - offset;
    if (length > block_bytes) {
      length = block_bytes;
    }
    if (internal_rbkc_pack_chunk(compressor,
                                 &sink,
                                 payload_read,
                                 payload_ctx,
                                 input,
                                 input_capacity,
                                 offset,
                                 length) != 0) {
      return 1;
    }
    relative = sink.offset - payload_offset;
    const uint64_t table_offset =
      (uint64_t)k_book_container_header_len + (((uint64_t)chunk + 1U) * k_book_container_entry_len);
    if (config->scratch->write(config->scratch->ctx, table_offset, &relative, sizeof(relative)) !=
        k_cb_io_ok) {
      return 1;
    }
  }
  *output_length        = sink.offset;
  config->scratch->size = sink.offset;
  memset(config->cache_backing, 0, config->cache_capacity);
  return 0;
}

RA8_INTERNAL
static ra8_err_t
internal_rbkc_offsets(cbs_rbkc_t* rb, uint32_t chunk, uint64_t* begin, uint64_t* end)
{
  const uint64_t table_offset =
    (uint64_t)k_book_container_header_len + ((uint64_t)chunk * k_book_container_entry_len);
  if ((rb->scratch->read(rb->scratch->ctx, table_offset, begin, sizeof(*begin)) != k_cb_io_ok) ||
      (rb->scratch->read(rb->scratch->ctx,
                         table_offset + k_book_container_entry_len,
                         end,
                         sizeof(*end)) != k_cb_io_ok) ||
      (*end < *begin) || ((rb->payload_offset + *end) > rb->scratch->size)) {
    return k_ra8_err_invalid_size;
  }
  return k_ra8_ok;
}

RA8_INTERNAL
static ra8_err_t internal_rbkc_inflate(cbs_rbkc_t* rb,
                                       uint64_t    begin,
                                       uint64_t    end,
                                       uint8_t*    output,
                                       uint32_t    output_length)
{
  tinfl_init(rb->inflater);
  uint64_t     source   = begin;
  uint32_t     produced = 0U;
  tinfl_status status   = TINFL_STATUS_NEEDS_MORE_INPUT;
  while (status != TINFL_STATUS_DONE) {
    uint32_t input_length = (uint32_t)(end - source);
    if (input_length > rb->input_capacity) {
      input_length = rb->input_capacity;
    }
    if ((input_length == 0U) || (rb->scratch->read(rb->scratch->ctx,
                                                   rb->payload_offset + source,
                                                   rb->input,
                                                   input_length) != k_cb_io_ok)) {
      return k_ra8_err_invalid_size;
    }
    rb->source_bytes += input_length;
    size_t          consumed  = input_length;
    size_t          available = output_length - produced;
    const mz_uint32 flags =
      (mz_uint32)(TINFL_FLAG_PARSE_ZLIB_HEADER | TINFL_FLAG_USING_NON_WRAPPING_OUTPUT_BUF |
                  (((source + input_length) < end) ? TINFL_FLAG_HAS_MORE_INPUT : 0U));
    status = tinfl_decompress(rb->inflater,
                              rb->input,
                              &consumed,
                              output,
                              &output[produced],
                              &available,
                              flags);
    source += consumed;
    produced += (uint32_t)available;
    if ((status < TINFL_STATUS_DONE) || (consumed == 0U)) {
      return k_ra8_err_invalid_size;
    }
  }
  return ((source == end) && (produced == output_length)) ? k_ra8_ok : k_ra8_err_invalid_size;
}

RA8_INTERNAL
static ra8_err_t internal_rbkc_read(void* ctx, uint64_t offset, uint8_t* buffer, uint32_t length)
{
  cbs_rbkc_t* rb = (cbs_rbkc_t*)ctx;
  if ((rb == nullptr) || (buffer == nullptr)) {
    return k_ra8_err_null_ptr;
  }
  if ((offset >= rb->total_bytes) || ((offset % rb->chunk_bytes) != 0U)) {
    return k_ra8_err_out_of_range;
  }
  const uint32_t chunk    = (uint32_t)(offset / rb->chunk_bytes);
  uint32_t       expected = (uint32_t)(rb->total_bytes - offset);
  if (expected > rb->chunk_bytes) {
    expected = rb->chunk_bytes;
  }
  if ((chunk >= rb->chunk_count) || (length != expected)) {
    return k_ra8_err_invalid_size;
  }
  uint64_t        begin  = 0U;
  uint64_t        end    = 0U;
  const ra8_err_t status = internal_rbkc_offsets(rb, chunk, &begin, &end);
  return (status == k_ra8_ok) ? internal_rbkc_inflate(rb, begin, end, buffer, length) : status;
}

RA8_INTERNAL
static void* internal_setup_take(cb_sweep_config_t* config, size_t bytes)
{
  const size_t alignment = alignof(max_align_t);
  const size_t span      = (bytes + alignment - 1U) & ~(alignment - 1U);
  if ((config->workspace_used > config->workspace_capacity) ||
      (span > (config->workspace_capacity - config->workspace_used))) {
    config->workspace_required = config->workspace_used + span;
    return nullptr;
  }
  void* result = &config->workspace[config->workspace_used];
  config->workspace_used += span;
  config->workspace_required = config->workspace_used;
  return result;
}

RA8_INTERNAL
static int internal_rbkc_setup(cbs_backend_t*      be,
                               ra8_vsource_read_fn payload_read,
                               void*               payload_ctx,
                               uint32_t            blob_bytes,
                               uint32_t            block_bytes,
                               cb_sweep_config_t*  config)
{
  if ((be == nullptr) || (payload_read == nullptr) || (config == nullptr) ||
      (config->scratch == nullptr) || (block_bytes == 0U)) {
    return 1;
  }
  uint8_t* input       = (uint8_t*)internal_setup_take(config, k_cbs_codec_input_bytes);
  uint64_t file_length = 0U;
  if ((input == nullptr) || (internal_rbkc_pack(payload_read,
                                                payload_ctx,
                                                blob_bytes,
                                                block_bytes,
                                                config,
                                                input,
                                                (uint32_t)k_cbs_codec_input_bytes,
                                                &file_length) != 0)) {
    return 1;
  }
  config->workspace_used = config->workspace_floor;
  cbs_rbkc_t*         rb = (cbs_rbkc_t*)internal_setup_take(config, sizeof(cbs_rbkc_t));
  tinfl_decompressor* inflater =
    (tinfl_decompressor*)internal_setup_take(config, sizeof(tinfl_decompressor));
  input = (uint8_t*)internal_setup_take(config, k_cbs_codec_input_bytes);
  if ((rb == nullptr) || (inflater == nullptr) || (input == nullptr)) {
    return 1;
  }
  const uint32_t count = (blob_bytes + block_bytes - 1U) / block_bytes;
  *rb      = (cbs_rbkc_t){.scratch        = config->scratch,
                          .inflater       = inflater,
                          .input          = input,
                          .input_capacity = k_cbs_codec_input_bytes,
                          .chunk_bytes    = block_bytes,
                          .chunk_count    = count,
                          .total_bytes    = blob_bytes,
                          .payload_offset = (uint64_t)k_book_container_header_len +
                                            (((uint64_t)count + 1U) * k_book_container_entry_len)};
  be->read = internal_rbkc_read;
  be->read_ctx      = rb;
  be->backing_bytes = file_length;
  be->src_bytes     = &rb->source_bytes;
  return 0;
}
 
static cbs_backend_t s_cbs_backends[] = {
  {.name = "mem", .setup = internal_mem_setup, .teardown = internal_backend_teardown},
  {.name = "rbkc-z9", .setup = internal_rbkc_setup, .teardown = internal_backend_teardown},
};
 
RA8_PRIV
cbs_backend_t* priv_backends(uint32_t* out_count)
{
  if (out_count != nullptr) {
    *out_count = (uint32_t)(sizeof(s_cbs_backends) / sizeof(s_cbs_backends[0]));
  }
  return s_cbs_backends;
}