<commit_before>#define _XOPEN_SOURCE 500

#define _POSIX_C_SOURCE 200809L
#define _DEFAULT_SOURCE
#define _GNU_SOURCE

/**
 * blockVerify2.c
 *
 *
 */
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <math.h>
#include <sys/uio.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <assert.h>

#include "utils.h"
#include "positions.h"
#include "blockVerify.h"


typedef struct {
  size_t seed;
  size_t merge; 
  size_t randomize;
  size_t o_direct;
  size_t threads;
  size_t isSetup;
  size_t inQueue;

  size_t numPositions;
  size_t maxAllocPositions;
  size_t nextpos;
  
  size_t inputComplete;
  size_t minbs, maxbs;
  
  positionType *p;
  jobType *j;

  double start, finish;
} blockVerifyType;


typedef struct {
  size_t id;
  size_t pleaseStop;
  size_t processed;
  size_t correct;
  size_t wrong;

  size_t bytesProcessed;

  blockVerifyType *b;
} threadInfoType;

pthread_t *pt;
threadInfoType *threadContext;
pthread_mutex_t mutex;

int verbose;
int keepRunning;

void addPosition(blockVerifyType *b, positionType *p) {
  assert(b);
  pthread_mutex_lock(&mutex);
  b->numPositions = b->numPositions+1;

  if (b->numPositions >= b->maxAllocPositions) {
    b->maxAllocPositions += 5000;
    b->p = realloc(b->p, (b->maxAllocPositions) * sizeof(positionType));
    //    fprintf(stderr,"*realloc to %zd\n", b->maxAllocPositions);
  }

  memcpy(&b->p[b->numPositions-1], p, sizeof(positionType));
  if (p->len > b->maxbs) b->maxbs = p->len;
  if (p->len < b->minbs) b->minbs = p->len;
  //  fprintf(stdout,"*added* %zd\n", b->p[b->numPositions-1].pos);
  
  //  b->p[b->numPositions-1].pos = pos;
  assert(b->p);
  pthread_mutex_unlock(&mutex);
}
  
  


positionType * isNextPosition(blockVerifyType *b) {
  positionType *ret = NULL;
  pthread_mutex_lock(&mutex);
  if (b && (b->nextpos < b->numPositions)) {
    ret = &b->p[b->nextpos];
    //    fprintf(stderr, "---%zd\n", nextpos);
    b->nextpos++;
  }
  pthread_mutex_unlock(&mutex);
  return ret;
}
  
static void *runInputThread(void *arg) {
  fprintf(stderr,"running input thread\n");
  threadInfoType *threadContext = (threadInfoType*)arg; // grab the thread threadContext args

  char *line = NULL;
  //  size_t len = 0;
  //  ssize_t nread;
  
  positionContainer pc;
  while (1) {
    positionContainerLoadLines(&pc, stdin, 1);
    if (pc.sz == 0) break;
    //    fprintf(stderr,"*data* %zd\n", pc.positions[0].pos);
    addPosition(threadContext->b, &pc.positions[0]);
  }
  
  //  while ((nread = getline(&line, &len, stream)) != -1) {
  //  }
  
  fprintf(stderr,"*info* input complete\n");
  fprintf(stderr,"should i randomize %zd\n", threadContext->b->randomize);
  if (threadContext->b->randomize) {
    positionsRandomize(threadContext->b->p, threadContext->b->numPositions, threadContext->b->seed);
  } else {
    positionsSortPositions(threadContext->b->p, threadContext->b->numPositions);
  }
  threadContext->b->inputComplete = 1;
  
  free(line);

  return NULL;
}
   
 

