<commit_before>#include <nchan_module.h>
#include "rbtree_util.h"
#include "spool.h"
#include <assert.h>

#define DEBUG_LEVEL NGX_LOG_DEBUG
//#define DEBUG_LEVEL NGX_LOG_WARN

#define DBG(fmt, arg...) ngx_log_error(DEBUG_LEVEL, ngx_cycle->log, 0, "SPOOL:" fmt, ##arg)
#define ERR(fmt, arg...) ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, "SPOOL:" fmt, ##arg)

//////// SPOOLs -- Subscriber Pools  /////////

static ngx_int_t spool_remove_subscriber(subscriber_pool_t *, spooled_subscriber_t *);
static void spool_bubbleup_dequeue_handler(subscriber_pool_t *spool, subscriber_t *sub, channel_spooler_t *spl);
//static void spool_bubbleup_bulk_dequeue_handler(subscriber_pool_t *spool, subscriber_type_t type, ngx_int_t count, channel_spooler_t *spl);
static ngx_int_t spool_respond_general(subscriber_pool_t *self, nchan_msg_t *msg, ngx_int_t status_code, const ngx_str_t *status_line);
static ngx_int_t spool_transfer_subscribers(subscriber_pool_t *spool, subscriber_pool_t *newspool, uint8_t update_subscriber_last_msgid);
static ngx_int_t destroy_spool(subscriber_pool_t *spool);
static ngx_int_t remove_spool(subscriber_pool_t *spool);
static ngx_int_t spool_fetch_msg(subscriber_pool_t *spool);

static nchan_msg_id_t     latest_msg_id = NCHAN_NEWEST_MSGID;

static subscriber_pool_t *find_spool(channel_spooler_t *spl, nchan_msg_id_t *id) {
  rbtree_seed_t      *seed = &spl->spoolseed;
  ngx_rbtree_node_t  *node;
  
  if((node = rbtree_find_node(seed, id)) != NULL) {
    return (subscriber_pool_t *)rbtree_data_from_node(node);
  }
  else {
    return NULL;
  }
}

static int msg_ids_equal(nchan_msg_id_t *id1, nchan_msg_id_t *id2) {
  int           i, max;
  int16_t      *tags1, *tags2;
  
  if(id1->time != id2->time || id1->tagcount != id2->tagcount) return 0;
  max = id1->tagcount;
  if(max <= NCHAN_FIXED_MULTITAG_MAX) {
    tags1 = id1->tag.fixed;
    tags2 = id2->tag.fixed;
  }
  else {
    tags1 = id1->tag.allocd;
    tags2 = id2->tag.allocd;
  }
  
  
  for(i=0; i < max; i++) {
    if(tags1[i] != tags2[i]) return 0;
  }
  return 1;
}

static ngx_inline void init_spool(channel_spooler_t *spl, subscriber_pool_t *spool, nchan_msg_id_t *id) {
  nchan_copy_new_msg_id(&spool->id, id);
  spool->msg = NULL;
  spool->msg_status = MSG_INVALID;
  
  spool->first = NULL;
  spool->pool = NULL;
  spool->sub_count = 0;
  spool->non_internal_sub_count = 0;
  spool->generation = 0;
  spool->responded_count = 0;
  
  spool->spooler = spl;
}

static subscriber_pool_t *get_spool(channel_spooler_t *spl, nchan_msg_id_t *id) {
  rbtree_seed_t      *seed = &spl->spoolseed;
  ngx_rbtree_node_t  *node;
  subscriber_pool_t *spool;
  
  if(id->time == -1) {
    spool = &spl->current_msg_spool;
    spool->msg_status = MSG_EXPECTED;
    return &spl->current_msg_spool;
  }
  
  if((node = rbtree_find_node(seed, id)) == NULL) {
    
    if((node = rbtree_create_node(seed, sizeof(*spool))) == NULL) {
      ERR("can't create rbtree node for spool");
      return NULL;
    }
    
   // DBG("CREATED spool node %p for msgid %V", node, msgid_to_str(id));
    spool = (subscriber_pool_t *)rbtree_data_from_node(node);
    
    init_spool(spl, spool, id);
    
    if(rbtree_insert_node(seed, node) != NGX_OK) {
      ERR("couldn't insert spool node");
      rbtree_destroy_node(seed, node);
      return NULL;
    }
  }
  else {
    spool = (subscriber_pool_t *)rbtree_data_from_node(node);
    DBG("found spool node %p with msgid %V", node, msgid_to_str(id));
    assert(spool->id.time == id->time);
  }
  return spool;
}

