diff --git a/.vscode/c_cpp_properties.json b/.vscode/c_cpp_properties.json new file mode 100644 index 0000000..4081e86 --- /dev/null +++ b/.vscode/c_cpp_properties.json @@ -0,0 +1,18 @@ +{ + "configurations": [ + { + "name": "Win32", + "includePath": [ + "${workspaceFolder}/**", + "C:\\Users\\chhy\\Desktop\\Code\\Linux_header\\include/**" + ], + "defines": [ + "_DEBUG", + "UNICODE", + "_UNICODE" + ], + "intelliSenseMode": "msvc-x64" + } + ], + "version": 4 +} \ No newline at end of file diff --git a/.vscode/ipch/4eee59091dc293a3/PROCESS_READ.ipch b/.vscode/ipch/4eee59091dc293a3/PROCESS_READ.ipch new file mode 100644 index 0000000..35b4a77 Binary files /dev/null and b/.vscode/ipch/4eee59091dc293a3/PROCESS_READ.ipch differ diff --git a/.vscode/ipch/4eee59091dc293a3/mmap_address.bin b/.vscode/ipch/4eee59091dc293a3/mmap_address.bin new file mode 100644 index 0000000..862b842 Binary files /dev/null and b/.vscode/ipch/4eee59091dc293a3/mmap_address.bin differ diff --git a/.vscode/ipch/756c62c8d5e4771d/mmap_address.bin b/.vscode/ipch/756c62c8d5e4771d/mmap_address.bin new file mode 100644 index 0000000..862b842 Binary files /dev/null and b/.vscode/ipch/756c62c8d5e4771d/mmap_address.bin differ diff --git a/.vscode/ipch/c0e71cfe49f0fe81/ASSEMBLY.ipch b/.vscode/ipch/c0e71cfe49f0fe81/ASSEMBLY.ipch new file mode 100644 index 0000000..2dd7dc6 Binary files /dev/null and b/.vscode/ipch/c0e71cfe49f0fe81/ASSEMBLY.ipch differ diff --git a/.vscode/ipch/c0e71cfe49f0fe81/mmap_address.bin b/.vscode/ipch/c0e71cfe49f0fe81/mmap_address.bin new file mode 100644 index 0000000..862b842 Binary files /dev/null and b/.vscode/ipch/c0e71cfe49f0fe81/mmap_address.bin differ diff --git a/.vscode/ipch/f2b98741f94c3e10/MAIN.ipch b/.vscode/ipch/f2b98741f94c3e10/MAIN.ipch new file mode 100644 index 0000000..456fe16 Binary files /dev/null and b/.vscode/ipch/f2b98741f94c3e10/MAIN.ipch differ diff --git a/.vscode/ipch/f2b98741f94c3e10/mmap_address.bin b/.vscode/ipch/f2b98741f94c3e10/mmap_address.bin new file mode 100644 index 0000000..862b842 Binary files /dev/null and b/.vscode/ipch/f2b98741f94c3e10/mmap_address.bin differ diff --git a/.vscode/settings.json b/.vscode/settings.json new file mode 100644 index 0000000..bf562e6 --- /dev/null +++ b/.vscode/settings.json @@ -0,0 +1,3 @@ +{ + "C_Cpp.errorSquiggles": "Enabled" +} \ No newline at end of file diff --git a/Assembly.cpp b/Assembly.cpp index d9056d6..1a4489e 100644 --- a/Assembly.cpp +++ b/Assembly.cpp @@ -3,7 +3,7 @@ #include #include #include "Process_Read.h" - +#include "CommandLines.h" @@ -33,3 +33,107 @@ void Counting() } + + + + + + + + + + + + + + + +void* Perform_Counting(void* arg) +{ + R_buffer_block curr_sub_block; + + init_R_buffer_block(&curr_sub_block); + + long long read_number = 0; + + int file_flag = 1; + while (file_flag != 0) + { + + file_flag = get_reads_mul_thread(&curr_sub_block); + + read_number = read_number + curr_sub_block.num; + + } + + fprintf(stdout, "#########read_number: %lld\n",read_number); + fflush(stdout); + +} + + + + + + + + + + + + + + + + + +void Counting_multiple_thr() +{ + pthread_t inputReadsHandle; + + init_R_buffer(thread_num); + + pthread_create(&inputReadsHandle, NULL, input_reads_muti_threads, NULL); + + + pthread_t *_r_threads; + + _r_threads = (pthread_t *)malloc(sizeof(pthread_t)*thread_num); + + int i = 0; + + for (i = 0; i < thread_num; i++) + { + int *arg = (int*)malloc(sizeof(*arg)); + *arg = i; + + pthread_create(_r_threads + i, NULL, Perform_Counting, (void*)arg); + + } + + + + pthread_join(inputReadsHandle, NULL); + + for (i = 0; i + + extern char* read_file_name; extern char* output_file_name; extern int thread_num; diff --git a/Process_Read.cpp b/Process_Read.cpp index 6c5a164..2c266e4 100644 --- a/Process_Read.cpp +++ b/Process_Read.cpp @@ -3,10 +3,22 @@ #include #include #include +#include gzFile fp; kseq_t *seq; +R_buffer RDB; + +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; + void init_kseq(char* file) { @@ -51,9 +63,240 @@ int get_read(kseq_t *s) } +void init_R_buffer_block(R_buffer_block* curr_sub_block) +{ + int i; + + for (i = 0; i < RDB.size; i++) + { + curr_sub_block->read = (kseq_t*)calloc(RDB.block_inner_size, sizeof(kseq_t)); + curr_sub_block->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]); + } + +} + + +inline void load_read_block(R_buffer_block* read_batch, int batch_read_size, + int* return_file_flag) +{ + int inner_i = 0; + int file_flag = 1; + while (inner_iread[inner_i]); + + if (file_flag == 1) + { + 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*) +{ + + + int i = 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); + + 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); + +} + + + +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 Counting_block() +{ + + long long read_number = 0; + int i = 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); + + + if (file_flag == 0) + { + break; + } + + for (i = 0; i < tmp_buf.num; i++) + { + fprintf(stderr,"@%s\n", tmp_buf.read[i].name.s); + fprintf(stderr,"%s\n",tmp_buf.read[i].seq.s); + fprintf(stderr,"+\n"); + fprintf(stderr,"%s\n",tmp_buf.read[i].qual.s); + + read_number++; + } + + } + + fprintf(stdout, "read_number: %lld\n",read_number); + + + +} \ No newline at end of file diff --git a/Process_Read.h b/Process_Read.h index f59fe31..76762d5 100644 --- a/Process_Read.h +++ b/Process_Read.h @@ -7,6 +7,13 @@ #include #include "kseq.h" +#define READ_BLOCK_SIZE 64 +#define READ_BLOCK_NUM_PRE_THR 32 + +#define IS_FULL(buffer) ((buffer.num >= buffer.size)?1:0) +#define IS_EMPTY(buffer) ((buffer.num == 0)?1:0) + + KSEQ_INIT(gzFile, gzread) @@ -15,4 +22,29 @@ void destory_kseq(); int get_read(kseq_t *s); + +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; + +void init_R_buffer(int thread_num); +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 Counting_block(); + #endif diff --git a/main.cpp b/main.cpp index a3fe80a..3484607 100644 --- a/main.cpp +++ b/main.cpp @@ -15,7 +15,8 @@ int main(int argc, char *argv[]) - Counting(); + ///Counting(); + Counting_multiple_thr();