From 8f664f80ceabb459ada38faca1365e4a58398005 Mon Sep 17 00:00:00 2001 From: Heng Li Date: Thu, 26 Mar 2020 16:26:54 -0400 Subject: [PATCH] removed the hifiasm FASTA/Q reader --- Process_Read.cpp | 370 ----------------------------------------------- Process_Read.h | 35 ----- 2 files changed, 405 deletions(-) diff --git a/Process_Read.cpp b/Process_Read.cpp index 757dcbe..5851933 100644 --- a/Process_Read.cpp +++ b/Process_Read.cpp @@ -5,22 +5,6 @@ #include #include - -gz_files fps; - -R_buffer RDB; -static uint64_t total_reads; - -pthread_mutex_t i_readinputMutex; -pthread_mutex_t i_queueMutex; -pthread_mutex_t i_terminateMutex; -pthread_cond_t i_flushCond; -pthread_cond_t i_readinputflushCond; -pthread_cond_t i_stallCond; -pthread_cond_t i_readinputstallCond; -pthread_mutex_t i_doneMutex; - - uint8_t seq_nt6_table[256] = { 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, @@ -244,11 +228,6 @@ void ha_insert_read_len(All_reads *r, int read_len, int name_len) r->name_index[r->total_reads] = r->name_index[r->total_reads - 1] + name_len; } -static inline void insert_read(All_reads* r, kstring_t* read, kstring_t* name) -{ - ha_insert_read_len(r, read->l, name->l); -} - void malloc_All_reads(All_reads* r) { r->read_size = (uint64_t*)malloc(sizeof(uint64_t)*r->total_reads); @@ -649,299 +628,6 @@ void ha_compress_base(uint8_t* dest, char* src, uint64_t src_l, uint64_t** N_sit } } -void open_file(gz_files* nfps, char* name) -{ - nfps->fp = gzopen(name, "r"); - if(nfps->fp == 0) - { - fprintf(stderr, "[ERROR] Cannot find the input file: %s\n", name); - exit(0); - } - nfps->seq = kseq_init(nfps->fp); -} - -void close_file(gz_files* nfps) -{ - kseq_destroy(nfps->seq); - gzclose(nfps->fp); -} - -void init_gz_files(hifiasm_opt_t* asm_opt) -{ - fps.idx = 0; - fps.num_reads = asm_opt->num_reads; - fps.reads = asm_opt->read_file_names; - fps.seq = NULL; - fps.fp = NULL; - if(fps.num_reads > 0) - { - open_file(&fps, fps.reads[fps.idx]); - fps.idx++; - } -} - -void destory_gz_files() -{ - close_file(&fps); -} - -int read_item() -{ - int l = kseq_read(fps.seq); - if(l >= 0 || (l < 0 && fps.idx >= fps.num_reads)) - { - return l; - } - ///l < 0 && fps.idx < fps.num_reads - close_file(&fps); - open_file(&fps, fps.reads[fps.idx]); - fps.idx++; - return read_item(); -} - - -inline void exchage_kstring_t(kstring_t* a, kstring_t* b) -{ - kstring_t tmp; - tmp = *a; - *a = *b; - *b = tmp; -} - -int get_read(kseq_t *s, int adapterLen) -{ - int l; - ///if ((l = kseq_read(seq)) >= 0) - if ((l = read_item()) >= 0) - { - exchage_kstring_t(&(fps.seq->comment), &s->comment); - exchage_kstring_t(&(fps.seq->name), &s->name); - exchage_kstring_t(&(fps.seq->qual), &s->qual); - exchage_kstring_t(&(fps.seq->seq), &s->seq); - - if(adapterLen > 0) - { - if((int)s->seq.l <= adapterLen*2) - { - s->seq.l = 0; - } - else - { - long long i; - for (i = 0; i < ((int)s->seq.l - adapterLen*2); i++) - { - s->seq.s[i] = s->seq.s[i + adapterLen]; - } - s->seq.l -= adapterLen*2; - } - - } - return 1; - } - else - { - return 0; - } -} - -void init_R_buffer_block(R_buffer_block* curr_sub_block) -{ - curr_sub_block->read = (kseq_t*)calloc(RDB.block_inner_size, sizeof(kseq_t)); - curr_sub_block->num = 0; -} - -void clear_R_buffer() -{ - RDB.all_read_end = 0; - RDB.num = 0; -} - -void init_R_buffer(int thread_num) -{ - RDB.all_read_end = 0; - RDB.num = 0; - RDB.block_inner_size = READ_BLOCK_SIZE; - RDB.size = thread_num*READ_BLOCK_NUM_PRE_THR; - - RDB.sub_block = (R_buffer_block*)malloc(sizeof(R_buffer_block)*RDB.size); - - int i = 0; - - for (i = 0; i < RDB.size; i++) - { - init_R_buffer_block(&RDB.sub_block[i]); - } - -} - -void destory_R_buffer_block(R_buffer_block* curr_sub_block) -{ - kseq_destroy(curr_sub_block->read); -} - -void destory_R_buffer() -{ - int i = 0; - - for (i = 0; i < RDB.size; i++) - { - destory_R_buffer_block(&RDB.sub_block[i]); - } - - free(RDB.sub_block); -} - - -inline void load_read_block(R_buffer_block* read_batch, int batch_read_size, - int* return_file_flag, int is_insert, int adapterLen) -{ - int inner_i = 0; - int file_flag = 1; - - while (inner_iread[inner_i], adapterLen); - - if (file_flag == 1) - { - read_batch->read[inner_i].ID = total_reads; - total_reads++; - - if (is_insert) - { - insert_read(&R_INF, &read_batch->read[inner_i].seq, - &read_batch->read[inner_i].name); - } - - inner_i++; - } - else if (file_flag == 0) - { - break; - } - } - - if (inner_i || file_flag) - { - file_flag = 1; - } - - *return_file_flag = file_flag; - read_batch->num = inner_i; -} - - -inline void push_R_block(R_buffer_block* tmp_sub_block) -{ - ///only exchange pointers - kseq_t *k1; - k1 = RDB.sub_block[RDB.num].read; - - RDB.sub_block[RDB.num].read = tmp_sub_block->read; - - tmp_sub_block->read = k1; - - RDB.sub_block[RDB.num].num = tmp_sub_block->num; - tmp_sub_block->num = 0; - - RDB.num++; -} - - -inline void pop_R_block(R_buffer_block* curr_sub_block) -{ - RDB.num--; - - ///only exchange pointers - kseq_t *k1; - k1 = RDB.sub_block[RDB.num].read; - - RDB.sub_block[RDB.num].read = curr_sub_block->read; - - curr_sub_block->read = k1; - - curr_sub_block->num = RDB.sub_block[RDB.num].num; - RDB.sub_block[RDB.num].num = 0; -} - -void* input_reads_muti_threads(void* arg) -{ - int is_insert = *((int*)arg); - - total_reads = 0; - - int file_flag = 1; - - R_buffer_block tmp_buf; - - init_R_buffer_block(&tmp_buf); - - while (1) - { - load_read_block(&tmp_buf, RDB.block_inner_size, &file_flag, is_insert, asm_opt.adapterLen); - - if (file_flag == 0) - { - break; - } - - pthread_mutex_lock(&i_readinputMutex); - while (IS_FULL(RDB)) - { - - pthread_cond_signal(&i_readinputstallCond); - pthread_cond_wait(&i_readinputflushCond, &i_readinputMutex); - } - - push_R_block(&tmp_buf); - - pthread_cond_signal(&i_readinputstallCond); - pthread_mutex_unlock(&i_readinputMutex); - } - - pthread_mutex_lock(&i_readinputMutex); - RDB.all_read_end = 1; - pthread_cond_signal(&i_readinputstallCond); //important - pthread_mutex_unlock(&i_readinputMutex); - - destory_R_buffer_block(&tmp_buf); - - fprintf(stderr, "Reads #: %lu\n", (unsigned long)total_reads); - fprintf(stderr, "Bases #: %lu\n", (unsigned long)R_INF.total_reads_bases); - - return NULL; -} - -int get_reads_mul_thread(R_buffer_block* curr_sub_block) -{ - pthread_mutex_lock(&i_readinputMutex); - - while (IS_EMPTY(RDB) && RDB.all_read_end == 0) - { - pthread_cond_signal(&i_readinputflushCond); - pthread_cond_wait(&i_readinputstallCond, &i_readinputMutex); - } - - if (!IS_EMPTY(RDB)) - { - pop_R_block(curr_sub_block); - pthread_cond_signal(&i_readinputflushCond); - pthread_mutex_unlock(&i_readinputMutex); - - return 1; - } - else - { - curr_sub_block->num = 0; - - pthread_cond_signal(&i_readinputstallCond); //important - - pthread_mutex_unlock(&i_readinputMutex); - - return 0; - } -} - void reverse_complement(char* pattern, uint64_t length) { uint64_t i = 0; @@ -962,59 +648,3 @@ void reverse_complement(char* pattern, uint64_t length) pattern[end] = RC_CHAR(pattern[end]); } } - -typedef struct { - char* tmp; - long long tmpSize; - char* dest; - long long destSize; - FILE* fp; -} LineReader; - -int get_single_line(LineReader* line) -{ - long long currentLen = 0, getLen = 0; - line->tmp[0] = '\0'; - while (fgets(line->tmp, line->tmpSize, line->fp) != NULL) - { - getLen = strlen(line->tmp); - if(getLen + currentLen >= line->destSize) - { - line->destSize = getLen + currentLen + 1; - line->dest = (char*)realloc(line->dest, line->destSize); - } - memcpy(line->dest + currentLen, line->tmp, getLen+1); - currentLen = currentLen + getLen; - if(currentLen > 0 && line->dest[currentLen - 1] == '\n') - { - return 1; - } - } - - if(currentLen > 0) - { - return 1; - } - else - { - return 0; - } -} - -void get_trio_info(char* input, uint8_t* pm) -{ - uint32_t i; - (*pm) = AMBIGU; - for (i = 0; input[i] != '\0'; i++) - { - if(input[i] == '\t') - { - break; - } - } - - input[i] = '\0'; - i++; - if(input[i] == 'p') (*pm) = FATHER; - if(input[i] == 'm') (*pm) = MOTHER; -} diff --git a/Process_Read.h b/Process_Read.h index 40e3160..a1fb5c1 100644 --- a/Process_Read.h +++ b/Process_Read.h @@ -139,21 +139,6 @@ typedef struct extern All_reads R_INF; -typedef struct -{ - kseq_t* read; - long long num; -} R_buffer_block; - -typedef struct -{ - R_buffer_block* sub_block; - long long block_inner_size; - long long size; - long long num; - int all_read_end; -} R_buffer; - typedef struct { char* seq; @@ -162,21 +147,8 @@ typedef struct long long RID; } UC_Read; -typedef struct -{ - gzFile fp; - kseq_t *seq; - char** reads; - int num_reads; - int idx; -} gz_files; - -void init_R_buffer(int thread_num); void init_All_reads(All_reads* r); void malloc_All_reads(All_reads* r); -void* input_reads_muti_threads(void*); -void init_R_buffer_block(R_buffer_block* curr_sub_block); -int get_reads_mul_thread(R_buffer_block* curr_sub_block); void ha_insert_read_len(All_reads *r, int read_len, int name_len); void ha_compress_base(uint8_t* dest, char* src, uint64_t src_l, uint64_t** N_site_lis, uint64_t N_site_occ); void init_UC_Read(UC_Read* r); @@ -189,11 +161,4 @@ void write_All_reads(All_reads* r, char* read_file_name); int load_All_reads(All_reads* r, char* read_file_name); void destory_All_reads(All_reads* r); -void destory_R_buffer_block(R_buffer_block* curr_sub_block); -void destory_R_buffer(); -void clear_R_buffer(); - -void init_gz_files(hifiasm_opt_t* asm_opt); -void destory_gz_files(); - #endif