static ngx_int_t spool_nextmsg(subscriber_pool_t *spool, nchan_msg_id_t *new_last_id) {
  subscriber_pool_t      *newspool;
  channel_spooler_t      *spl = spool->spooler;
  
  ngx_int_t               immortal_spool = spool->id.time == -1;
  int16_t                 largetags[NCHAN_MULTITAG_MAX];
  nchan_msg_id_t          new_id = NCHAN_ZERO_MSGID;
    
  nchan_copy_msg_id(&new_id, &spool->id, largetags);
  nchan_update_multi_msgid(&new_id, new_last_id);
  
  //ERR("spool %p nextmsg (%V) --", spool, msgid_to_str(&spool->id));
  //ERR(" --  update with               (%V) --", msgid_to_str(new_last_id));
  //ERR(" -- newid                       %V", msgid_to_str(&new_id));
  
  if(msg_ids_equal(&spool->id, &new_id)) {
    ERR("nextmsg id same as curmsg (%V)", msgid_to_str(&spool->id));
    assert(0);
  }
  else {
    newspool = !immortal_spool ? find_spool(spl, &new_id) : get_spool(spl, &new_id);
    
    if(newspool != NULL) {
      assert(spool != newspool);
      spool_transfer_subscribers(spool, newspool, 0);
      if(!immortal_spool) destroy_spool(spool);
    }
    else {
      ngx_rbtree_node_t       *node;
      assert(!immortal_spool);
      node = rbtree_node_from_data(spool);
      rbtree_remove_node(&spl->spoolseed, node);
      nchan_copy_new_msg_id(&spool->id, &new_id);
      rbtree_insert_node(&spl->spoolseed, node);
      spool->msg_status = MSG_INVALID;
      spool->msg = NULL;
      newspool = spool;
      
      /*
      newspool = get_spool(spl, &new_id);
      assert(spool != newspool);
      spool_transfer_subscribers(spool, newspool, 0);
      destroy_spool(spool);
      */
    }

    
    if(newspool->non_internal_sub_count > 0 && spl->bulk_post_subscribe_handler != NULL) {
      spl->bulk_post_subscribe_handler(spl, newspool->non_internal_sub_count, spl->bulk_post_subscribe_privdata);
    }
    
    if(newspool->sub_count > 0) {
      switch(newspool->msg_status) {
        case MSG_INVALID:
          spool_fetch_msg(newspool);
          break;
        case MSG_EXPECTED:
          spool_respond_general(newspool, NULL, NGX_HTTP_NO_CONTENT, NULL);
          break;
      }
    }
  }
  return NGX_OK;
}
typedef struct {
  channel_spooler_t   *spooler;
  nchan_msg_id_t       msgid;
} fetchmsg_data_t;

