mdl_politeness.c

Go to the documentation of this file.

#include "mdl_politeness.h"
 
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
 
#include "ra8_attributes.h"

typedef enum : uint64_t {
  k_seed_fallback = 0x9E3779B97F4A7C15ULL, 
} mdl_seed_t;

typedef enum : uint8_t {
  k_xs_shift_a = 13U, 
  k_xs_shift_b = 7U,  
  k_xs_shift_c = 17U, 
} mdl_xorshift_t;

typedef enum : uint32_t {
  k_ms_per_s  = 1000U,    
  k_ns_per_ms = 1000000U, 
  k_dec_base  = 10U,      
} mdl_time_unit_t;
 
void mdl_politeness_init(mdl_politeness_t* p, uint64_t seed)
{
  mdl_politeness_init_clock(p, seed, nullptr, nullptr);
}
 
void mdl_politeness_init_clock(mdl_politeness_t* p,
                               uint64_t          seed,
                               mdl_sleep_fn      sleep_fn,
                               void*             sleep_ctx)
{
  if (p == nullptr) {
    return;
  }
  p->state     = (seed == 0U) ? (uint64_t)k_seed_fallback : seed;
  p->sleep_fn  = sleep_fn;
  p->sleep_ctx = sleep_ctx;
}

RA8_INTERNAL static uint64_t internal_next_rand(uint64_t* state)
{
  uint64_t x = *state;
  x ^= x << (uint64_t)k_xs_shift_a;
  x ^= x >> (uint64_t)k_xs_shift_b;
  x ^= x << (uint64_t)k_xs_shift_c;
  *state = x;
  return x;
}

RA8_INTERNAL static uint32_t internal_draw_range(uint64_t* state, uint32_t min_ms, uint32_t max_ms)
{
  if (max_ms < min_ms) {
    max_ms = min_ms;
  }
  /* 64-bit span keeps the full-range (min=0, max=UINT32_MAX) case from wrapping
   * while preserving the exact modulo of the original jitter for smaller spans. */
  const uint64_t span = (uint64_t)(max_ms - min_ms) + 1U;
  return min_ms + (uint32_t)(internal_next_rand(state) % span);
}

RA8_INTERNAL static void internal_host_sleep_ms(uint32_t ms)
{
  struct timespec ts = {.tv_sec  = (time_t)(ms / k_ms_per_s),
                        .tv_nsec = (long)((ms % k_ms_per_s) * k_ns_per_ms)};
  (void)nanosleep(&ts, nullptr);
}
 
uint32_t mdl_politeness_wait(mdl_politeness_t* p, uint32_t min_ms, uint32_t max_ms)
{
  if (p == nullptr) {
    return 0U;
  }
  const uint32_t delayms = internal_draw_range(&p->state, min_ms, max_ms);
 
  if (p->sleep_fn != nullptr) {
    p->sleep_fn(p->sleep_ctx, delayms);
  } else {
    internal_host_sleep_ms(delayms);
  }
  return delayms;
}
 
/* ======================================================================== *
 *  Retry-After parsing (#301)                                              *
 * ======================================================================== */

typedef enum : uint16_t {
  k_http_too_many_req = 429U, 
  k_http_unavailable  = 503U, 
} mdl_http_throttle_t;

typedef enum : uint32_t {
  k_def_rate_per_min    = 60U,    
  k_def_burst           = 4U,     
  k_def_backoff_base_ms = 1000U,  
  k_def_backoff_max_ms  = 60000U, 
} mdl_gov_defaults_t;

typedef enum : uint16_t {
  k_def_decay_after  = 4U, 
  k_def_max_inflight = 1U, 
} mdl_gov_defaults16_t;

RA8_INTERNAL static uint32_t internal_ms_from_secs(int64_t secs)
{
  if (secs <= 0) {
    return 0U;
  }
  if (secs > (int64_t)(UINT32_MAX / k_ms_per_s)) {
    return UINT32_MAX;
  }
  return (uint32_t)(secs * (int64_t)k_ms_per_s);
}

RA8_INTERNAL static bool internal_all_digits(const char* s)
{
  if (*s == '\0') {
    return false;
  }
  for (const char* c = s; *c != '\0'; ++c) {
    if ((*c < '0') || (*c > '9')) {
      return false;
    }
  }
  return true;
}

