<commit_before>#include <ngx_http_push_module.h>
#include <ngx_channel.h>
#include "shmem.h"
#include "ipc.h"
#include "ipc-handlers.h"
#include "store-private.h"
#include <assert.h>
#include "../../subscribers/memstore.h"

#define IPC_SUBSCRIBE               1
#define IPC_SUBSCRIBE_REPLY         2
#define IPC_UNSUBSCRIBE             3  //NOT USED
#define IPC_UNSUBSCRIBED            4
#define IPC_PUBLISH_MESSAGE         5
#define IPC_PUBLISH_MESSAGE_REPLY   6
#define IPC_PUBLISH_STATUS          7
#define IPC_GET_MESSAGE             8
#define IPC_GET_MESSAGE_REPLY       9
#define IPC_DELETE                  10
#define IPC_DELETE_REPLY            11
#define IPC_GET_CHANNEL_INFO        12
#define IPC_GET_CHANNEL_INFO_REPLY  13
#define IPC_SUBSCRIBER_KEEPALIVE    14
#define IPC_SUBSCRIBER_KEEPALIVE_REPLY 15


#define DEBUG_LEVEL NGX_LOG_WARN
//#define DEBUG_LEVEL NGX_LOG_DEBUG
#define DBG(fmt, args...) ngx_log_error(DEBUG_LEVEL, ngx_cycle->log, 0, "IPC-HANDLERS: " fmt, ##args)
#define ERR(fmt, args...) ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, "IPC-HANDLERS: " fmt, ##args)
static ngx_int_t empty_callback() {
  return NGX_OK;
}

static ngx_str_t *str_shm_copy(ngx_str_t *str){
  return shm_copy_string(ngx_http_push_memstore_get_shm(), str);
}

static void str_shm_free(ngx_str_t *str) {
  shm_free(ngx_http_push_memstore_get_shm(), str);
}

////////// SUBSCRIBE ////////////////
typedef struct {
  ngx_str_t               *shm_chid;
  nhpm_channel_head_shm_t *shared_channel_data;
  nhpm_channel_head_t     *origin_chanhead;
  subscriber_t            *subscriber;
} subscribe_data_t;
ngx_int_t memstore_ipc_send_subscribe(ngx_int_t dst, ngx_str_t *chid, nhpm_channel_head_t *origin_chanhead) {
  DBG("send subscribe to %i, %V", dst, chid);
  subscribe_data_t        data = {str_shm_copy(chid), NULL, origin_chanhead, NULL};
  return ipc_alert(ngx_http_push_memstore_get_ipc(), dst, IPC_SUBSCRIBE, &data, sizeof(data));
}
static void receive_subscribe(ngx_int_t sender, void *data) {
  nhpm_channel_head_t     *head;
  subscribe_data_t        *d = (subscribe_data_t *)data;
  subscriber_t            *sub;
  
  DBG("received subscribe request for channel %V", d->shm_chid);
  head = ngx_http_push_memstore_get_chanhead(d->shm_chid);
  
  if(head == NULL) {
    ERR("couldn't get chanhead while receiving subscribe ipc msg");
    d->shared_channel_data = NULL;
    d->subscriber = NULL;
  }
  else {
    sub = memstore_subscriber_create(sender, &head->id, d->origin_chanhead);
    sub->enqueue(sub);
    head->spooler.add(&head->spooler, sub);
    d->subscriber = sub;
    d->shared_channel_data = head->shared;
  }

  ipc_alert(ngx_http_push_memstore_get_ipc(), sender, IPC_SUBSCRIBE_REPLY, d, sizeof(*d));
}
static void receive_subscribe_reply(ngx_int_t sender, void *data) {
  subscribe_data_t *d = (subscribe_data_t *)data;
  DBG("received subscribe reply for channel %V", d->shm_chid);
  //we have the chanhead address, but are too afraid to use it.
  nhpm_channel_head_t   *head = ngx_http_push_memstore_get_chanhead(d->shm_chid);
  if(head == NULL) {
    ERR("Error regarding an aspect of life or maybe freshly fallen cookie crumbles");
  }
  if(head->shared) {
    assert(head->shared == d->shared_channel_data);
  }
  head->shared = d->shared_channel_data;
  assert(head->shared != NULL);
  assert(head->ipc_sub == NULL);
  head->ipc_sub = d->subscriber;
  //TODO: shared counts
  head->status = READY;
  str_shm_free(d->shm_chid);
}



