client: Hold presses in queues, one queue per send_count

[tattlekey] / client / press.c

#include "press.h"
#include "blink.h"
#include "config.h"

static uint32_t press_next_send(press_t *s) {
  return s->timestamp + (1 << s->send_count) - 1;
}

static uint32_t press_queue_next_send(queue_t *q) {
  press_t press;
  return queue_try_peek(q, &press) ? press_next_send(&press) : UINT32_MAX;
}

static bool next_send_less_than(void *user_data, pheap_node_id_t a,
                                pheap_node_id_t b) {
  queue_t **sleeps = (queue_t **)user_data;
  return press_queue_next_send(sleeps[a]) < press_queue_next_send(sleeps[b]);
}

static void *xcalloc(size_t nmemb, size_t size) {
  void *p = calloc(nmemb, size);
  if (p == NULL)
    signal_error_by_blinking();
  return p;
}

press_pile_t *create_press_pile() {
  press_pile_t *pp = (press_pile_t *)xcalloc(1, sizeof(press_pile_t));
  pp->presses = (queue_t *)xcalloc(config_resend_count, sizeof(queue_t));
  pp->sleeps = (queue_t **)xcalloc(config_resend_count, sizeof(queue_t *));
  for (int i = 0; i < config_resend_count; i++) {
    uint element_count = 1 << i;
    element_count = MAX(element_count, 32);
    element_count = MIN(element_count, 512);
    queue_init(&pp->presses[i], sizeof(press_t), element_count);
  }
  pp->sleeps_heap =
      ph_create(config_resend_count, next_send_less_than, pp->sleeps);
  if (pp->sleeps_heap == NULL)
    signal_error_by_blinking();
  return pp;
}

void add_press(press_pile_t *pp, press_t *press) {
  int sc = press->send_count;
  if (sc >= config_resend_count)
    signal_error_by_blinking();
  bool was_empty = queue_is_empty(&pp->presses[sc]);
  /* No error check; blithely continue if the queue was full. */
  queue_try_add(&pp->presses[sc], press);
  if (was_empty) {
    pheap_node_id_t i = ph_new_node(pp->sleeps_heap);
    pp->sleeps[i] = &pp->presses[sc];
    ph_insert_node(pp->sleeps_heap, i);
  }
}

int32_t next_scheduled_send(press_pile_t *pp) {
  pheap_node_id_t i = ph_peek_head(pp->sleeps_heap);
  if (i == 0)
    return -1;
  return press_queue_next_send(pp->sleeps[i]);
}

bool get_press_due_for_resend(press_pile_t *pp, uint32_t now, press_t *press) {
  pheap_node_id_t i = ph_peek_head(pp->sleeps_heap);
  if (i == 0 || press_queue_next_send(pp->sleeps[i]) > now)
    return false;
  if (!queue_try_remove(pp->sleeps[i], press))
    signal_error_by_blinking();
  bool became_empty = queue_is_empty(pp->sleeps[i]);
  if (ph_remove_head(pp->sleeps_heap, became_empty) != i)
    signal_error_by_blinking();
  if (became_empty) {
    pp->sleeps[i] = NULL;
  } else {
    ph_insert_node(pp->sleeps_heap, i);
  }
  return true;
}