| 329 | } |
| 330 | |
| 331 | void SingleEndProcessor::readerTask() |
| 332 | { |
| 333 | if(mOptions->verbose) |
| 334 | loginfo("start to load data"); |
| 335 | long lastReported = 0; |
| 336 | int slept = 0; |
| 337 | long readNum = 0; |
| 338 | bool splitSizeReEvaluated = false; |
| 339 | Read** data = new Read*[PACK_SIZE]; |
| 340 | memset(data, 0, sizeof(Read*)*PACK_SIZE); |
| 341 | FastqReader reader(mOptions->in, true); |
| 342 | reader.setReadPool(mReadPool); |
| 343 | int count=0; |
| 344 | bool needToBreak = false; |
| 345 | while(true){ |
| 346 | Read* read = reader.read(); |
| 347 | if(!read || needToBreak){ |
| 348 | // the last pack |
| 349 | ReadPack* pack = new ReadPack; |
| 350 | pack->data = data; |
| 351 | pack->count = count; |
| 352 | mInputLists[mPackReadCounter % mOptions->thread]->produce(pack); |
| 353 | mPackReadCounter++; |
| 354 | data = NULL; |
| 355 | if(read) { |
| 356 | delete read; |
| 357 | read = NULL; |
| 358 | } |
| 359 | break; |
| 360 | } |
| 361 | data[count] = read; |
| 362 | count++; |
| 363 | // configured to process only first N reads |
| 364 | if(mOptions->readsToProcess >0 && count + readNum >= mOptions->readsToProcess) { |
| 365 | needToBreak = true; |
| 366 | } |
| 367 | if(mOptions->verbose && count + readNum >= lastReported + 1000000) { |
| 368 | lastReported = count + readNum; |
| 369 | string msg = "loaded " + to_string((lastReported/1000000)) + "M reads"; |
| 370 | loginfo(msg); |
| 371 | } |
| 372 | // a full pack |
| 373 | if(count == PACK_SIZE || needToBreak){ |
| 374 | ReadPack* pack = new ReadPack; |
| 375 | pack->data = data; |
| 376 | pack->count = count; |
| 377 | mInputLists[mPackReadCounter % mOptions->thread]->produce(pack); |
| 378 | mPackReadCounter++; |
| 379 | //re-initialize data for next pack |
| 380 | data = new Read*[PACK_SIZE]; |
| 381 | memset(data, 0, sizeof(Read*)*PACK_SIZE); |
| 382 | // if the processor is far behind this reader, sleep and wait to limit memory usage |
| 383 | while( mPackReadCounter - mPackProcessedCounter > PACK_IN_MEM_LIMIT){ |
| 384 | //cerr<<"sleep"<<endl; |
| 385 | slept++; |
| 386 | usleep(100); |
| 387 | } |
| 388 | readNum += count; |
nothing calls this directly
no test coverage detected