RA8_INTERNAL static bool
internal_parse_http_date(const char* value, int64_t now_wall_s, uint32_t* out_ms)
{
  static const char* const k_fmts[] = {
    "%a, %d %b %Y %H:%M:%S GMT", /* IMF-fixdate: Sun, 06 Nov 1994 08:49:37 GMT  */
    "%A, %d-%b-%y %H:%M:%S GMT", /* RFC 850:     Sunday, 06-Nov-94 08:49:37 GMT */
  };
  for (size_t i = 0U; i < (sizeof(k_fmts) / sizeof(k_fmts[0])); ++i) {
    struct tm tmv = {};
    if (strptime(value, k_fmts[i], &tmv) != nullptr) {
      const time_t t = timegm(&tmv);
      if (t == (time_t)-1) {
        continue;
      }
      *out_ms = internal_ms_from_secs((int64_t)t - now_wall_s);
      return true;
    }
  }
  return false;
}
 
bool mdl_retry_after_parse(const char* value, int64_t now_wall_s, uint32_t* out_ms)
{
  if ((value == nullptr) || (out_ms == nullptr)) {
    return false;
  }
  while ((*value == ' ') || (*value == '\t')) {
    ++value;
  }
  if (*value == '\0') {
    return false;
  }
  if (internal_all_digits(value)) {
    *out_ms = internal_ms_from_secs((int64_t)strtoull(value, nullptr, k_dec_base));
    return true;
  }
  return internal_parse_http_date(value, now_wall_s, out_ms);
}
 
/* ======================================================================== *
 *  Per-host politeness governor (#301)                                     *
 * ======================================================================== */

RA8_INTERNAL static int64_t internal_min_i64(int64_t a, int64_t b)
{
  return (a < b) ? a : b;
}

RA8_INTERNAL static int64_t internal_max_i64(int64_t a, int64_t b)
{
  return (a > b) ? a : b;
}

RA8_INTERNAL static int64_t internal_gov_now(const mdl_governor_t* g)
{
  if (g->now_fn != nullptr) {
    return g->now_fn(g->now_ctx);
  }
  /* CLOCK_MONOTONIC, not CLOCK_REALTIME (#509). The governor only ever
   * subtracts two readings -- token refill, request spacing, the backoff
   * gate -- and CLOCK_REALTIME is steppable: an NTP correction forward makes
   * the governor believe the spacing has elapsed and hammer the remote host,
   * and one backward makes it wait out the step. A ~4 minute backward step
   * was measured happening repeatedly on a fleet host, so this is an observed
   * hazard rather than a theoretical one. */
  struct timespec ts = {};
  (void)clock_gettime(CLOCK_MONOTONIC, &ts);
  return ((int64_t)ts.tv_sec * (int64_t)k_ms_per_s) + ((int64_t)ts.tv_nsec / (int64_t)k_ns_per_ms);
}

RA8_INTERNAL static void internal_gov_sleep(mdl_governor_t* g, int64_t ms)
{
  if (ms <= 0) {
    return;
  }
  const uint32_t d = (ms > (int64_t)UINT32_MAX) ? UINT32_MAX : (uint32_t)ms;
  if (g->sleep_fn != nullptr) {
    g->sleep_fn(g->sleep_ctx, d);
  } else {
    internal_host_sleep_ms(d);
  }
}

RA8_INTERNAL static int64_t internal_gov_interval_ms(const mdl_gov_cfg_t* cfg)
{
  return (cfg->rate_per_min > 0U) ? ((int64_t)k_mdl_gov_ms_per_req / (int64_t)cfg->rate_per_min)
                                  : 0;
}

RA8_INTERNAL static int64_t internal_gov_cap_ms(const mdl_gov_cfg_t* cfg)
{
  return internal_gov_interval_ms(cfg) * (int64_t)cfg->burst;
}

RA8_INTERNAL static mdl_host_rec_t* internal_gov_find(mdl_governor_t* g, const char* host)
{
  if (host == nullptr) {
    return nullptr;
  }
  for (uint16_t i = 0U; i < (uint16_t)k_mdl_gov_max_hosts; ++i) {
    if (g->hosts[i].used && (strcmp(g->hosts[i].host, host) == 0)) {
      return &g->hosts[i];
    }
  }
  return nullptr;
}

