split the working code out from the XS file

[poe-xs-queue-array.git] / queue.c

#include "EXTERN.h"\r
#include "perl.h"\r
#include "XSUB.h"\r
\r
#include "queue.h"\r
\r
#define DEBUG(x) x\r
/*#define DEBUG(x)*/\r
\r
#define PQ_START_SIZE 10\r
#define AT_START 0\r
#define AT_END 1\r
\r
pq_id_t queue_seq;\r
\r
/*\r
We store the queue in a similar way to the way perl deals with arrays,\r
we keep a block of memory, but the first element may or may not be in use,\r
depending on the pattern of usage.\r
\r
There's 3 value controlling usage of the array:\r
\r
  - alloc - the number of elements allocated in total\r
  - start - the first element in use in the array\r
  - end - one past the end of the last element in the array\r
\r
This has the properties that:\r
\r
  start == 0 - no space at the front\r
  end == alloc - no space at the end\r
  end - start - number of elements in the queue\r
\r
We use a perl hash (HV *) to store the mapping from ids to priorities.\r
\r
*/\r
struct poe_queue_tag {\r
  /* the first entry in use */\r
  int start;\r
\r
  /* 1 past the last entry in use, hence end - start is the number of \r
     entries in the queue */\r
  int end;\r
\r
  /* the total number of entries allocated */\r
  int alloc;\r
\r
  /* used to generate item ids */\r
  pq_id_t queue_seq;\r
\r
  /* used to track in use item ids */\r
  HV *ids;\r
\r
  /* the actual entries */\r
  pq_entry *entries;\r
};\r
\r
/*\r
poe_create - create a new queue object.\r
\r
No parameters.  returns the new queue object.\r
\r
*/\r
poe_queue *\r
pq_create(void) {\r
  poe_queue *pq = malloc(sizeof(poe_queue));\r
  \r
  if (pq == NULL)\r
    croak("Out of memory");\r
  pq->start = 0;\r
  pq->end = 0;\r
  pq->alloc = PQ_START_SIZE;\r
  pq->queue_seq = 0;\r
  pq->ids = newHV();\r
  pq->entries = calloc(sizeof(pq_entry), PQ_START_SIZE);\r
  if (pq->entries == NULL)\r
    croak("Out of memory");\r
\r
  DEBUG( fprintf(stderr, "pq_create() => %p\n", pq) );\r
\r
  return pq;\r
}\r
\r
/*\r
pq_delete - release the queue object.\r
\r
This also releases one reference from each SV in the queue.\r
\r
*/\r
void\r
pq_delete(poe_queue *pq) {\r
  int i;\r
\r
  DEBUG( fprintf(stderr, "pq_delete(%p)\n", pq) );\r
  if (pq->end > pq->start) {\r
    for (i = pq->start; i < pq->end; ++i) {\r
      SvREFCNT_dec(pq->entries[i].payload);\r
    }\r
  }\r
  SvREFCNT_dec((SV *)pq->ids);\r
  pq->ids = NULL;\r
  if (pq->entries)\r
    free(pq->entries);\r
  pq->entries = NULL;\r
  free(pq);\r
}\r
\r
/*\r
pq_new_id - generate a new item id.\r
\r
Internal use only.\r
\r
This, the following 3 functions and pq_create, pq_delete, should be\r
all that needs to be modified if we change hash implementations.\r
\r
*/\r
static\r
pq_id_t\r
pq_new_id(poe_queue *pq, pq_priority_t priority) {\r
  int seq = ++pq->queue_seq;\r
  SV *index = newSViv(seq);\r
\r
  while (hv_exists_ent(pq->ids, index, 0)) {\r
    seq = ++pq->queue_seq;\r
    sv_setiv(index, seq);\r
  }\r
  hv_store_ent(pq->ids, index, newSVnv(priority), 0);\r
\r
  return seq;\r
}\r
\r
/*\r
pq_release_id - releases an id for future use.\r
*/\r
static\r
void\r
pq_release_id(poe_queue *pq, pq_id_t id) {\r
  SV *id_sv = sv_2mortal(newSViv(id));\r
\r
  hv_delete_ent(pq->ids, id_sv, 0, 0);\r
}\r
\r
/*\r
pq_item_priority - get the priority of an item given it's id\r
*/\r
static\r
int\r
pq_item_priority(poe_queue *pq, pq_id_t id, pq_priority_t *priority) {\r
  HE *entry = hv_fetch_ent(pq->ids, sv_2mortal(newSViv(id)), 0, 0);\r
\r
  if (!entry)\r
    return 0;\r
\r
  *priority = SvNV(HeVAL(entry));\r
\r
  return 1;\r
}\r
\r
/*\r
pq_set_id_priority - set the priority of an item in the id hash\r
*/\r
static\r
void\r
pq_set_id_priority(poe_queue *pq, pq_id_t id, pq_priority_t new_priority) {\r
  HE *entry = hv_fetch_ent(pq->ids, sv_2mortal(newSViv(id)), 0, 0);\r
\r
  if (!entry)\r
    croak("pq_set_priority: id not found");\r
\r
  sv_setnv(HeVAL(entry), new_priority);\r
}\r
\r
/*\r
pq_realloc - make space at the front of back of the queue.\r
\r
This adjusts the queue to allow insertion of a single item at the\r
front or the back of the queue.\r
\r
If the queue has 33% or more space available we simple adjust the\r
position of the in-use items within the array.  We try not to push the\r
items right up against the opposite end of the array, since we might\r
need to insert items there too.\r
\r
If the queue has less than 33% space available we allocate another 50%\r
space.  We then only move the queue elements if we need space at the\r
front, since the reallocation has just opened up a huge space at the\r
back.  Since we're reallocating exponentially larger sizes we should\r
have a constant time cost on reallocation per queue item stored (but\r
other costs are going to be higher.)\r
\r
*/\r
static\r
void\r
pq_realloc(poe_queue *pq, int at_end) {\r
  int count = pq->end - pq->start;\r
\r
  DEBUG( fprintf(stderr, "pq_realloc((%d, %d, %d), %d)\n", pq->start, pq->end, pq->alloc, at_end) );\r
  pq_dump(pq);\r
  if (count * 3 / 2 < pq->alloc) {\r
    /* 33 % or more space available, use some of it */\r
    int new_start;\r
\r
    if (at_end) {\r
      new_start = (pq->alloc - count) / 3;\r
    }\r
    else {\r
      new_start = (pq->alloc - count) * 2 / 3;\r
    }\r
    DEBUG( fprintf(stderr, "  moving start to %d\n", new_start) );\r
    memmove(pq->entries + new_start, pq->entries + pq->start,\r
	    count * sizeof(pq_entry));\r
    pq->start = new_start;\r
    pq->end = new_start + count;\r
  }\r
  else {\r
    int new_alloc = pq->alloc * 3 / 2;\r
    pq->entries = realloc(pq->entries, sizeof(pq_entry) * new_alloc);\r
    pq->alloc = new_alloc;\r
\r
    if (!pq->entries)\r
      croak("Out of memory");\r
\r
    DEBUG( fprintf(stderr, "  - expanding to %d entries\n", new_alloc) );\r
\r
    if (!at_end) {\r
      int new_start = (new_alloc - count) * 2 / 3;\r
      DEBUG( fprintf(stderr, "  moving start to %d\n", new_start) );\r
      memmove(pq->entries + new_start, pq->entries + pq->start,\r
	      count * sizeof(pq_entry));\r
      pq->start = new_start;\r
      pq->end = new_start + count;\r
    }\r
  }\r
  pq_dump(pq);\r
  DEBUG( fprintf(stderr, "  final: %d %d %d\n", pq->start, pq->end, pq->alloc) );\r
}\r
\r
/*\r
pq_insertion_point - figure out where to insert an item with the given\r
priority\r
\r
Internal.\r
*/\r
static\r
int\r
pq_insertion_point(poe_queue *pq, pq_priority_t priority) {\r
  /* for now this is just a linear search, later we should make it \r
     binary */\r
  int i = pq->end;\r
  while (i > pq->start &&\r
         priority < pq->entries[i-1].priority) {\r
    --i;\r
  }\r
\r
  return i;\r
}\r
\r
int\r
pq_enqueue(poe_queue *pq, pq_priority_t priority, SV *payload) {\r
  int fill_at;\r
  pq_id_t id = pq_new_id(pq, priority);\r
\r
  DEBUG( fprintf(stderr, "pq_enqueue(%f, %p)\n", priority, payload) );\r
  if (pq->start == pq->end) {\r
    DEBUG( fprintf(stderr, "  - on empty queue\n") );\r
    /* allow room at front and back for new entries */\r
    pq->start = pq->alloc / 3;\r
    pq->end = pq->start + 1;\r
    fill_at = pq->start;\r
  }\r
  else if (priority >= pq->entries[pq->end-1].priority) {\r
    DEBUG( fprintf(stderr, "  - at the end\n") );\r
    if (pq->end == pq->alloc)\r
      /* past the end - need to realloc or make some space */\r
      pq_realloc(pq, AT_END);\r
    \r
    fill_at = pq->end;\r
    ++pq->end;\r
  }\r
  else if (priority < pq->entries[pq->start].priority) {\r
    DEBUG( fprintf(stderr, "  - at the front\n") );\r
    if (pq->start == 0)\r
      /* no space at the front, make some */\r
      pq_realloc(pq, AT_START);\r
\r
    --pq->start;\r
    fill_at = pq->start;\r
  }\r
  else {\r
    int i;\r
    DEBUG( fprintf(stderr, "  - in the middle\n") );\r
    i = pq_insertion_point(pq, priority);\r
    \r
    /* if we're near the end we want to push entries up, otherwise down */\r
    if (i - pq->start > (pq->end - pq->start) / 2) {\r
      DEBUG( fprintf(stderr, "    - closer to the back (%d -> [ %d %d ])\n",\r
                     i, pq->start, pq->end) );\r
      /* make sure we have space, this might end up copying twice, \r
	 but too bad for now */\r
      if (pq->end == pq->alloc) {\r
        int old_start = pq->start;\r
	pq_realloc(pq, AT_END);\r
        i += pq->start - old_start;\r
      }\r
      \r
      memmove(pq->entries + i + 1, pq->entries + i, (pq->end - i) * sizeof(pq_entry));\r
      ++pq->end;\r
      fill_at = i;\r
    }\r
    else {\r
      DEBUG( fprintf(stderr, "    - closer to the front (%d -> [ %d %d ])\n",\r
                     i, pq->start, pq->end) );\r
      if (pq->start == 0) {\r
	pq_realloc(pq, AT_START);\r
	i += pq->start;\r
      }\r
      memmove(pq->entries + pq->start - 1, pq->entries + pq->start,\r
	     (i - pq->start) * sizeof(pq_entry));\r
      --pq->start;\r
      fill_at = i-1;\r
    }\r
  }\r
  pq->entries[fill_at].priority = priority;\r
  pq->entries[fill_at].id = id;\r
  pq->entries[fill_at].payload = newSVsv(payload);\r
\r
  return id;\r
}\r
\r
/*\r
  Note: it's up to the caller to release the SV.  The XS code does this \r
  by making it mortal.\r
*/\r
int\r
pq_dequeue_next(poe_queue *pq, pq_priority_t *priority, pq_id_t *id, SV **payload) {\r
  pq_entry *entry;\r
  /* the caller needs to release the payload (somehow) */\r
  if (pq->start == pq->end)\r
    return 0;\r
\r
  entry = pq->entries + pq->start++;\r
  *priority = entry->priority;\r
  *id = entry->id;\r
  *payload = entry->payload;\r
  pq_release_id(pq, entry->id);\r
\r
  return 1;\r
}\r
\r
int\r
pq_get_next_priority(poe_queue *pq, pq_priority_t *priority) {\r
  if (pq->start == pq->end)\r
    return 0;\r
\r
  *priority = pq->entries[pq->start].priority;\r
  return 1;\r
}\r
\r
int\r
pq_get_item_count(poe_queue *pq) {\r
  return pq->end - pq->start;\r
}\r
\r
/*\r
pq_test_filter - the XS magic involved in passing the payload to a\r
filter function.\r
*/\r
static\r
int\r
pq_test_filter(pq_entry *entry, SV *filter) {\r
  /* man perlcall for the magic here */\r
  dSP;\r
  int count;\r
  SV *result_sv;\r
  int result;\r
\r
  ENTER;\r
  SAVETMPS;\r
  PUSHMARK(SP);\r
  XPUSHs(sv_2mortal(newSVsv(entry->payload)));\r
  PUTBACK;\r
\r
  count = call_sv(filter, G_SCALAR);\r
\r
  SPAGAIN;\r
\r
  if (count != 1) \r
    croak("got other than 1 value in scalar context");\r
\r
  result_sv = POPs;\r
  result = SvTRUE(result_sv);\r
\r
  PUTBACK;\r
  FREETMPS;\r
  LEAVE;\r
\r
  return result;\r
}\r
\r
/*\r
pq_find_item - search for an item we know is there.\r
\r
Internal.\r
*/\r
static\r
int\r
pq_find_item(poe_queue *pq, pq_id_t id, pq_priority_t priority) {\r
  int i;\r
\r
  for (i = pq->start; i < pq->end; ++i) {\r
    if (pq->entries[i].id == id)\r
      return i;\r
  }\r
  DEBUG(fprintf(stderr, "pq_find_item %d => %f\n", id, priority) );\r
  croak("Internal inconsistency: event should have been found");\r
}\r
\r
int\r
pq_remove_item(poe_queue *pq, pq_id_t id, SV *filter, pq_entry *removed) {\r
  pq_priority_t priority;\r
  int index;\r
\r
  if (!pq_item_priority(pq, id, &priority)) {\r
    errno = ESRCH;\r
    return 0;\r
  }\r
\r
  index = pq_find_item(pq, id, priority);\r
\r
  if (!pq_test_filter(pq->entries + index, filter)) {\r
    errno = EPERM;\r
    return 0;\r
  }\r
\r
  *removed = pq->entries[index];\r
  pq_release_id(pq, id);\r
  if (index == pq->start) {\r
    ++pq->start;\r
  }\r
  else if (index == pq->end - 1) {\r
    --pq->end;\r
  }\r
  else {\r
    memmove(pq->entries + index, pq->entries + index + 1,\r
	    sizeof(pq_entry) * (pq->end - index - 1));\r
    --pq->end;\r
  }\r
\r
  return 1;\r
}\r
\r
int\r
pq_remove_items(poe_queue *pq, SV *filter, int max_count, pq_entry **entries) {\r
  int in_index, out_index;\r
  int remove_count = 0;\r
  \r
  *entries = NULL;\r
  if (pq->start == pq->end)\r
    return 0;\r
\r
  *entries = malloc(sizeof(pq_entry) * (pq->end - pq->start));\r
  if (!*entries)\r
    croak("Out of memory");\r
  \r
  in_index = out_index = pq->start;\r
  while (in_index < pq->end && remove_count < max_count) {\r
    if (pq_test_filter(pq->entries + in_index, filter)) {\r
      pq_release_id(pq, pq->entries[in_index].id);\r
      (*entries)[remove_count++] = pq->entries[in_index++];\r
    }\r
    else {\r
      pq->entries[out_index++] = pq->entries[in_index++];\r
    }\r
  }\r
  while (in_index < pq->end) {\r
    pq->entries[out_index++] = pq->entries[in_index++];\r
  }\r
  pq->end = out_index;\r
  \r
  return remove_count;\r
}\r
\r
/*\r
We need to keep the following 2 functions in sync (or combine the\r
common code.)\r
*/\r
int\r
pq_set_priority(poe_queue *pq, pq_id_t id, SV *filter, pq_priority_t new_priority) {\r
  pq_priority_t old_priority;\r
  int index, insert_at;\r
\r
  if (!pq_item_priority(pq, id, &old_priority)) {\r
    errno = ESRCH;\r
    return 0;\r
  }\r
\r
  index = pq_find_item(pq, id, old_priority);\r
\r
  if (!pq_test_filter(pq->entries + index, filter)) {\r
    errno = EPERM;\r
    return 0;\r
  }\r
\r
  DEBUG( fprintf(stderr, " - index %d  oldp %f newp %f\n", index, old_priority, new_priority) );\r
\r
  if (pq->end - pq->start == 1) {\r
    DEBUG( fprintf(stderr, "   -- one item\n") );\r
    /* only the one item anyway */\r
    pq->entries[pq->start].priority = new_priority;\r
  }\r
  else {\r
    insert_at = pq_insertion_point(pq, new_priority);\r
    DEBUG( fprintf(stderr, "   - new index %d\n", insert_at) );\r
    /* the item is still in the queue, so either side of it means it\r
       won't move */\r
    if (insert_at == index || insert_at == index+1) {\r
      DEBUG( fprintf(stderr, "   -- change in place\n") );\r
      pq->entries[index].priority = new_priority;\r
    }\r
    else {\r
      pq_entry saved = pq->entries[index];\r
      saved.priority = new_priority;\r
\r
      if (insert_at < index) {\r
        DEBUG( fprintf(stderr, "  - insert_at < index\n") );\r
        memmove(pq->entries + insert_at + 1, pq->entries + insert_at,\r
	        sizeof(pq_entry) * (index - insert_at));\r
        pq->entries[insert_at] = saved;\r
      }\r
      else {\r
        DEBUG( fprintf(stderr, "  - insert_at > index\n") );\r
	--insert_at;\r
	memmove(pq->entries + index, pq->entries + index + 1,\r
	        sizeof(pq_entry) * (insert_at - index));\r
        pq->entries[insert_at] = saved;\r
      }\r
    }\r
  }\r
\r
  pq_set_id_priority(pq, id, new_priority);\r
\r
  return 1;  \r
}\r
\r
int\r
pq_adjust_priority(poe_queue *pq, pq_id_t id, SV *filter, double delta, pq_priority_t *priority) {\r
  pq_priority_t old_priority, new_priority;\r
  int index, insert_at;\r
\r
  DEBUG( fprintf(stderr, "pq_adjust_priority(..., %d, %p, %f, ...)\n", id, filter, delta) );\r
\r
  if (!pq_item_priority(pq, id, &old_priority)) {\r
    errno = ESRCH;\r
    return 0;\r
  }\r
\r
  index = pq_find_item(pq, id, old_priority);\r
\r
  if (!pq_test_filter(pq->entries + index, filter)) {\r
    errno = EPERM;\r
    return 0;\r
  }\r
\r
  new_priority = old_priority + delta;\r
\r
  DEBUG( fprintf(stderr, " - index %d  oldp %f newp %f\n", index, old_priority, new_priority) );\r
\r
  if (pq->end - pq->start == 1) {\r
    DEBUG( fprintf(stderr, "   -- one item\n") );\r
    /* only the one item anyway */\r
    pq->entries[pq->start].priority = new_priority;\r
  }\r
  else {\r
    insert_at = pq_insertion_point(pq, new_priority);\r
    DEBUG( fprintf(stderr, "   - new index %d\n", insert_at) );\r
    /* the item is still in the queue, so either side of it means it\r
       won't move */\r
    if (insert_at == index || insert_at == index+1) {\r
      DEBUG( fprintf(stderr, "   -- change in place\n") );\r
      pq->entries[index].priority = new_priority;\r
    }\r
    else {\r
      pq_entry saved = pq->entries[index];\r
      saved.priority = new_priority;\r
\r
      if (insert_at < index) {\r
        DEBUG( fprintf(stderr, "  - insert_at < index\n") );\r
        memmove(pq->entries + insert_at + 1, pq->entries + insert_at,\r
	        sizeof(pq_entry) * (index - insert_at));\r
        pq->entries[insert_at] = saved;\r
      }\r
      else {\r
        DEBUG( fprintf(stderr, "  - insert_at > index\n") );\r
	--insert_at;\r
	memmove(pq->entries + index, pq->entries + index + 1,\r
	        sizeof(pq_entry) * (insert_at - index));\r
        pq->entries[insert_at] = saved;\r
      }\r
    }\r
  }\r
\r
  pq_set_id_priority(pq, id, new_priority);\r
  *priority = new_priority;\r
\r
  return 1;  \r
}\r
\r
int\r
pq_peek_items(poe_queue *pq, SV *filter, int max_count, pq_entry **items) {\r
  int count = 0;\r
  int i;\r
\r
  *items = NULL;\r
  if (pq->end == pq->start)\r
    return 0;\r
\r
  *items = malloc(sizeof(pq_entry) * (pq->end - pq->start));\r
  for (i = pq->start; i < pq->end; ++i) {\r
    if (pq_test_filter(pq->entries + i, filter)) {\r
      (*items)[count++] = pq->entries[i];\r
    }\r
  }\r
  if (!count) {\r
    free(*items);\r
    *items = NULL;\r
  }\r
\r
  return count;\r
}\r
\r
/*\r
pq_dump - dump the internals of the queue structure.\r
*/\r
void\r
pq_dump(poe_queue *pq) {\r
  int i;\r
  HE *he;\r
\r
  fprintf(stderr, "poe_queue\n");\r
  fprintf(stderr, "  start: %d\n", pq->start);\r
  fprintf(stderr, "    end: %d\n", pq->end);\r
  fprintf(stderr, "  alloc: %d\n", pq->alloc);\r
  fprintf(stderr, "    seq: %d\n", pq->queue_seq);\r
  fprintf(stderr, "  **Queue Entries:\n"\r
         "      index:   id  priority    SV\n");\r
  for (i = pq->start; i < pq->end; ++i) {\r
    pq_entry *entry = pq->entries + i;\r
    fprintf(stderr, "      %5d: %5d %8f  %p (%u)\n", i, entry->id, entry->priority,\r
	   entry->payload, (unsigned)SvREFCNT(entry->payload));\r
  }\r
  fprintf(stderr, "  **Hash entries:\n");\r
  hv_iterinit(pq->ids);\r
  while ((he = hv_iternext(pq->ids)) != NULL) {\r
    STRLEN len;\r
    fprintf(stderr, "   %s => %f\n", HePV(he, len), SvNV(hv_iterval(pq->ids, he)));\r
  }\r
}\r