afin/process.cpp at master · afinit/afin · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
// $Author: benine $
// $Date$
// $Log$
// Contains the Process class for afin

#include "process.hpp"
#include <thread>

using namespace std;

int max_search_loops;
int contig_sub_len;
int extend_len;
int max_sort_char;
int min_cov;
int min_overlap;
int max_threads;
int initial_trim;
int max_missed;
double stop_ext;
bool test_run;
int print_fused;
int screen_output;
int log_output;
int verbose;
int no_fusion;
double mismatch_threshold;
mutex log_mut;

////////////////////////////////////
//////// PROCESS DEFINITIONS ///////
////////////////////////////////////

// Process constructor
Process::Process(){
  outfile = "afin_out";
  readsfiles = "";
  contigsfiles = "";
  reads = 0;
  contigs = 0;
  fuse = 0;
}

Process::~Process(){
  delete fuse;
  delete reads;
  delete contigs;
}

// initalize logfile
void Process::logfile_init(){
  Log::Inst()->open_log( outfile + ".log" );

  // output starting option values
  Log::Inst()->log_it( "OPTION VALUES" );
  Log::Inst()->log_it( "contig_sub_len: " + to_string(contig_sub_len) );
  Log::Inst()->log_it( "extend_len: " + to_string(extend_len) );
  Log::Inst()->log_it( "max_search_loops: " + to_string(max_search_loops) );
  Log::Inst()->log_it( "max_sort_char: " + to_string(max_sort_char) );
  Log::Inst()->log_it( "min_cov: " + to_string(min_cov) );
  Log::Inst()->log_it( "min_overlap: " + to_string(min_overlap) );
  Log::Inst()->log_it( "initial_trim: " + to_string(initial_trim) );
  Log::Inst()->log_it( "max_missed: " + to_string(max_missed) );
  Log::Inst()->log_it( "mismatch_threshold: " + to_string(mismatch_threshold) );
  Log::Inst()->log_it( "max_threads: " + to_string(max_threads) );
}

// Initializes data structures and turns over control to run_manager()
void Process::start_run(){
  // initialize logfile if logging enabled
  if( log_output || screen_output ){
    logfile_init();
  }

  // prevent printing of
  if( no_fusion )
    print_fused = 0;

  // log output file
  Log::Inst()->log_it( string("output file: ") + outfile );

  // initialize objects
  reads = new Readlist( readsfiles );
  contigs = new Contiglist( reads, contigsfiles, outfile );
  fuse = new Fusion( contigs, reads );

  Log::Inst()->log_it( "End initialization phase" );

  // make initial attempt to fuse contigs
  if( ! no_fusion )
    fuse->run_fusion( true );

  if( test_run )
    contigs->output_contigs( 0, outfile + ".fus", "mid" );

  run_manager();

  contigs->create_final_fasta();
}

// Manages run
void Process::run_manager(){
  // create thread array with max_thread entries
  vector<thread> t;
  Queue<int> qu;

  // loop max search loops
  for( int j=0; j<max_search_loops; j++ ){
    Log::Inst()->log_it( "Begin Extensions" );

    // initialize threads
    for( int i=0; i<max_threads; i++ ){
      t.push_back(thread( &Process::thread_worker, this, ref(qu), i ));
    }

    if( test_run ){
      Log::Inst()->log_it( "contigs.size(): " + to_string(contigs->get_list_size()) + " max_threads: " + to_string(max_threads) );
    }

    // push each thread onto queue
    for( int i=0; i<contigs->get_list_size(); i++ ){
      qu.push( i );
    }

    // push stop signals onto queue for each thread
    for( int i=0; i<max_threads; i++ ){
      qu.push( -1 );
    }

    // join threads
    for( int i=0; i<max_threads; i++ ){
      t[i].join();
    }

    // remove threads from vector
    t.erase( t.begin(), t.begin()+max_threads );

    // removed for master branch until algorithm can be adjusted
    if( ! no_fusion )
      fuse->run_fusion( false );

    if( test_run ){
      contigs->output_contigs( 0, outfile + ".fus" + to_string(j), "mid" );
    }
  }
}

// Consume function which will act as the driver for an individual thread
void Process::thread_worker( Queue<int>& q, unsigned int id) {
  for (;;) {
    auto item = q.pop();
    if( item == -1 ){
      break;
    }
    else{
      contigs->get_contig_ref(item)->extend( false );
      contigs->get_contig_ref(item)->extend( true );
    }
  }
}

//////////////////////////
// END PROCESS ///////////
//////////////////////////