RA8_INTERNAL static mdl_host_rec_t*
internal_gov_get(mdl_governor_t* g, const char* host, int64_t now)
{
  mdl_host_rec_t* rec = internal_gov_find(g, host);
  if ((rec != nullptr) || (host == nullptr)) {
    return rec;
  }
  for (uint16_t i = 0U; i < (uint16_t)k_mdl_gov_max_hosts; ++i) {
    if (!g->hosts[i].used) {
      g->hosts[i]                  = (mdl_host_rec_t){};
      g->hosts[i].used             = true;
      g->hosts[i].credit_ms        = internal_gov_cap_ms(&g->cfg); /* start full: allow a burst */
      g->hosts[i].last_ms          = now;
      g->hosts[i].earliest_next_ms = now;
      (void)snprintf(g->hosts[i].host, sizeof(g->hosts[i].host), "%s", host);
      return &g->hosts[i];
    }
  }
  return nullptr;
}

RA8_INTERNAL static int64_t
internal_gov_schedule(mdl_governor_t* g, mdl_host_rec_t* rec, int64_t now)
{
  const int64_t interval  = internal_gov_interval_ms(&g->cfg);
  const int64_t cap       = internal_gov_cap_ms(&g->cfg);
  const int64_t elapsed   = internal_max_i64(now - rec->last_ms, 0);
  rec->credit_ms          = internal_min_i64(rec->credit_ms + elapsed, cap);
  const int64_t rate_wait = (rec->credit_ms < interval) ? (interval - rec->credit_ms) : 0;
  const int64_t target    = internal_max_i64(now + rate_wait, rec->earliest_next_ms);
  const int64_t wait      = target - now;
  rec->credit_ms = internal_max_i64(internal_min_i64(rec->credit_ms + wait, cap) - interval, 0);
  rec->last_ms   = target;
  return wait;
}
 
mdl_gov_cfg_t mdl_gov_cfg_default(void)
{
  return (mdl_gov_cfg_t){
    .rate_per_min    = (uint32_t)k_def_rate_per_min,
    .burst           = (uint32_t)k_def_burst,
    .backoff_base_ms = (uint32_t)k_def_backoff_base_ms,
    .backoff_max_ms  = (uint32_t)k_def_backoff_max_ms,
    .decay_after     = (uint16_t)k_def_decay_after,
    .max_inflight    = (uint16_t)k_def_max_inflight,
  };
}
 
void mdl_governor_init(mdl_governor_t* g, const mdl_gov_cfg_t* cfg, uint64_t seed)
{
  mdl_governor_init_clock(g, cfg, seed, nullptr, nullptr, nullptr, nullptr);
}
 
void mdl_governor_init_clock(mdl_governor_t*      g,
                             const mdl_gov_cfg_t* cfg,
                             uint64_t             seed,
                             mdl_now_fn           now_fn,
                             void*                now_ctx,
                             mdl_sleep_fn         sleep_fn,
                             void*                sleep_ctx)
{
  if (g == nullptr) {
    return;
  }
  memset(g, 0, sizeof(*g));
  g->cfg = (cfg != nullptr) ? *cfg : mdl_gov_cfg_default();
  if (g->cfg.burst == 0U) {
    g->cfg.burst = 1U;
  }
  if (g->cfg.max_inflight == 0U) {
    g->cfg.max_inflight = 1U;
  }
  g->rng       = (seed == 0U) ? (uint64_t)k_seed_fallback : seed;
  g->now_fn    = now_fn;
  g->now_ctx   = now_ctx;
  g->sleep_fn  = sleep_fn;
  g->sleep_ctx = sleep_ctx;
}
 
ra8_err_t mdl_governor_acquire(mdl_governor_t* g,
                               const char*     host,
                               uint32_t        jitter_min_ms,
                               uint32_t        jitter_max_ms)
{
  if (g == nullptr) {
    return k_ra8_ok; /* pacing disabled -- matches a NULL jitter source */
  }
  const int64_t   now = internal_gov_now(g);
  mdl_host_rec_t* rec = internal_gov_get(g, host, now);
  if (rec == nullptr) {
    /* NULL host or table full: still space requests with jitter, no tracking. */
    internal_gov_sleep(g, (int64_t)internal_draw_range(&g->rng, jitter_min_ms, jitter_max_ms));
    return k_ra8_ok;
  }
  if (rec->inflight >= g->cfg.max_inflight) {
    return k_ra8_err_would_block;
  }
  const int64_t wait   = internal_gov_schedule(g, rec, now);
  const int64_t jitter = (int64_t)internal_draw_range(&g->rng, jitter_min_ms, jitter_max_ms);
  internal_gov_sleep(g, wait + jitter);
  rec->inflight += 1U;
  return k_ra8_ok;
}
 