static void *runVerifyThread(void *arg)
{
  threadInfoType *threadContext = (threadInfoType*)arg; // grab the thread threadContext args
  //    fprintf(stderr,"running thread %zd\n", threadContext->id);

  char *buf = NULL, *randombuf = NULL;
  //  fprintf(stderr,"max len = %zd\n", threadContext->b->maxbs);
  assert(threadContext->b->maxbs < 100*1024*1024);
  CALLOC(randombuf, 100*1024*1024+1, 1);
  CALLOC(buf, 100*1024*1024+1, 1);

  int lastdevid = -1, lastseed = -1, fd = -1;
  while (threadContext->pleaseStop == 0) {
    // process if:
    //   if merge and input complete
    //   or
    //   if !merge
    size_t go = 0;
    if ((threadContext->b->merge == 1) && threadContext->b->inputComplete == 1) {
      go = 1;
    }
    if (threadContext->b->merge == 0) {
      go = 1;
    }

    positionType *pp = NULL;
    if (go && ((pp = isNextPosition(threadContext->b)) != NULL)) {

      if (pp->deviceid != lastdevid) {
	// check it's open
	if (fd > 0) {
	  close(fd);
	}
	
	char *fn = threadContext->b->j->devices[pp->deviceid];
	//	fprintf(stderr,"*info* %d / id %zd -> opening %s\n", pp->deviceid, threadContext->id, fn);
	fd = open(fn, O_RDONLY | threadContext->b->o_direct);
	if (fd < 0) {
	  fprintf(stderr,"*info* if the device doesn't support O_DIRECT you may need the '-o' option to turn it off\n");
	  perror(fn);
	}
	lastdevid = pp->deviceid;
      }

      if (pp->action == 'W' && pp->finishTime > 0) {
	// make sure correct comparison
	if (pp->seed != lastseed) {
	  generateRandomBuffer(randombuf, threadContext->b->maxbs, pp->seed);
	  lastseed = pp->seed;
	}
	
	size_t diff = 0;
	memcpy(randombuf, pp, sizeof(size_t));
	int ret = verifyPosition(fd, pp, randombuf, buf, &diff, lastseed, 0, 0);
	if (ret == 0) {
	  //	  	  fprintf(stdout,"verify %lf [%zd] %d %zd, num %zd\n", timedouble(), threadContext->id, pp->deviceid, pp->pos, threadContext->b->numPositions);
	  threadContext->correct++;
	} else {
	  fprintf(stdout,"error %lf [%zd] %d %zd, num %zd\n", timedouble(), threadContext->id, pp->deviceid, pp->pos, threadContext->b->numPositions);
	  fprintf(stderr,"*error* incorrect block at position %zd\n", pp->pos);
	  threadContext->wrong++;
	  if (threadContext->wrong > 2) {
	    exit(1);
	  }
	}
	//	fprintf(stderr,"ret = %d\n", ret);
	threadContext->processed++;
	threadContext->bytesProcessed += pp->len;
      }
      
      //usleep(1);
      pthread_yield();
    } else {
      usleep(100);
    }
  }
  if (fd > 0) close(fd);
  //  fprintf(stderr,"*%zd* is finished %zd\n", threadContext->id, threadContext->pleaseStop);
  return NULL;
}

