wip: fixing thread delay issues

gpertea · gpertea · commit b94793d804a1 · 2015-04-29T16:48:58.000-04:00
diff --git a/rlink.h b/rlink.h
@@ -370,72 +370,13 @@ struct BundleData {
  }
 };
 
-/*
-struct BundleData {
- BundleStatus status;
- int numreads;
- //GBamReader bamreader;
- int64_t bamStart; //start of bundle in BAM file
- char firstPass; //0=2nd pass, 1=1st pass, 2=single pass requested
- int idx; //index in the main bundles array
- int start;
- int end;
- bool covSaturated;
- GStr refseq;
- GList<CReadAln> readlist;
- GVec<float> bpcov;
- GList<CJunction> junction;
- GPVec<GffObj> keepguides;
- GList<CPrediction> pred;
- BundleData():status(BUNDLE_STATUS_CLEAR), numreads(0), bamStart(-1), 
-         firstPass(singlePass ? 2 : 1),
-		 idx(0), start(0), end(0), covSaturated(false), refseq(), readlist(false,true),
-		 bpcov(1024), junction(true, true, true), keepguides(false), pred(false) { }
-
-  void getReady(int currentstart, int currentend, GStr& ref) {
-	 start=currentstart;
-	 end=currentend;
-	 refseq=ref;
-	 status=BUNDLE_STATUS_READY;
-  }
-
- void Clear() {
-	keepguides.Clear();
-	pred.Clear();
-	status=BUNDLE_STATUS_CLEAR;
-	numreads=0;
-	bamStart=-1;
-	firstPass = singlePass ? 2 : 1;
-	start=0;
-	end=0;
-	covSaturated=false;
-	refseq="";
-	readlist.Clear();
-	bpcov.Clear();
-	bpcov.setCapacity(1024);
-	junction.Clear();
- }
-
- ~BundleData() {
-	Clear();
- }
-};
-*/
 int processRead(int currentstart, int currentend, BundleData& bdata,
 		 GHash<int>& hashread, GBamRecord& brec, char strand, int nh, int hi);
 
 void countRead(BundleData& bdata, GBamRecord& brec, int hi);
 
-//int process_read(int currentstart, int currentend, GList<CReadAln>& readlist, GHash<int>& hashread,
-//		GList<CJunction>& junction, GBamRecord& brec, char strand, int nh, int hi, GVec<float>& bpcov);
-
 int printResults(BundleData* bundleData, int ngenes, int geneno, GStr& refname);
 
-//int print_transcripts(GList<CPrediction>& pred, int ngenes, int geneno, GStr& refname);
-
-//int infer_transcripts(int refstart, GList<CReadAln>& readlist,
-//		GList<CJunction>& junction, GPVec<GffObj>& guides, GVec<float>& bpcov, GList<CPrediction>& pred, bool fast);
-
 int infer_transcripts(BundleData* bundle, bool fast);
 
 // --- utility functions
diff --git a/stringtie.cpp b/stringtie.cpp
@@ -137,21 +137,26 @@ int refseqCount=0;
 
 #ifndef NOTHREADS
 
-GFastMutex dataMutex; //to manage availability of data records ready to be loaded by main thread
+GFastMutex dataMutex; //manage availability of data records ready to be loaded by main thread
 GVec<int> dataClear; //indexes of data bundles cleared for loading by main thread (clear data pool)
 
-GFastMutex waitMutex; // for main program to make sure there are threads ready/waiting
+GFastMutex waitMutex; // controls threadsWaiting (idle threads counter)
 int threadsWaiting=0; // how many worker threads are waiting
 
+GMutex queueMutex; //controls bundleQueue and bundleWork access
+int bundleWork=1; // bit 0 set if bundles are still being prepared (BAM file not exhausted yet)
+                  // bit 1 set if there are Bundles ready in the queue
+GConditionVar haveBundles; //will notify all threads when bundles are pushed in the ready queue
+                           //or no more bundles are coming
+
+
+
 GFastMutex printMutex; //for writing the output to file
+
 GFastMutex logMutex; //only when verbose - to avoid mangling the log output
-GMutex queueMutex; //whenever bundleQueue is updated
+
 GFastMutex bamReadingMutex;
-GConditionVar haveBundles; //will notify all threads when bundles are pushed in the ready queue
-                           //or no more bundles are coming
 
-int bundleWork=1; // bit 0 set if bundles are still being prepared (BAM file not exhausted yet)
-                  // bit 1 set if there are Bundles ready in the queue
 #endif
 
 bool NoMoreBundles=false;
@@ -285,9 +290,13 @@ const char* ERR_BAM_SORT="\nError: the input alignment file is not sorted!\n";
 if (ballgown)
  Ballgown_setupFiles(f_tdata, f_edata, f_idata, f_e2t, f_i2t);
 #ifndef NOTHREADS
