c_cpp/salusFstReCoord/src3/input.c

   1 #include <stdio.h>  // fprintf
   2
   3 #ifdef __linux__
   4 #include <sys/mman.h>  // madvise
   5 #endif
   6
   7 #include "kseq.h"
   8 #ifdef USE_ZLIBNG
   9 #include <zlib-ng.h>
  10 KSEQ_INIT(gzFile, zng_gzread)
  11 #define _GZ_OPEN_ zng_gzopen
  12 #define _GZ_BUFFER_ zng_gzbuffer
  13 #define _GZ_CLOSE_ zng_gzclose
  14 #else
  15 #ifdef USE_ZLIB
  16 #include <zlib.h>
  17 #define _GZ_BUFFER_ gzbuffer
  18 #else /* USE_LIBISAL */
  19 #include "izlib.h"
  20 #endif
  21 #define _GZ_OPEN_ gzopen
  22 #define _GZ_CLOSE_ gzclose
  23 KSEQ_INIT(gzFile, gzread)
  24 #endif
  25
  26 #include "common.h"
  27
  28 void fqReader_init(void) {
  29         int rc;
  30         const char *pattern = "R[0-9]{3}(:)?C[0-9]{3}";
  31         if ((rc = regcomp(&Parameters.regex, pattern, REG_EXTENDED)) != 0) {
  32                 regerror(rc, &Parameters.regex, Parameters.buffer, PARAMETERS_BUFFER_SIZE);
  33                 fprintf(stderr, "[x]=%d=regcomp(\"%s\") failed with '%s'.\n", rc, pattern, Parameters.buffer);
  34                 exit(EXIT_FAILURE);
  35         }
  36         Parameters.ksfp = _GZ_OPEN_(Parameters.inFastqFilename, "r");
  37         if (unlikely(Parameters.ksfp == NULL)) {
  38                 fprintf(stderr, "[x]gzopen error on opening [%s]: %s.\n", Parameters.inFastqFilename, strerror(errno));
  39                 exit(1);
  40         }
  41 #if defined(_GZ_BUFFER_)
  42         rc = _GZ_BUFFER_(Parameters.ksfp, GZBUFSIZE);
  43         if (unlikely(rc != 0)) {
  44                 fprintf(stderr, "[x]gzbuffer error: %s.\n", strerror(errno));
  45                 exit(1);
  46         }
  47 #endif
  48         Parameters.kseq = kseq_init(Parameters.ksfp);
  49         Parameters.ksflag = 1;
  50         // Parameters.worksQuene = (workerArray_t *)calloc(JOBQUEUESIZE, sizeof(workerArray_t));
  51         size_t HPAGE_shift = 21;  // 1 << 21 = 2M
  52         size_t HPAGE_alignment = 1 << HPAGE_shift;
  53         size_t worksQuene_size = JOBQUEUESIZE * sizeof(workerArray_t);
  54         size_t worksQuene_aligned_cnt = worksQuene_size >> HPAGE_shift;
  55         if (worksQuene_size % HPAGE_alignment != 0) {
  56                 worksQuene_aligned_cnt++;
  57         }
  58         size_t worksQuene_aligned_size = HPAGE_alignment * worksQuene_aligned_cnt;
  59         fprintf(stderr, "[!]aligned_alloc(%zu, %zu) for [%zu].\n", HPAGE_alignment, worksQuene_aligned_size, worksQuene_size);
  60         Parameters.worksQuene = (workerArray_t *)aligned_alloc(HPAGE_alignment, worksQuene_aligned_size);
  61         if (Parameters.worksQuene != NULL) {
  62 #ifdef __linux__
  63                 madvise(Parameters.worksQuene, worksQuene_aligned_size, MADV_HUGEPAGE);
  64 #endif
  65                 memset(Parameters.worksQuene, 0, worksQuene_aligned_size);
  66         } else {
  67                 perror("[x]aligned_alloc failed.");
  68                 exit(EXIT_FAILURE);
  69         }
  70 }
  71
  72 void fqReader_destroy(void) {
  73         regfree(&Parameters.regex);
  74         kseq_destroy(Parameters.kseq);
  75         _GZ_CLOSE_(Parameters.ksfp);
  76         free(Parameters.worksQuene);
  77 }
  78
  79 void fill_worker(int_least16_t worker_id) {
  80         kseq_t *seq = Parameters.kseq;
  81         workerArray_t *worker = &Parameters.worksQuene[worker_id];
  82         // if (atomic_load(&worker->flag) == 1) return;
  83         uint64_t index = 0;
  84         int kseq_ret = 0;
  85         for (index = 0; index < JOBITEMSIZE; index++) {
  86                 fstBCdata_t *fstBCdata_p = &worker->jobDatArray[index];
  87                 if ((kseq_ret = kseq_read(seq)) >= MINBARCODELEN) {
  88                         // strncpy((char *)fstBCdata_p->name, seq->name.s, sizeof(fstBCdata_p->name));
  89                         STRcpyARRAY(fstBCdata_p->name, seq->name.s);
  90                         /* seq->comment.s is discarded */
  91                         STRcpyARRAY(fstBCdata_p->seq, seq->seq.s);
  92                         STRcpyARRAY(fstBCdata_p->qual, seq->qual.s);
  93 #ifdef DEBUG
  94                         fprintf(stderr, "- %llu -\n", index);
  95                         ARRAYcpySTR(Parameters.buffer, fstBCdata_p->name);
  96                         // snprintf(Parameters.buffer, 1 + sizeof(fstBCdata_p->name), "%s", fstBCdata_p->name);
  97                         fprintf(stderr, "->Name:[%s]\n", Parameters.buffer);
  98                         ARRAYcpySTR(Parameters.buffer, fstBCdata_p->seq);
  99                         fprintf(stderr, "->Sequ:[%s]\n", Parameters.buffer);
 100                         ARRAYcpySTR(Parameters.buffer, fstBCdata_p->qual);
 101                         fprintf(stderr, "->Qual:[%s]\n", Parameters.buffer);
 102 #endif
 103                 } else {
 104                         fstBCdata_p->name[0] = '\0';
 105                         fstBCdata_p->seq[0] = '\0';
 106                         fstBCdata_p->qual[0] = '\0';
 107                         if (kseq_ret < 0) {  // -1 for FEOF
 108                                 Parameters.ksflag = kseq_ret;
 109                         }
 110                 }
 111                 // continue;
 112         }
 113         // Update the workerArray's flag to indicate it is filled, thus ready for worker.
 114         atomic_store(&worker->flag, 1);
 115 }