void setupVerificationThreads(blockVerifyType *b) {
  // add one for the input thread
  CALLOC(pt, b->threads + 1, sizeof(pthread_t)); <commit_msg>Added -V<commit_after>#define _XOPEN_SOURCE 500

#define _POSIX_C_SOURCE 200809L
#define _DEFAULT_SOURCE
#define _GNU_SOURCE

/**
 * blockVerify2.c
 *
 *
 */
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>
#include <math.h>
#include <sys/uio.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <assert.h>

#include "utils.h"
#include "positions.h"
#include "blockVerify.h"


typedef struct {
  size_t seed;
  size_t merge; 
  size_t randomize;
  size_t o_direct;
  size_t threads;
  size_t isSetup;
  size_t inQueue;

  size_t numPositions;
  size_t maxAllocPositions;
  size_t nextpos;
  
  size_t inputComplete;
  size_t minbs, maxbs;
  
  positionType *p;
  jobType *j;

  double start, finish;
} blockVerifyType;


typedef struct {
  size_t id;
  size_t pleaseStop;
  size_t processed;
  size_t correct;
  size_t wrong;

  size_t bytesProcessed;

  blockVerifyType *b;
} threadInfoType;

pthread_t *pt;
threadInfoType *threadContext;
pthread_mutex_t mutex;

int verbose;
int keepRunning;

void addPosition(blockVerifyType *b, positionType *p) {
  assert(b);
  pthread_mutex_lock(&mutex);
  b->numPositions = b->numPositions+1;

  if (b->numPositions >= b->maxAllocPositions) {
    b->maxAllocPositions += 5000;
    b->p = realloc(b->p, (b->maxAllocPositions) * sizeof(positionType));
    //    fprintf(stderr,"*realloc to %zd\n", b->maxAllocPositions);
  }

  memcpy(&b->p[b->numPositions-1], p, sizeof(positionType));
  if (p->len > b->maxbs) b->maxbs = p->len;
  if (p->len < b->minbs) b->minbs = p->len;
  //  fprintf(stdout,"*added* %zd\n", b->p[b->numPositions-1].pos);
  
  //  b->p[b->numPositions-1].pos = pos;
  assert(b->p);
  pthread_mutex_unlock(&mutex);
}
  
  


positionType * isNextPosition(blockVerifyType *b) {
  positionType *ret = NULL;
  pthread_mutex_lock(&mutex);
  if (b && (b->nextpos < b->numPositions)) {
    ret = &b->p[b->nextpos];
    //    fprintf(stderr, "---%zd\n", nextpos);
    b->nextpos++;
  }
  pthread_mutex_unlock(&mutex);
  return ret;
}
  
static void *runInputThread(void *arg) {
  fprintf(stderr,"running input thread\n");
  threadInfoType *threadContext = (threadInfoType*)arg; // grab the thread threadContext args

  char *line = NULL;
  //  size_t len = 0;
  //  ssize_t nread;
  
  positionContainer pc;
  while (1) {
    positionContainerLoadLines(&pc, stdin, 1);
    if (pc.sz == 0) break;
    //    fprintf(stderr,"*data* %zd\n", pc.positions[0].pos);
    addPosition(threadContext->b, &pc.positions[0]);
  }
  
  //  while ((nread = getline(&line, &len, stream)) != -1) {
  //  }
  
  fprintf(stderr,"*info* input complete\n");
  fprintf(stderr,"should i randomize %zd\n", threadContext->b->randomize);
  if (threadContext->b->randomize) {
    positionsRandomize(threadContext->b->p, threadContext->b->numPositions, threadContext->b->seed);
  } else {
    positionsSortPositions(threadContext->b->p, threadContext->b->numPositions);
  }
  threadContext->b->inputComplete = 1;
  
  free(line);

  return NULL;
}
   
 

static void *runVerifyThread(void *arg)
{
  threadInfoType *threadContext = (threadInfoType*)arg; // grab the thread threadContext args
  //    fprintf(stderr,"running thread %zd\n", threadContext->id);

  char *buf = NULL, *randombuf = NULL;
  //  fprintf(stderr,"max len = %zd\n", threadContext->b->maxbs);
  assert(threadContext->b->maxbs < 100*1024*1024);
  CALLOC(randombuf, 100*1024*1024+1, 1);
  CALLOC(buf, 100*1024*1024+1, 1);

  int lastdevid = -1, lastseed = -1, fd = -1;
  while (threadContext->pleaseStop == 0) {
    // process if:
    //   if merge and input complete
    //   or
    //   if !merge
    size_t go = 0;
    if ((threadContext->b->merge == 1) && threadContext->b->inputComplete == 1) {
      go = 1;
    }
    if (threadContext->b->merge == 0) {
      go = 1;
    }

    positionType *pp = NULL;
    if (go && ((pp = isNextPosition(threadContext->b)) != NULL)) {

      if (pp->deviceid != lastdevid) {
	// check it's open
	if (fd > 0) {
	  close(fd);
	}
	
	char *fn = threadContext->b->j->devices[pp->deviceid];
	//	fprintf(stderr,"*info* %d / id %zd -> opening %s\n", pp->deviceid, threadContext->id, fn);
	fd = open(fn, O_RDONLY | threadContext->b->o_direct);
	if (fd < 0) {
	  fprintf(stderr,"*info* if the device doesn't support O_DIRECT you may need the '-o' option to turn it off\n");
	  perror(fn);
	}
	lastdevid = pp->deviceid;
      }

      if (pp->action == 'W' && pp->finishTime > 0) {
	// make sure correct comparison
	if (pp->seed != lastseed) {
	  generateRandomBuffer(randombuf, threadContext->b->maxbs, pp->seed);
	  lastseed = pp->seed;
	}
	
	size_t diff = 0;
	memcpy(randombuf, pp, sizeof(size_t));
	int ret = verifyPosition(fd, pp, randombuf, buf, &diff, lastseed, 0, 0);
	if (ret == 0) {
	  if (verbose) fprintf(stdout,"verify %lf [%zd] %d %zd, num %zd\n", timedouble(), threadContext->id, pp->deviceid, pp->pos, threadContext->b->numPositions);
	  threadContext->correct++;
	} else {
	  fprintf(stdout,"error %lf [%zd] %d %zd, num %zd\n", timedouble(), threadContext->id, pp->deviceid, pp->pos, threadContext->b->numPositions);
	  fprintf(stderr,"*error* incorrect block at position %zd\n", pp->pos);
	  threadContext->wrong++;
	  if (threadContext->wrong > 2) {
	    exit(1);
	  }
	}
	//	fprintf(stderr,"ret = %d\n", ret);
	threadContext->processed++;
	threadContext->bytesProcessed += pp->len;
      }
      
      //usleep(1);
      pthread_yield();
    } else {
      usleep(100);
    }
  }
  if (fd > 0) close(fd);
  //  fprintf(stderr,"*%zd* is finished %zd\n", threadContext->id, threadContext->pleaseStop);
  return NULL;
}

void setupVerificationThreads(blockVerifyType *b) {
  // add one for the input thread
  CALLOC(pt, b->threads + 1, sizeof(pthread_t)); 
{'fraction_non_alphanumeric': 0.1708816906287892, 'fraction_numerical': 0.01524337432877187, 'mean_word_length': 3.512309495896835, 'pattern_counts': {'":': 0, '<': 33, '<?xml version=': 0, '>': 231, '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': '7164072', 'n_tokens_mistral': 4481, 'n_tokens_neox': 4281, 'n_words': 929}