- GThread* threads=new GThread[num_cpus];
- GPVec<BundleData> bundleQueue(false);
- BundleData* bundles=new BundleData[num_cpus+1]; //extra one being prepared while all others are processed
+ GThread* threads=new GThread[num_cpus]; //bundle processing threads
+
+ GPVec<BundleData> bundleQueue(false); //queue of loaded bundles
+
+ BundleData* bundles=new BundleData[num_cpus+1];
+ //bundles[1..num_cpus] are processed by threads, bundles[0] is being prepared
+
  dataClear.setCapacity(num_cpus+1);
  for (int b=0;b<num_cpus;b++) {
 	 threads[b].kickStart(workerThread, (void*) &bundleQueue);
@@ -366,20 +375,28 @@ if (ballgown)
 			}*/
 			bundle->getReady(currentstart, currentend);
 #ifndef NOTHREADS
-			//push this in the bundle queue, where it'll be picked up by the threads
+			//push this in the bundle queue where it'll be picked up by the threads
 			DBGPRINT2("##> Locking queueMutex to push loaded bundle into the queue (bundle.start=%d)\n", bundle->start);
+			int qCount=0;
 			queueMutex.lock();
-			bundleQueue.Push(bundle);
-			bundleWork |= 0x02; //set bit 1
-			int qCount=bundleQueue.Count();
+			  //push the bundle in the processing queue
+			  bundleQueue.Push(bundle);
+			  bundleWork |= 0x02; //set bit 1
+			  qCount=bundleQueue.Count();
+			  DBGPRINT2("##> bundleQueue.Count()=%d)\n", qCount);
 			queueMutex.unlock();
+			/*
 			do {
 			     waitForThreads();
 			     DBGPRINT("##> NOTIFY any thread...\n");
 			     haveBundles.notify_one();
 			     //this_thread::sleep_for(chrono::milliseconds(1));
-			     sleep(0);
+			     //sleep(0);
 			} while (!queuePopped(bundleQueue, qCount));
+			*/
+			waitForThreads(); //wait for any threads to become available
+			queuePopped(bundleQueue, qCount); //pop the next bundle from the queue and process it
+
 #else //no threads
 			Num_Fragments+=bundle->num_fragments;
 			Frag_Len+=bundle->frag_len;
@@ -394,7 +411,7 @@ if (ballgown)
 	dataClear.Push(bundle->idx);
 	dataMutex.unlock();
 #endif
-		 }
+		 } //nothing to do with this bundle
 
 		 if (chr_changed) {
 			 if (guided) {
@@ -450,25 +467,15 @@ if (ballgown)
 				 ng_ovlstart++;
 			 }
 			 if (ng_ovlstart>ng_start) ng_end=ng_ovlstart-1;
-				 /*
-				 while(ng_end+1<ng && (int)(*guides)[ng_end+1]->start<=pos) {
-					 ng_end++;
-					 if(currentend<(int)(*guides)[ng_end]->end) {
-						 currentend=(*guides)[ng_end]->end;
-					 }
-				 }
-				 */
 		 } //guides present on the current chromosome
 		bundle->refseq=lastref;
 		bundle->start=currentstart;
 		bundle->end=currentend;
-	 } //<---- new bundle
-	 //currentend=process_read(currentstart, currentend, bundle->readlist, hashread,
-		//	 bundle->junction, *brec, strand, nh, hi, bundle->bpcov);
-     //currentend=
+	 } //<---- new bundle just started
+
 	 if (currentend<(int)brec->end) {
-		 //current read just pushed upper boundary of the bundle
-		 //this might never happen if a longer guide was added already to the bundle
+		 //current read extends the bundle
+		 //this might not happen if a longer guide had already been added to the bundle
 		 currentend=brec->end;
 		 if (guides) { //add any newly overlapping guides to bundle
 			 bool cend_changed;
@@ -969,8 +976,10 @@ bool waitForThreads() {
 	  waitMutex.lock();
 	  noneWaiting=(threadsWaiting<1);
 	  waitMutex.unlock();
-	  if (noneWaiting)
-	    this_thread::sleep_for(chrono::milliseconds(30));
+	  if (noneWaiting) {
+		DBGPRINT("##>none waiting, sleep_for(2ms)\n");
+	    this_thread::sleep_for(chrono::milliseconds(2));
+	  }
 	}
  DBGPRINT("##> there are workers ready now.\n");
  return(!noneWaiting);
@@ -1033,7 +1042,7 @@ int waitForData(BundleData* bundles) {
 	    return bidx;
 	    }
 	  dataMutex.unlock();
-	  this_thread::sleep_for(chrono::milliseconds(20));
+	  //this_thread::sleep_for(chrono::milliseconds(20));
 	}
 	return -1; // should NEVER happen
 }
