<commit_before>//worker processes of the world, unite.
#include <nchan_module.h>
#include <ngx_channel.h>
#include <assert.h>
#include "ipc.h"

#if FAKESHARD
#include "shmem.h"
#include "store-private.h"
#endif

#define DEBUG_LEVEL NGX_LOG_DEBUG
//#define DEBUG_LEVEL NGX_LOG_WARN

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

//#define DEBUG_DELAY_IPC_RECEIVE_ALERT_MSEC 100

static void ipc_read_handler(ngx_event_t *ev);

ngx_int_t ipc_init(ipc_t *ipc) {
  int                             i = 0;
  ipc_process_t                  *proc;
  
  for(i=0; i< NGX_MAX_PROCESSES; i++) {
    proc = &ipc->process[i];
    proc->ipc = ipc;
    proc->pipe[0]=NGX_INVALID_FILE;
    proc->pipe[1]=NGX_INVALID_FILE;
    proc->c=NULL;
    proc->active = 0;
    ngx_memzero(proc->wbuf.alerts, sizeof(proc->wbuf.alerts));
    proc->wbuf.first = 0;
    proc->wbuf.n = 0;
    proc->wbuf.overflow_first = NULL;
    proc->wbuf.overflow_last = NULL;
    proc->wbuf.overflow_n = 0;
  }
  return NGX_OK;
}

ngx_int_t ipc_set_handler(ipc_t *ipc, void (*alert_handler)(ngx_int_t, ngx_uint_t, void *)) {
  ipc->handler=alert_handler;
  return NGX_OK;
}

static void ipc_try_close_fd(ngx_socket_t *fd) {
  if(*fd != NGX_INVALID_FILE) {
    ngx_close_socket(*fd);
    *fd=NGX_INVALID_FILE;
  }
}

ngx_int_t ipc_open(ipc_t *ipc, ngx_cycle_t *cycle, ngx_int_t workers, void (*slot_callback)(int slot, int worker)) {
//initialize pipes for workers in advance.
  int                             i, j, s = 0;
  ngx_int_t                       last_expected_process = ngx_last_process;
  ipc_process_t                  *proc;
  ngx_socket_t                   *socks;
  
  /* here's the deal: we have no control over fork()ing, nginx's internal 
    * socketpairs are unusable for our purposes (as of nginx 0.8 -- check the 
    * code to see why), and the module initialization callbacks occur before
    * any workers are spawned. Rather than futzing around with existing 
    * socketpairs, we make our own pipes array. 
    * Trouble is, ngx_spawn_process() creates them one-by-one, and we need to 
    * do it all at once. So we must guess all the workers' ngx_process_slots in 
    * advance. Meaning the spawning logic must be copied to the T.
    * ... with some allowances for already-opened sockets...
    */
  for(i=0; i < workers; i++) {
    //copypasta from os/unix/ngx_process.c (ngx_spawn_process)
    while (s < last_expected_process && ngx_processes[s].pid != -1) {
      //find empty existing slot
      s++;
    }
    
    if(slot_callback) {
      slot_callback(s, i);
    }
    
    proc = &ipc->process[s];
    
    if(!proc->active) {
      socks = proc->pipe;
      
      assert(socks[0] == NGX_INVALID_FILE && socks[1] == NGX_INVALID_FILE);
      
      //make-a-pipe
      if (pipe(socks) == -1) {
        ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "pipe() failed while initializing nchan IPC");
        return NGX_ERROR;
      }
      //make both ends nonblocking
      for(j=0; j <= 1; j++) {
        if (ngx_nonblocking(socks[j]) == -1) {
          ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, ngx_nonblocking_n " failed on pipe socket %i while initializing nchan", j);
          ipc_try_close_fd(&socks[0]);
          ipc_try_close_fd(&socks[1]);
          return NGX_ERROR;
        }
      }
      //It's ALIIIIIVE! ... erm.. active...
      proc->active = 1;
    }
    s++; //NEXT!!
  }
  
  //ERR("ipc_alert_t size %i bytes", sizeof(ipc_alert_t));
  
  return NGX_OK;
}