void mdl_governor_release(mdl_governor_t* g, const char* host)
{
  if (g == nullptr) {
    return;
  }
  mdl_host_rec_t* rec = internal_gov_find(g, host);
  if ((rec != nullptr) && (rec->inflight > 0U)) {
    rec->inflight -= 1U;
  }
}

RA8_INTERNAL static int64_t internal_gov_backoff_window(const mdl_gov_cfg_t* cfg, uint16_t level)
{
  int64_t        window = (int64_t)cfg->backoff_base_ms;
  const uint16_t shifts = (level > 1U) ? (uint16_t)(level - 1U) : 0U;
  for (uint16_t i = 0U; (i < shifts) && (i < (uint16_t)k_mdl_gov_level_max); ++i) {
    window <<= 1;
  }
  if ((window < 0) || (window > (int64_t)cfg->backoff_max_ms)) {
    window = (int64_t)cfg->backoff_max_ms;
  }
  return window;
}

RA8_INTERNAL static void
internal_gov_on_throttle(mdl_governor_t* g, mdl_host_rec_t* rec, int64_t now, uint32_t retry_ms)
{
  if (rec->backoff_level < (uint16_t)k_mdl_gov_level_max) {
    rec->backoff_level += 1U;
  }
  rec->success_streak   = 0U;
  const int64_t window  = internal_gov_backoff_window(&g->cfg, rec->backoff_level);
  const int64_t backoff = (int64_t)internal_draw_range(&g->rng, 0U, (uint32_t)window);
  const int64_t gate    = now + internal_max_i64(backoff, (int64_t)retry_ms);
  rec->earliest_next_ms = internal_max_i64(rec->earliest_next_ms, gate);
}

RA8_INTERNAL static void internal_gov_on_success(mdl_governor_t* g,
                                                 mdl_host_rec_t* rec,
                                                 int64_t         now,
                                                 bool            has_retry,
                                                 uint32_t        retry_ms)
{
  rec->success_streak += 1U;
  if ((g->cfg.decay_after > 0U) && (rec->success_streak >= g->cfg.decay_after)) {
    if (rec->backoff_level > 0U) {
      rec->backoff_level -= 1U;
    }
    rec->success_streak = 0U;
  }
  if (has_retry) {
    rec->earliest_next_ms = internal_max_i64(rec->earliest_next_ms, now + (int64_t)retry_ms);
  }
}
 
void mdl_governor_observe_at_wall(mdl_governor_t* g,
                                  const char*     host,
                                  long            status,
                                  const char*     retry_after,
                                  int64_t         now_wall_s)
{
  if (g == nullptr) {
    return;
  }
  const int64_t   now = internal_gov_now(g);
  mdl_host_rec_t* rec = internal_gov_get(g, host, now);
  if (rec == nullptr) {
    return; /* NULL host or table full */
  }
  uint32_t   retry_ms = 0U;
  const bool has_retry =
    (retry_after != nullptr) && mdl_retry_after_parse(retry_after, now_wall_s, &retry_ms);
  const bool throttled =
    (status == (long)k_http_too_many_req) || (status == (long)k_http_unavailable);
  if (throttled) {
    internal_gov_on_throttle(g, rec, now, retry_ms);
  } else {
    internal_gov_on_success(g, rec, now, has_retry, retry_ms);
  }
}
 
void mdl_governor_observe(mdl_governor_t* g, const char* host, long status, const char* retry_after)
{
  mdl_governor_observe_at_wall(g, host, status, retry_after, (int64_t)time(nullptr));
}
 
bool mdl_governor_peek(const mdl_governor_t* g,
                       const char*           host,
                       uint16_t*             backoff_level,
                       int64_t*              earliest_next_ms)
{
  if ((g == nullptr) || (host == nullptr)) {
    return false;
  }
  for (uint16_t i = 0U; i < (uint16_t)k_mdl_gov_max_hosts; ++i) {
    if (g->hosts[i].used && (strcmp(g->hosts[i].host, host) == 0)) {
      if (backoff_level != nullptr) {
        *backoff_level = g->hosts[i].backoff_level;
      }
      if (earliest_next_ms != nullptr) {
        *earliest_next_ms = g->hosts[i].earliest_next_ms;
      }
      return true;
    }
  }
  return false;
}