X-Git-Url: https://git.donarmstrong.com/?a=blobdiff_plain;f=weighted.cpp;h=49bf6bf1fd2a2053ed7f3f7e0ea30fb5595400a0;hb=250e3b11b1c9c1e1ad458ab6c7e71ac2e67e11d9;hp=cf1291dfa91af84e4d821541b2a8f43a169d64f7;hpb=90708fe9701e3827e477c82fb3652539c3bf2a0d;p=mothur.git diff --git a/weighted.cpp b/weighted.cpp index cf1291d..49bf6bf 100644 --- a/weighted.cpp +++ b/weighted.cpp @@ -36,30 +36,20 @@ EstOutput Weighted::getValues(Tree* t, int p, string o) { } } - #if defined (__APPLE__) || (__MACH__) || (linux) || (__linux) || (__linux__) || (__unix__) || (__unix) - if(processors == 1){ - data = driver(t, namesOfGroupCombos, 0, namesOfGroupCombos.size(), ct); - }else{ - int numPairs = namesOfGroupCombos.size(); - - int numPairsPerProcessor = numPairs / processors; - - for (int i = 0; i < processors; i++) { - int startPos = i * numPairsPerProcessor; - if(i == processors - 1){ - numPairsPerProcessor = numPairs - i * numPairsPerProcessor; - } - lines.push_back(linePair(startPos, numPairsPerProcessor)); - } + int remainingPairs = namesOfGroupCombos.size(); + int startIndex = 0; + for (int remainingProcessors = processors; remainingProcessors > 0; remainingProcessors--) { + int numPairs = remainingPairs; //case for last processor + if (remainingProcessors != 1) { numPairs = ceil(remainingPairs / remainingProcessors); } + lines.push_back(linePair(startIndex, numPairs)); //startIndex, numPairs + startIndex = startIndex + numPairs; + remainingPairs = remainingPairs - numPairs; + } + + data = createProcesses(t, namesOfGroupCombos, ct); + + lines.clear(); - data = createProcesses(t, namesOfGroupCombos, ct); - - lines.clear(); - } - #else - data = driver(t, namesOfGroupCombos, 0, namesOfGroupCombos.size(), ct); - #endif - return data; } catch(exception& e) { @@ -71,11 +61,10 @@ EstOutput Weighted::getValues(Tree* t, int p, string o) { EstOutput Weighted::createProcesses(Tree* t, vector< vector > namesOfGroupCombos, CountTable* ct) { try { + vector processIDS; + EstOutput results; #if defined (__APPLE__) || (__MACH__) || (linux) || (__linux) || (__linux__) || (__unix__) || (__unix) int process = 1; - vector processIDS; - - EstOutput results; //loop through and create all the processes you want while (process != processors) { @@ -85,17 +74,12 @@ EstOutput Weighted::createProcesses(Tree* t, vector< vector > namesOfGro processIDS.push_back(pid); //create map from line number to pid so you can append files in correct order later process++; }else if (pid == 0){ - EstOutput Myresults; Myresults = driver(t, namesOfGroupCombos, lines[process].start, lines[process].num, ct); - //m->mothurOut("Merging results."); m->mothurOutEndLine(); - //pass numSeqs to parent ofstream out; - string tempFile = outputDir + toString(getpid()) + ".weighted.results.temp"; - m->openOutputFile(tempFile, out); out << Myresults.size() << endl; @@ -143,11 +127,48 @@ EstOutput Weighted::createProcesses(Tree* t, vector< vector > namesOfGro in.close(); m->mothurRemove(s); } +#else + + //fill in functions + vector pDataArray; + DWORD dwThreadIdArray[processors-1]; + HANDLE hThreadArray[processors-1]; + vector cts; + vector trees; - //m->mothurOut("DONE."); m->mothurOutEndLine(); m->mothurOutEndLine(); + //Create processor worker threads. + for( int i=1; icopy(ct); + Tree* copyTree = new Tree(copyCount); + copyTree->getCopy(t); + + cts.push_back(copyCount); + trees.push_back(copyTree); + + weightedData* tempweighted = new weightedData(m, lines[i].start, lines[i].num, namesOfGroupCombos, copyTree, copyCount, includeRoot); + pDataArray.push_back(tempweighted); + processIDS.push_back(i); + + hThreadArray[i-1] = CreateThread(NULL, 0, MyWeightedThreadFunction, pDataArray[i-1], 0, &dwThreadIdArray[i-1]); + } + + results = driver(t, namesOfGroupCombos, lines[0].start, lines[0].num, ct); - return results; -#endif + //Wait until all threads have terminated. + WaitForMultipleObjects(processors-1, hThreadArray, TRUE, INFINITE); + + //Close all thread handles and free memory allocations. + for(int i=0; i < pDataArray.size(); i++){ + for (int j = 0; j < pDataArray[i]->results.size(); j++) { results.push_back(pDataArray[i]->results[j]); } + delete cts[i]; + delete trees[i]; + CloseHandle(hThreadArray[i]); + delete pDataArray[i]; + } +#endif + + return results; } catch(exception& e) { m->errorOut(e, "Weighted", "createProcesses");