<commit_before>

#include <Python.h>
#include <numpy/arrayobject.h>
#include <stdlib.h>
#include <stdint.h>
#include <stdio.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <string.h>
#include <time.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <pthread.h>

#include "readingdb_py.h"

#include "c6/readingdb.h"
#include "c6/rpc.h"
#include "c6/commands.h"

static char host[512];
static short port;
static int workers;
static int substream;
static int setup = 0;

void db_setup(char *a_host, 
              short a_port,
              int a_workers,
              int a_substream) {
  strncpy(host, a_host, sizeof(host));
  port = a_port;
  workers = a_workers;
  substream = a_substream;
  setup = 1;
  import_array();
}

struct np_point {
  double ts;
  double val;
};

struct request {
  pthread_mutex_t lock;
  struct sock_request *conn;

  // server conf
  const char *host;
  const short port;

  // work queue
  int next_streamid;
  int n_streams;

  // request params
  const struct request_desc *r;

  struct np_point **return_data;
  int *return_data_len;
  int errors;
};

int setup_request(struct sock_request *conn,
                  const struct request_desc *req,
                  unsigned long long streamid, 
                  unsigned long long starttime) {
  switch (req->type) {
  case REQ_QUERY:
    return db_query_all(conn, streamid, starttime, req->endtime, QUERY_DATA);
  case REQ_ITER:
    return db_iter(conn, streamid, starttime, req->direction, req->limit);
  default:
    return -1;
  }
}

int read_numpy_resultset(struct sock_request *ipp, 
                         struct np_point **buf, int *off) {
  Response *r;
  struct pbuf_header h;
  void *reply;
  int len, i;

  /* read the reply */
  if (fread(&h, sizeof(h), 1, ipp->sock_fp) <= 0) {
    // PyErr_Format(PyExc_IOError, "read_resultset: error reading from socket: %s", strerror(errno));
    return -errno;
  }
  len = ntohl(h.body_length);
  reply = malloc(len);
  if (!reply) return -1;
  if (fread(reply, len, 1, ipp->sock_fp) <= 0) {
    free(reply);
    // PyErr_Format(PyExc_IOError, "read_resultset: error reading from socket: %s", strerror(errno));
    return -errno;
  }
  r = response__unpack(NULL, len, reply);
  free(reply);
  if (!r) {
    // PyErr_Format(PyExc_IOError, "read_resultset: error unpacking");
    return -errno;
  }
/*   printf("Received reply code: %i results: %li len: %i\n", */
/*          r->error, r->data->n_data, len); */
  if (r->error != RESPONSE__ERROR_CODE__OK) {
    // PyErr_Format(PyExc_Exception, "read_resultset: received error from server: %i", r->error);
    return -errno;
  }

  /* build the python data structure  */
  if (*buf == NULL || *off == 0) {
    len = sizeof(struct np_point) * r->data->n_data;
    *buf = malloc(len);
  } else {
    len = sizeof(struct np_point) * (r->data->n_data + *off);
    *buf = realloc(*buf, len);
    // printf("reallocing\n");
  }
  if (!*buf) {
    fprintf(stderr, "Alloc/realloc failed: request: %i\n", len);
    response__free_unpacked(r, NULL);
    return -1;
  }

  for (i = *off; i < *off + r->data->n_data; i++) {
    ((struct np_point *)(*buf))[i].ts = r->data->data[i - *off]->timestamp;
    ((struct np_point *)(*buf))[i].val = r->data->data[i - *off]->value;
  }
  *off += r->data->n_data;
  response__free_unpacked(r, NULL);
  
  return r->data->n_data;
<commit_msg> * fix stupid use-after-free bug<commit_after>

#include <Python.h>
#include <numpy/arrayobject.h>
#include <stdlib.h>
#include <stdint.h>
#include <stdio.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <errno.h>
#include <string.h>
#include <time.h>
#include <signal.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <pthread.h>

#include "readingdb_py.h"

#include "c6/readingdb.h"
#include "c6/rpc.h"
#include "c6/commands.h"

static char host[512];
static short port;
static int workers;
static int substream;
static int setup = 0;

void db_setup(char *a_host, 
              short a_port,
              int a_workers,
              int a_substream) {
  strncpy(host, a_host, sizeof(host));
  port = a_port;
  workers = a_workers;
  substream = a_substream;
  setup = 1;
  import_array();
}

struct np_point {
  double ts;
  double val;
};

struct request {
  pthread_mutex_t lock;
  struct sock_request *conn;

  // server conf
  const char *host;
  const short port;

  // work queue
  int next_streamid;
  int n_streams;

  // request params
  const struct request_desc *r;

  struct np_point **return_data;
  int *return_data_len;
  int errors;
};

int setup_request(struct sock_request *conn,
                  const struct request_desc *req,
                  unsigned long long streamid, 
                  unsigned long long starttime) {
  switch (req->type) {
  case REQ_QUERY:
    return db_query_all(conn, streamid, starttime, req->endtime, QUERY_DATA);
  case REQ_ITER:
    return db_iter(conn, streamid, starttime, req->direction, req->limit);
  default:
    return -1;
  }
}

int read_numpy_resultset(struct sock_request *ipp, 
                         struct np_point **buf, int *off) {
  Response *r;
  struct pbuf_header h;
  void *reply;
  int len, i, rv;

  /* read the reply */
  if (fread(&h, sizeof(h), 1, ipp->sock_fp) <= 0) {
    // PyErr_Format(PyExc_IOError, "read_resultset: error reading from socket: %s", strerror(errno));
    return -errno;
  }
  len = ntohl(h.body_length);
  reply = malloc(len);
  if (!reply) return -1;
  if (fread(reply, len, 1, ipp->sock_fp) <= 0) {
    free(reply);
    // PyErr_Format(PyExc_IOError, "read_resultset: error reading from socket: %s", strerror(errno));
    return -errno;
  }
  r = response__unpack(NULL, len, reply);
  free(reply);
  if (!r) {
    // PyErr_Format(PyExc_IOError, "read_resultset: error unpacking");
    return -errno;
  }
/*   printf("Received reply code: %i results: %li len: %i\n", */
/*          r->error, r->data->n_data, len); */
  if (r->error != RESPONSE__ERROR_CODE__OK) {
    // PyErr_Format(PyExc_Exception, "read_resultset: received error from server: %i", r->error);
    return -errno;
  }

  /* build the python data structure  */
  if (*buf == NULL || *off == 0) {
    len = sizeof(struct np_point) * r->data->n_data;
    *buf = malloc(len);
  } else {
    len = sizeof(struct np_point) * (r->data->n_data + *off);
    *buf = realloc(*buf, len);
    // printf("reallocing\n");
  }
  if (!*buf) {
    fprintf(stderr, "Alloc/realloc failed: request: %i\n", len);
    response__free_unpacked(r, NULL);
    return -1;
  }

  for (i = *off; i < *off + r->data->n_data; i++) {
    ((struct np_point *)(*buf))[i].ts = r->data->data[i - *off]->timestamp;
    ((struct np_point *)(*buf))[i].val = r->data->data[i - *off]->value;
  }
  *off += r->data->n_data;
  rv = r->data->n_data;
  response__free_unpacked(r, NULL);
  
  return rv;
{'fraction_non_alphanumeric': 0.15433989788475566, 'fraction_numerical': 0.0046681254558716264, 'mean_word_length': 3.1029323758228604, 'pattern_counts': {'":': 0, '<': 37, '<?xml version=': 0, '>': 85, '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': '3552753', 'n_tokens_mistral': 2590, 'n_tokens_neox': 2527, 'n_words': 631}