+ //do your part
+ driverMPIRun(startIndex, numSeqsPerProcessor, inMPI, outMPI, MPIPos);
+
+ if (m->control_pressed) { MPI_File_close(&inMPI); MPI_File_close(&outMPI); return 0; }
+
+ //wait on chidren
+ for(int i = 1; i < processors; i++) {
+ char buf[5];
+ MPI_Recv(buf, 5, MPI_CHAR, i, tag, MPI_COMM_WORLD, &status);
+ }
+
+ }else { //you are a child process
+ MPI_Recv(&num, 1, MPI_INT, 0, tag, MPI_COMM_WORLD, &status);
+ MPIPos.resize(num+1);
+ numSeqs += num;
+ MPI_Recv(&MPIPos[0], (num+1), MPI_LONG, 0, tag, MPI_COMM_WORLD, &status);
+
+ //figure out how many sequences you have to align
+ numSeqsPerProcessor = num / processors;
+ int startIndex = pid * numSeqsPerProcessor;
+ if(pid == (processors - 1)){ numSeqsPerProcessor = num - pid * numSeqsPerProcessor; }
+
+
+ //align your part
+ driverMPIRun(startIndex, numSeqsPerProcessor, inMPI, outMPI, MPIPos);
+
+ if (m->control_pressed) { MPI_File_close(&inMPI); MPI_File_close(&outMPI); return 0; }
+
+ char buf[5];
+ strcpy(buf, "done");
+
+ //tell parent you are done.
+ MPI_Send(buf, 5, MPI_CHAR, 0, tag, MPI_COMM_WORLD);
+ }
+
+ MPI_File_close(&outMPI);
+ MPI_File_close(&inMPI);
+ MPI_Barrier(MPI_COMM_WORLD); //make everyone wait - just in case
+
+#else
+
+ vector<unsigned long long> positions;
+ if (savedPositions.size() != 0) { positions = savedPositions[s]; }
+ else {
+#if defined (__APPLE__) || (__MACH__) || (linux) || (__linux) || (__linux__) || (__unix__) || (__unix)
+ positions = m->divideFile(fastafileNames[s], processors);
+#else
+ if(processors != 1){
+ int numFastaSeqs = 0;
+ positions = m->setFilePosFasta(fastafileNames[s], numFastaSeqs);
+ if (positions.size() < processors) { processors = positions.size(); }
+ }
+#endif
+ }
+ #if defined (__APPLE__) || (__MACH__) || (linux) || (__linux) || (__linux__) || (__unix__) || (__unix)
+ //vector<unsigned long long> positions = m->divideFile(fastafileNames[s], processors);
+
+ for (int i = 0; i < (positions.size()-1); i++) {
+ lines.push_back(new linePair(positions[i], positions[(i+1)]));
+ }
+
+ if(processors == 1){
+ int numFastaSeqs = driverRunFilter(filter, filteredFasta, fastafileNames[s], lines[0]);
+ numSeqs += numFastaSeqs;
+ }else{
+ int numFastaSeqs = createProcessesRunFilter(filter, fastafileNames[s], filteredFasta);
+ numSeqs += numFastaSeqs;
+ }
+
+ if (m->control_pressed) { return 1; }
+ #else
+ if(processors == 1){
+ lines.push_back(new linePair(0, 1000));
+ int numFastaSeqs = driverRunFilter(filter, filteredFasta, fastafileNames[s], lines[0]);
+ numSeqs += numFastaSeqs;
+ }else {
+ int numFastaSeqs = positions.size()-1;
+ //positions = m->setFilePosFasta(fastafileNames[s], numFastaSeqs);
+
+ //figure out how many sequences you have to process
+ int numSeqsPerProcessor = numFastaSeqs / processors;
+ for (int i = 0; i < processors; i++) {
+ int startIndex = i * numSeqsPerProcessor;
+ if(i == (processors - 1)){ numSeqsPerProcessor = numFastaSeqs - i * numSeqsPerProcessor; }
+ lines.push_back(new linePair(positions[startIndex], numSeqsPerProcessor));
+ }
+
+ numFastaSeqs = createProcessesRunFilter(filter, fastafileNames[s], filteredFasta);
+ numSeqs += numFastaSeqs;
+ }
+
+ if (m->control_pressed) { return 1; }
+ #endif
+#endif
+ outputNames.push_back(filteredFasta); outputTypes["fasta"].push_back(filteredFasta);
+ }
+
+ return 0;
+ }
+ catch(exception& e) {
+ m->errorOut(e, "FilterSeqsCommand", "filterSequences");
+ exit(1);
+ }
+}
+#ifdef USE_MPI
+/**************************************************************************************/
+int FilterSeqsCommand::driverMPIRun(int start, int num, MPI_File& inMPI, MPI_File& outMPI, vector<unsigned long long>& MPIPos) {
+ try {
+ string outputString = "";
+ int count = 0;
+ MPI_Status status;
+
+ for(int i=0;i<num;i++){
+
+ if (m->control_pressed) { return 0; }
+
+ //read next sequence
+ int length = MPIPos[start+i+1] - MPIPos[start+i];
+ char* buf4 = new char[length];
+ MPI_File_read_at(inMPI, MPIPos[start+i], buf4, length, MPI_CHAR, &status);
+
+ string tempBuf = buf4;
+ if (tempBuf.length() > length) { tempBuf = tempBuf.substr(0, length); }
+ istringstream iss (tempBuf,istringstream::in);
+ delete buf4;
+
+ Sequence seq(iss); m->gobble(iss);
+
+ if (seq.getName() != "") {
+ string align = seq.getAligned();
+ string filterSeq = "";
+
+ for(int j=0;j<alignmentLength;j++){
+ if(filter[j] == '1'){
+ filterSeq += align[j];
+ }
+ }
+
+ count++;
+ outputString += ">" + seq.getName() + "\n" + filterSeq + "\n";
+
+ if(count % 10 == 0){ //output to file
+ //send results to parent
+ int length = outputString.length();
+ char* buf = new char[length];
+ memcpy(buf, outputString.c_str(), length);
+
+ MPI_File_write_shared(outMPI, buf, length, MPI_CHAR, &status);
+ outputString = "";
+ delete buf;
+ }
+
+ }
+
+ if((i+1) % 100 == 0){ cout << (i+1) << endl; }
+ }
+
+ if(outputString != ""){ //output to file
+ //send results to parent
+ int length = outputString.length();
+ char* buf = new char[length];
+ memcpy(buf, outputString.c_str(), length);
+
+ MPI_File_write_shared(outMPI, buf, length, MPI_CHAR, &status);
+ outputString = "";
+ delete buf;
+ }
+
+ if((num) % 100 != 0){ cout << (num) << endl; }
+
+ return 0;
+ }
+ catch(exception& e) {
+ m->errorOut(e, "FilterSeqsCommand", "driverRunFilter");
+ exit(1);
+ }
+}
+#endif
+/**************************************************************************************/
+int FilterSeqsCommand::driverRunFilter(string F, string outputFilename, string inputFilename, linePair* filePos) {
+ try {
+ ofstream out;
+ m->openOutputFile(outputFilename, out);
+
+ ifstream in;
+ m->openInputFile(inputFilename, in);
+
+ in.seekg(filePos->start);
+
+ bool done = false;
+ int count = 0;
+
+ while (!done) {
+
+ if (m->control_pressed) { in.close(); out.close(); return 0; }
+
+ Sequence seq(in); m->gobble(in);