////////// UNSUBSCRIBED ////////////////
typedef struct {
  ngx_str_t    *shm_chid;
  void         *privdata;
} unsubscribed_data_t;
ngx_int_t memstore_ipc_send_unsubscribed(ngx_int_t dst, ngx_str_t *chid, void* privdata) {
  DBG("send unsubscribed to %i %V", dst, chid);
  unsubscribed_data_t        data = {str_shm_copy(chid), privdata};
  return ipc_alert(ngx_http_push_memstore_get_ipc(), dst, IPC_UNSUBSCRIBED, &data, sizeof(data));
}
static void receive_unsubscribed(ngx_int_t sender, void *data) {
  unsubscribed_data_t    *d = (unsubscribed_data_t *)data;
  DBG("received unsubscribed request for channel %V pridata", d->shm_chid, d->privdata);
  if(memstore_channel_owner(d->shm_chid) != memstore_slot()) {
    nhpm_channel_head_t    *head;
    //find channel
    head = ngx_http_push_memstore_find_chanhead(d->shm_chid);<commit_msg>oh, crashy<commit_after>#include <ngx_http_push_module.h>
#include <ngx_channel.h>
#include "shmem.h"
#include "ipc.h"
#include "ipc-handlers.h"
#include "store-private.h"
#include <assert.h>
#include "../../subscribers/memstore.h"

#define IPC_SUBSCRIBE               1
#define IPC_SUBSCRIBE_REPLY         2
#define IPC_UNSUBSCRIBE             3  //NOT USED
#define IPC_UNSUBSCRIBED            4
#define IPC_PUBLISH_MESSAGE         5
#define IPC_PUBLISH_MESSAGE_REPLY   6
#define IPC_PUBLISH_STATUS          7
#define IPC_GET_MESSAGE             8
#define IPC_GET_MESSAGE_REPLY       9
#define IPC_DELETE                  10
#define IPC_DELETE_REPLY            11
#define IPC_GET_CHANNEL_INFO        12
#define IPC_GET_CHANNEL_INFO_REPLY  13
#define IPC_SUBSCRIBER_KEEPALIVE    14
#define IPC_SUBSCRIBER_KEEPALIVE_REPLY 15


#define DEBUG_LEVEL NGX_LOG_WARN
//#define DEBUG_LEVEL NGX_LOG_DEBUG
#define DBG(fmt, args...) ngx_log_error(DEBUG_LEVEL, ngx_cycle->log, 0, "IPC-HANDLERS: " fmt, ##args)
#define ERR(fmt, args...) ngx_log_error(NGX_LOG_ERR, ngx_cycle->log, 0, "IPC-HANDLERS: " fmt, ##args)
static ngx_int_t empty_callback() {
  return NGX_OK;
}

static ngx_str_t *str_shm_copy(ngx_str_t *str){
  return shm_copy_string(ngx_http_push_memstore_get_shm(), str);
}

static void str_shm_free(ngx_str_t *str) {
  shm_free(ngx_http_push_memstore_get_shm(), str);
}

////////// SUBSCRIBE ////////////////
typedef struct {
  ngx_str_t               *shm_chid;
  nhpm_channel_head_shm_t *shared_channel_data;
  nhpm_channel_head_t     *origin_chanhead;
  subscriber_t            *subscriber;
} subscribe_data_t;
ngx_int_t memstore_ipc_send_subscribe(ngx_int_t dst, ngx_str_t *chid, nhpm_channel_head_t *origin_chanhead) {
  DBG("send subscribe to %i, %V", dst, chid);
  subscribe_data_t        data = {str_shm_copy(chid), NULL, origin_chanhead, NULL};
  return ipc_alert(ngx_http_push_memstore_get_ipc(), dst, IPC_SUBSCRIBE, &data, sizeof(data));
}
static void receive_subscribe(ngx_int_t sender, void *data) {
  nhpm_channel_head_t     *head;
  subscribe_data_t        *d = (subscribe_data_t *)data;
  subscriber_t            *sub;
  
  DBG("received subscribe request for channel %V", d->shm_chid);
  head = ngx_http_push_memstore_get_chanhead(d->shm_chid);
  
  if(head == NULL) {
    ERR("couldn't get chanhead while receiving subscribe ipc msg");
    d->shared_channel_data = NULL;
    d->subscriber = NULL;
  }
  else {
    sub = memstore_subscriber_create(sender, &head->id, d->origin_chanhead);
    sub->enqueue(sub);
    head->spooler.add(&head->spooler, sub);
    d->subscriber = sub;
    d->shared_channel_data = head->shared;
  }

  ipc_alert(ngx_http_push_memstore_get_ipc(), sender, IPC_SUBSCRIBE_REPLY, d, sizeof(*d));
}
static void receive_subscribe_reply(ngx_int_t sender, void *data) {
  subscribe_data_t *d = (subscribe_data_t *)data;
  DBG("received subscribe reply for channel %V", d->shm_chid);
  //we have the chanhead address, but are too afraid to use it.
  nhpm_channel_head_t   *head = ngx_http_push_memstore_get_chanhead(d->shm_chid);
  if(head == NULL) {
    ERR("Error regarding an aspect of life or maybe freshly fallen cookie crumbles");
  }
  if(head->shared) {
    assert(head->shared == d->shared_channel_data);
  }
  head->shared = d->shared_channel_data;
  assert(head->shared != NULL);
  if(head->ipc_sub) {
    assert(head->ipc_sub == d->subscriber);
  }
  else {
    head->ipc_sub = d->subscriber;
  }
  head->status = READY;
  str_shm_free(d->shm_chid);
}



////////// UNSUBSCRIBED ////////////////
typedef struct {
  ngx_str_t    *shm_chid;
  void         *privdata;
} unsubscribed_data_t;
ngx_int_t memstore_ipc_send_unsubscribed(ngx_int_t dst, ngx_str_t *chid, void* privdata) {
  DBG("send unsubscribed to %i %V", dst, chid);
  unsubscribed_data_t        data = {str_shm_copy(chid), privdata};
  return ipc_alert(ngx_http_push_memstore_get_ipc(), dst, IPC_UNSUBSCRIBED, &data, sizeof(data));
}
static void receive_unsubscribed(ngx_int_t sender, void *data) {
  unsubscribed_data_t    *d = (unsubscribed_data_t *)data;
  DBG("received unsubscribed request for channel %V pridata", d->shm_chid, d->privdata);
  if(memstore_channel_owner(d->shm_chid) != memstore_slot()) {
    nhpm_channel_head_t    *head;
    //find channel
    head = ngx_http_push_memstore_find_chanhead(d->shm_chid);
{'fraction_non_alphanumeric': 0.12162631761844087, 'fraction_numerical': 0.00532839105756979, 'mean_word_length': 4.016850668216153, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 73, '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': '2340279', 'n_tokens_mistral': 3445, 'n_tokens_neox': 3324, 'n_words': 637}