diff --git a/tablemaker.cpp b/tablemaker.cpp
@@ -31,7 +31,6 @@ void BundleData::rc_store_t(GffObj* t) {
 	  rc_init(t);
 	}
 	rc_data->addTranscript(*t);
- //check this read alignment against ref exons and introns
 }
 
 struct COvlSorter {
diff --git a/tablemaker.h b/tablemaker.h
@@ -215,36 +215,21 @@ struct RC_BundleData {
  int init_lmin;
  int lmin;
  int rmax;
- //set<RC_ScaffData> tdata; //all transcripts in this bundle
- //map<uint, set<uint> > e2t; //mapping exon ID to transcript IDs
- //map<uint, set<uint> > i2t; //mapping intron ID to transcript IDs
- //set<RC_Feature> exons; //all exons in this bundle, by their start coordinate
- //set<RC_Feature> introns; //all introns in this bundle, by their start coordinate
- //GList<RC_ScaffData> tdata;
- GPVec<RC_ScaffData> tdata;
- GList<RC_Feature> exons;
- GList<RC_Feature> introns;
+ GPVec<RC_ScaffData> tdata; //all transcripts in this bundle
+ GList<RC_Feature> exons; //all unique exons in this bundle, by their start coordinate
+ GList<RC_Feature> introns; //all unique introns in this bundle, by their start coordinate
  //RC_FeatIt xcache; //cache the first exon overlapping xcache_pos to speed up exon-overlap queries (findExons())
  int xcache; //exons index of the first exon overlapping xcache_pos
  int xcache_pos; // left coordinate of last cached exon overlap query (findExons())
- // -- output files
- /*
- FILE* ftdata; //t_data
- FILE* fedata; //e_data
- FILE* fidata; //i_data
- FILE* fe2t;   //e2t
- FILE* fi2t;   //i2t
- */
  vector<float> f_mcov; //coverage data, multi-map aware, per strand
  vector<int> f_cov;
  vector<float> r_mcov; //coverage data on the reverse strand
  vector<int> r_cov;
  //
  RC_BundleData(int t_l=0, int t_r=0):init_lmin(0), lmin(t_l), rmax(t_r),
 	 tdata(false), // e2t(), i2t(), exons(), introns(),
-	 exons(true, false, true), introns(true,false,true),
+	 exons(true, false, true), introns(true, false, true),
 	 xcache(0), xcache_pos(0)
-     //, ftdata(NULL), fedata(NULL), fidata(NULL), fe2t(NULL), fi2t(NULL)
 	 {
 	 if (rmax>lmin) updateCovSpan();
  }
@@ -256,25 +241,6 @@ struct RC_BundleData {
 	 r_mcov.clear();
  }
 
-/*
- void addBundleFeature(uint t_id, int l, int r, char strand, uint f_id, set<RC_Feature>& fset,
-	                         map<uint, set<uint> >& f2t) {
-   RC_Feature feat(l, r, strand, f_id);
-   fset.insert(feat);
-   //pair<RC_FeatIt, bool> in = fset.insert(feat);
-   //if (!in.second) { //existing f_id
-   // f_id=in.first->id;
-   //}
-   set<uint> tset;
-   tset.insert(t_id);
-   pair<RC_Map2SetIt, bool> mapin=f2t.insert(pair<uint, set<uint> >(f_id, tset));
-   if (!mapin.second) {
-	 //existing f_id
-	 (*mapin.first).second.insert(t_id);
-   }
-  }
-*/
-
  void addTranscript(GffObj& t) {
    //if (!ps.rc_id_data()) return;
    //RC_ScaffIds& sdata = *(ps.rc_id_data());
@@ -288,12 +254,9 @@ struct RC_BundleData {
    if (boundary_changed) updateCovSpan();
    //for (vector<RC_ScaffSeg>::iterator it=sdata.exons.begin();it!=sdata.exons.end();++it) {
    for (int i=0;i<sdata.t_exons.Count();i++) {
-	 //addBundleFeature(sdata.t_id, sdata.exons[i], sdata.strand, exons);
 	 exons.Add(sdata.t_exons[i]);
    }
    //store introns:
-   //for (vector<RC_ScaffSeg>::iterator it=sdata.introns.begin();it!=sdata.introns.end();++it) {
-   //   addBundleFeature(sdata.t_id, it->l, it->r, sdata.strand, it->id, introns, i2t);
    for (int i=0;i<sdata.t_introns.Count();i++) {
 	   introns.Add(sdata.t_introns[i]);
    }

Original file line number	Diff line number	Diff line change
`@@ -31,7 +31,6 @@ void BundleData::rc_store_t(GffObj* t) {`
`31`	`31`	`rc_init(t);`
`32`	`32`	`}`
`33`	`33`	`rc_data->addTranscript(*t);`
`34`		`- //check this read alignment against ref exons and introns`
`35`	`34`	`}`
`36`	`35`
`37`	`36`	`struct COvlSorter {`