static ngx_int_t spool_fetch_msg_callback(nchan_msg_status_t findmsg_status, nchan_msg_t *msg, fetchmsg_data_t *data) {
  nchan_msg_id_t        anymsg;
  anymsg.time = 0;
  anymsg.tag.fixed[0] = 0;
  anymsg.tagcount = 1;
  
  subscriber_pool_t    *spool, *nuspool;
  <commit_msg>compiler warning no warn! (thanks for noticing @vtslothy)<commit_after>#include <nchan_module.h>
#include "rbtree_util.h"
#include "spool.h"
#include <assert.h>

#define DEBUG_LEVEL NGX_LOG_DEBUG
//#define DEBUG_LEVEL NGX_LOG_WARN

#define DBG(fmt, arg...) ngx_log_error(DEBUG_LEVEL, ngx_cycle->log, 0, "SPOOL:" fmt, ##arg)
#define ERR(fmt, arg...) ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, "SPOOL:" fmt, ##arg)

//////// SPOOLs -- Subscriber Pools  /////////

static ngx_int_t spool_remove_subscriber(subscriber_pool_t *, spooled_subscriber_t *);
static void spool_bubbleup_dequeue_handler(subscriber_pool_t *spool, subscriber_t *sub, channel_spooler_t *spl);
//static void spool_bubbleup_bulk_dequeue_handler(subscriber_pool_t *spool, subscriber_type_t type, ngx_int_t count, channel_spooler_t *spl);
static ngx_int_t spool_respond_general(subscriber_pool_t *self, nchan_msg_t *msg, ngx_int_t status_code, const ngx_str_t *status_line);
static ngx_int_t spool_transfer_subscribers(subscriber_pool_t *spool, subscriber_pool_t *newspool, uint8_t update_subscriber_last_msgid);
static ngx_int_t destroy_spool(subscriber_pool_t *spool);
static ngx_int_t remove_spool(subscriber_pool_t *spool);
static ngx_int_t spool_fetch_msg(subscriber_pool_t *spool);

static nchan_msg_id_t     latest_msg_id = NCHAN_NEWEST_MSGID;

static subscriber_pool_t *find_spool(channel_spooler_t *spl, nchan_msg_id_t *id) {
  rbtree_seed_t      *seed = &spl->spoolseed;
  ngx_rbtree_node_t  *node;
  
  if((node = rbtree_find_node(seed, id)) != NULL) {
    return (subscriber_pool_t *)rbtree_data_from_node(node);
  }
  else {
    return NULL;
  }
}

static int msg_ids_equal(nchan_msg_id_t *id1, nchan_msg_id_t *id2) {
  int           i, max;
  int16_t      *tags1, *tags2;
  
  if(id1->time != id2->time || id1->tagcount != id2->tagcount) return 0;
  max = id1->tagcount;
  if(max <= NCHAN_FIXED_MULTITAG_MAX) {
    tags1 = id1->tag.fixed;
    tags2 = id2->tag.fixed;
  }
  else {
    tags1 = id1->tag.allocd;
    tags2 = id2->tag.allocd;
  }
  
  
  for(i=0; i < max; i++) {
    if(tags1[i] != tags2[i]) return 0;
  }
  return 1;
}

static ngx_inline void init_spool(channel_spooler_t *spl, subscriber_pool_t *spool, nchan_msg_id_t *id) {
  nchan_copy_new_msg_id(&spool->id, id);
  spool->msg = NULL;
  spool->msg_status = MSG_INVALID;
  
  spool->first = NULL;
  spool->pool = NULL;
  spool->sub_count = 0;
  spool->non_internal_sub_count = 0;
  spool->generation = 0;
  spool->responded_count = 0;
  
  spool->spooler = spl;
}

static subscriber_pool_t *get_spool(channel_spooler_t *spl, nchan_msg_id_t *id) {
  rbtree_seed_t      *seed = &spl->spoolseed;
  ngx_rbtree_node_t  *node;
  subscriber_pool_t *spool;
  
  if(id->time == -1) {
    spool = &spl->current_msg_spool;
    spool->msg_status = MSG_EXPECTED;
    return &spl->current_msg_spool;
  }
  
  if((node = rbtree_find_node(seed, id)) == NULL) {
    
    if((node = rbtree_create_node(seed, sizeof(*spool))) == NULL) {
      ERR("can't create rbtree node for spool");
      return NULL;
    }
    
   // DBG("CREATED spool node %p for msgid %V", node, msgid_to_str(id));
    spool = (subscriber_pool_t *)rbtree_data_from_node(node);
    
    init_spool(spl, spool, id);
    
    if(rbtree_insert_node(seed, node) != NGX_OK) {
      ERR("couldn't insert spool node");
      rbtree_destroy_node(seed, node);
      return NULL;
    }
  }
  else {
    spool = (subscriber_pool_t *)rbtree_data_from_node(node);
    DBG("found spool node %p with msgid %V", node, msgid_to_str(id));
    assert(spool->id.time == id->time);
  }
  return spool;
}

static ngx_int_t spool_nextmsg(subscriber_pool_t *spool, nchan_msg_id_t *new_last_id) {
  subscriber_pool_t      *newspool;
  channel_spooler_t      *spl = spool->spooler;
  
  ngx_int_t               immortal_spool = spool->id.time == -1;
  int16_t                 largetags[NCHAN_MULTITAG_MAX];
  nchan_msg_id_t          new_id = NCHAN_ZERO_MSGID;
    
  nchan_copy_msg_id(&new_id, &spool->id, largetags);
  nchan_update_multi_msgid(&new_id, new_last_id);
  
  //ERR("spool %p nextmsg (%V) --", spool, msgid_to_str(&spool->id));
  //ERR(" --  update with               (%V) --", msgid_to_str(new_last_id));
  //ERR(" -- newid                       %V", msgid_to_str(&new_id));
  
  if(msg_ids_equal(&spool->id, &new_id)) {
    ERR("nextmsg id same as curmsg (%V)", msgid_to_str(&spool->id));
    assert(0);
  }
  else {
    newspool = !immortal_spool ? find_spool(spl, &new_id) : get_spool(spl, &new_id);
    
    if(newspool != NULL) {
      assert(spool != newspool);
      spool_transfer_subscribers(spool, newspool, 0);
      if(!immortal_spool) destroy_spool(spool);
    }
    else {
      ngx_rbtree_node_t       *node;
      assert(!immortal_spool);
      node = rbtree_node_from_data(spool);
      rbtree_remove_node(&spl->spoolseed, node);
      nchan_copy_new_msg_id(&spool->id, &new_id);
      rbtree_insert_node(&spl->spoolseed, node);
      spool->msg_status = MSG_INVALID;
      spool->msg = NULL;
      newspool = spool;
      
      /*
      newspool = get_spool(spl, &new_id);
      assert(spool != newspool);
      spool_transfer_subscribers(spool, newspool, 0);
      destroy_spool(spool);
      */
    }

    
    if(newspool->non_internal_sub_count > 0 && spl->bulk_post_subscribe_handler != NULL) {
      spl->bulk_post_subscribe_handler(spl, newspool->non_internal_sub_count, spl->bulk_post_subscribe_privdata);
    }
    
    if(newspool->sub_count > 0) {
      switch(newspool->msg_status) {
        case MSG_INVALID:
          spool_fetch_msg(newspool);
          break;
        case MSG_EXPECTED:
          spool_respond_general(newspool, NULL, NGX_HTTP_NO_CONTENT, NULL);
          break;
        default:
          break;
      }
    }
  }
  return NGX_OK;
}

typedef struct {
  channel_spooler_t   *spooler;
  nchan_msg_id_t       msgid;
} fetchmsg_data_t;

static ngx_int_t spool_fetch_msg_callback(nchan_msg_status_t findmsg_status, nchan_msg_t *msg, fetchmsg_data_t *data) {
  nchan_msg_id_t        anymsg;
  anymsg.time = 0;
  anymsg.tag.fixed[0] = 0;
  anymsg.tagcount = 1;
  
  subscriber_pool_t    *spool, *nuspool;
  
{'fraction_non_alphanumeric': 0.12535845964768538, 'fraction_numerical': 0.007374027038099139, 'mean_word_length': 3.559581621217781, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 105, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1631249', 'n_tokens_mistral': 5042, 'n_tokens_neox': 4946, 'n_words': 819}