ngx_int_t ipc_close(ipc_t *ipc, ngx_cycle_t *cycle) {
  int i;
  ipc_process_t            *proc;
  ipc_writebuf_overflow_t  *of, *of_next;
  
  for (i=0; i<NGX_MAX_PROCESSES; i++) {
    proc = &ipc->process[i];<commit_msg>fix receiving stray IPC messages after a couple of reloads<commit_after>//worker processes of the world, unite.
#include <nchan_module.h>
#include <ngx_channel.h>
#include <assert.h>
#include "ipc.h"

#if FAKESHARD
#include "shmem.h"
#include "store-private.h"
#endif

#define DEBUG_LEVEL NGX_LOG_DEBUG
//#define DEBUG_LEVEL NGX_LOG_WARN

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

//#define DEBUG_DELAY_IPC_RECEIVE_ALERT_MSEC 100

static void ipc_read_handler(ngx_event_t *ev);

ngx_int_t ipc_init(ipc_t *ipc) {
  int                             i = 0;
  ipc_process_t                  *proc;
  
  for(i=0; i< NGX_MAX_PROCESSES; i++) {
    proc = &ipc->process[i];
    proc->ipc = ipc;
    proc->pipe[0]=NGX_INVALID_FILE;
    proc->pipe[1]=NGX_INVALID_FILE;
    proc->c=NULL;
    proc->active = 0;
    ngx_memzero(proc->wbuf.alerts, sizeof(proc->wbuf.alerts));
    proc->wbuf.first = 0;
    proc->wbuf.n = 0;
    proc->wbuf.overflow_first = NULL;
    proc->wbuf.overflow_last = NULL;
    proc->wbuf.overflow_n = 0;
  }
  return NGX_OK;
}

ngx_int_t ipc_set_handler(ipc_t *ipc, void (*alert_handler)(ngx_int_t, ngx_uint_t, void *)) {
  ipc->handler=alert_handler;
  return NGX_OK;
}

static void ipc_try_close_fd(ngx_socket_t *fd) {
  if(*fd != NGX_INVALID_FILE) {
    ngx_close_socket(*fd);
    *fd=NGX_INVALID_FILE;
  }
}

ngx_int_t ipc_open(ipc_t *ipc, ngx_cycle_t *cycle, ngx_int_t workers, void (*slot_callback)(int slot, int worker)) {
//initialize pipes for workers in advance.
  int                             i, j, s = 0;
  ngx_int_t                       last_expected_process = ngx_last_process;
  ipc_process_t                  *proc;
  ngx_socket_t                   *socks;
  
  /* here's the deal: we have no control over fork()ing, nginx's internal 
    * socketpairs are unusable for our purposes (as of nginx 0.8 -- check the 
    * code to see why), and the module initialization callbacks occur before
    * any workers are spawned. Rather than futzing around with existing 
    * socketpairs, we make our own pipes array. 
    * Trouble is, ngx_spawn_process() creates them one-by-one, and we need to 
    * do it all at once. So we must guess all the workers' ngx_process_slots in 
    * advance. Meaning the spawning logic must be copied to the T.
    * ... with some allowances for already-opened sockets...
    */
  for(i=0; i < workers; i++) {
    //copypasta from os/unix/ngx_process.c (ngx_spawn_process)
    while (s < last_expected_process && ngx_processes[s].pid != -1) {
      //find empty existing slot
      s++;
    }
    
    if(slot_callback) {
      slot_callback(s, i);
    }
    
    proc = &ipc->process[s];

    socks = proc->pipe;
    
    if(proc->active) {
      // reinitialize already active pipes. This is done to prevent IPC alerts
      // from a previous restart that were never read from being received by
      // a newly restarted worker
      ipc_try_close_fd(&socks[0]);
      ipc_try_close_fd(&socks[1]);
      proc->active = 0;
    }
    
    assert(socks[0] == NGX_INVALID_FILE && socks[1] == NGX_INVALID_FILE);
    
    //make-a-pipe
    if (pipe(socks) == -1) {
      ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, "pipe() failed while initializing nchan IPC");
      return NGX_ERROR;
    }
    //make both ends nonblocking
    for(j=0; j <= 1; j++) {
      if (ngx_nonblocking(socks[j]) == -1) {
	ngx_log_error(NGX_LOG_ALERT, cycle->log, ngx_errno, ngx_nonblocking_n " failed on pipe socket %i while initializing nchan", j);
	ipc_try_close_fd(&socks[0]);
	ipc_try_close_fd(&socks[1]);
	return NGX_ERROR;
      }
    }
    //It's ALIIIIIVE! ... erm.. active...
    proc->active = 1;
    
    s++; //NEXT!!
  }
  
  //ERR("ipc_alert_t size %i bytes", sizeof(ipc_alert_t));
  
  return NGX_OK;
}



ngx_int_t ipc_close(ipc_t *ipc, ngx_cycle_t *cycle) {
  int i;
  ipc_process_t            *proc;
  ipc_writebuf_overflow_t  *of, *of_next;
  
  for (i=0; i<NGX_MAX_PROCESSES; i++) {
    proc = &ipc->process[i];
{'fraction_non_alphanumeric': 0.1272590361445783, 'fraction_numerical': 0.0074046184738955825, 'mean_word_length': 2.80381861575179, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 56, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4974545', 'n_tokens_mistral': 3133, 'n_tokens_neox': 3019, 'n_words': 705}