sparqlGraphLibrary/src/main/java/com/ge/research/semtk/sparqlX/dispatch/EdcDispatcher.java
private int runEdcThreads(Table queryGenTable, int currentStatus, int ceiling) throws Exception{
		
		if(queryGenTable.getNumRows() == 0){
			throw new Exception("Query generator indicates no work to perform");
		}
		
		// note (for updates to the status service, we have a range of 50 percentage units to play with: 20% to 70%
		float increment = 50 / queryGenTable.getNumRows();
		if(increment < 1){ increment = 1; }
		int highestStatPercent = ceiling;
		
		int numberThreadsCompletedSuccessfully = 0;
		String queryExecClientName = "unknown";
		// array of entries to determine success count. 
		Exception[] threadExceptionArr = new Exception[queryGenTable.getNumRows()];
		
		LocalLogger.logToStdErr("About to start edc work");
		
		// create the group of threads 
		int numQueries = queryGenTable.getNumRows();   // the total we will try.
		DispatcherWorkThread[] spunup = new DispatcherWorkThread[numQueries]; // the array of them. 
		try{
			int threadLimit = MAX_NUMBER_SIMULTANEOUS_QUERIES_PER_USER;
			int finished = 0;
			int iteration = 0;
			int running = 0;
			int currentReq = 0;
			while(finished < numQueries){
				if(running == numQueries){
					// nothing left to do... just wait for them to finish. 
					break;
				}
				if(running >= threadLimit || currentReq >= numQueries){
					for(int joined = 0; joined < threadLimit; joined += 1){
						try {
							spunup[joined + (threadLimit * iteration)].join();
							finished += 1;
							running = running - 1;
							this.jobTracker.incrementPercentComplete(this.jobID, Math.round(increment), highestStatPercent);
							if(finished >= numQueries){break;}
						} catch (Exception e) {
							LocalLogger.printStackTrace(e);
							throw new IOException("(Join failure in dispatch threading : joined value was "+ joined + ":: " + e.getClass().toString() + " : "+ e.getMessage() + ")");
						}
					}
					if(currentReq%threadLimit == 0){ iteration += 1;}
				}
				else{
					if (currentReq < numQueries) {
						try {
							// get the right DispatchQueryGroup
							DispatchQueryGroup currDQG = this.queryGroupColl.getGroupByUUID(UUID.fromString(queryGenTable.getCell(currentReq, Utility.COL_NAME_UUID)));
													
							// get the JSON config from the generator return statement
							JSONObject configJSON = null;
							JSONParser jParser = new JSONParser();
							
							configJSON = (JSONObject) jParser.parse(queryGenTable.getCell(currentReq, Utility.COL_NAME_CONFIGJSON));
	
							String currentJobId = this.getJobId() + "_edc_" + String.valueOf(currentReq);
							LocalLogger.logToStdErr("config for current qry:");
							LocalLogger.logToStdErr(configJSON.toString() + " jobId: " + currentJobId);
							
							QueryExecuteClient clnt = this.dispatchServiceMgr.getExecuteClient(configJSON, currentJobId );
							ResultsClient rClient = new ResultsClient(this.resConfig);
							queryExecClientName = clnt.getClass().getSimpleName();
							JobTracker tracker = new JobTracker(this.jobTrackerSei);
							String query = queryGenTable.getColumn(Utility.COL_NAME_QUERY)[currentReq];
							
							// TESTING ONLY
							// clnt.getConfig().setServiceServer("localhost");
							
							spunup[currentReq] = new DispatcherWorkThread(currDQG, query, clnt, tracker, rClient, threadExceptionArr, currentReq);
							LocalLogger.logToStdErr("Starting EDC thread " + currentReq + " of a total of " + numQueries);
							spunup[currentReq].start();
						} catch (Exception EEE) {
							throw new Exception("spin up of query thread failed. reported:" + EEE.getMessage());
						}
						// update the counters
						running += 1;
						currentReq += 1;	
					}	
				}
			
			}
			// make sure it all really closed.
			for(int joined = 0; joined < running; joined += 1){
				try {
					spunup[joined + (iteration * threadLimit)].join();
					this.jobTracker.incrementPercentComplete(this.jobID, Math.round(increment), highestStatPercent);
					continue;
				} catch (Exception e) {
					LocalLogger.printStackTrace(e);
					throw new Exception("(Join failure in dispatch threading : joined value was "+ joined + ":: " + e.getClass().toString() + " : "+ e.getMessage() + ")");
				}
			}
		}
		catch(Exception e){
			throw new Exception("Dispatcher threading failure: " + e.getMessage());
		}
		
		// determine if any threads failed 
		for(Exception e : threadExceptionArr){
			if(e == null) { 
				numberThreadsCompletedSuccessfully += 1; 
			} else {
				// at least one thread failed - throw a new Exception based on the first failure
				String message = e.getMessage();
				int MAX_MESSAGE_LENGTH = 500;  	// this message will get displayed in a dialog, so limit its length
				if(message != null && message.length() > MAX_MESSAGE_LENGTH){  
					message = message.substring(0, MAX_MESSAGE_LENGTH) + "...";
				}
				throw new Exception(queryExecClientName + " could not retrieve data: " + message, e);
			}
		}
		
		return numberThreadsCompletedSuccessfully;
	}
/** 
	 * execute the threads to be completed. periodically, update the status
	 * @param queryGenTable
	 * @return
	 * @throws Exception 
	 */
execute the threads to be completed. periodically, update the status
@param queryGenTable
@return
@throws Exception
execute the threads to be completed . periodically update the status @param queryGenTable @return @throws Exception
private int runEdcThreads(Table queryGenTable, int currentStatus, int ceiling) throws Exception{
		if(queryGenTable.getNumRows() == 0){
			throw new Exception("Query generator indicates no work to perform");
		}
		float increment = 50 / queryGenTable.getNumRows();
		if(increment < 1){ increment = 1; }
		int highestStatPercent = ceiling;
		int numberThreadsCompletedSuccessfully = 0;
		String queryExecClientName = "unknown";
		Exception[] threadExceptionArr = new Exception[queryGenTable.getNumRows()];
		LocalLogger.logToStdErr("About to start edc work");
		int numQueries = queryGenTable.getNumRows();   
		DispatcherWorkThread[] spunup = new DispatcherWorkThread[numQueries]; 
		try{
			int threadLimit = MAX_NUMBER_SIMULTANEOUS_QUERIES_PER_USER;
			int finished = 0;
			int iteration = 0;
			int running = 0;
			int currentReq = 0;
			while(finished < numQueries){
				if(running == numQueries){
					break;
				}
				if(running >= threadLimit || currentReq >= numQueries){
					for(int joined = 0; joined < threadLimit; joined += 1){
						try {
							spunup[joined + (threadLimit * iteration)].join();
							finished += 1;
							running = running - 1;
							this.jobTracker.incrementPercentComplete(this.jobID, Math.round(increment), highestStatPercent);
							if(finished >= numQueries){break;}
						} catch (Exception e) {
							LocalLogger.printStackTrace(e);
							throw new IOException("(Join failure in dispatch threading : joined value was "+ joined + ":: " + e.getClass().toString() + " : "+ e.getMessage() + ")");
						}
					}
					if(currentReq%threadLimit == 0){ iteration += 1;}
				}
				else{
					if (currentReq < numQueries) {
						try {
							DispatchQueryGroup currDQG = this.queryGroupColl.getGroupByUUID(UUID.fromString(queryGenTable.getCell(currentReq, Utility.COL_NAME_UUID)));
							JSONObject configJSON = null;
							JSONParser jParser = new JSONParser();
							configJSON = (JSONObject) jParser.parse(queryGenTable.getCell(currentReq, Utility.COL_NAME_CONFIGJSON));
							String currentJobId = this.getJobId() + "_edc_" + String.valueOf(currentReq);
							LocalLogger.logToStdErr("config for current qry:");
							LocalLogger.logToStdErr(configJSON.toString() + " jobId: " + currentJobId);
							QueryExecuteClient clnt = this.dispatchServiceMgr.getExecuteClient(configJSON, currentJobId );
							ResultsClient rClient = new ResultsClient(this.resConfig);
							queryExecClientName = clnt.getClass().getSimpleName();
							JobTracker tracker = new JobTracker(this.jobTrackerSei);
							String query = queryGenTable.getColumn(Utility.COL_NAME_QUERY)[currentReq];
							spunup[currentReq] = new DispatcherWorkThread(currDQG, query, clnt, tracker, rClient, threadExceptionArr, currentReq);
							LocalLogger.logToStdErr("Starting EDC thread " + currentReq + " of a total of " + numQueries);
							spunup[currentReq].start();
						} catch (Exception EEE) {
							throw new Exception("spin up of query thread failed. reported:" + EEE.getMessage());
						}
						running += 1;
						currentReq += 1;	
					}	
				}
			}
			for(int joined = 0; joined < running; joined += 1){
				try {
					spunup[joined + (iteration * threadLimit)].join();
					this.jobTracker.incrementPercentComplete(this.jobID, Math.round(increment), highestStatPercent);
					continue;
				} catch (Exception e) {
					LocalLogger.printStackTrace(e);
					throw new Exception("(Join failure in dispatch threading : joined value was "+ joined + ":: " + e.getClass().toString() + " : "+ e.getMessage() + ")");
				}
			}
		}
		catch(Exception e){
			throw new Exception("Dispatcher threading failure: " + e.getMessage());
		}
		for(Exception e : threadExceptionArr){
			if(e == null) { 
				numberThreadsCompletedSuccessfully += 1; 
			} else {
				String message = e.getMessage();
				int MAX_MESSAGE_LENGTH = 500;  	
				if(message != null && message.length() > MAX_MESSAGE_LENGTH){  
					message = message.substring(0, MAX_MESSAGE_LENGTH) + "...";
				}
				throw new Exception(queryExecClientName + " could not retrieve data: " + message, e);
			}
		}
		return numberThreadsCompletedSuccessfully;
	}
private int runEdcThreads ( Table queryGenTable , int currentStatus , int ceiling ) throws Exception { if ( queryGenTable . getNumRows ( ) == 0 ) { throw new Exception ( "Query generator indicates no work to perform" ) ; } float increment = 50 / queryGenTable . getNumRows ( ) ; if ( increment < 1 ) { increment = 1 ; } int highestStatPercent = ceiling ; int numberThreadsCompletedSuccessfully = 0 ; String queryExecClientName = "unknown" ; Exception [ ] threadExceptionArr = new Exception [ queryGenTable . getNumRows ( ) ] ; LocalLogger . logToStdErr ( "About to start edc work" ) ; int numQueries = queryGenTable . getNumRows ( ) ; DispatcherWorkThread [ ] spunup = new DispatcherWorkThread [ numQueries ] ; try { int threadLimit = MAX_NUMBER_SIMULTANEOUS_QUERIES_PER_USER ; int finished = 0 ; int iteration = 0 ; int running = 0 ; int currentReq = 0 ; while ( finished < numQueries ) { if ( running == numQueries ) { break ; } if ( running >= threadLimit || currentReq >= numQueries ) { for ( int joined = 0 ; joined < threadLimit ; joined += 1 ) { try { spunup [ joined + ( threadLimit * iteration ) ] . join ( ) ; finished += 1 ; running = running - 1 ; this . jobTracker . incrementPercentComplete ( this . jobID , Math . round ( increment ) , highestStatPercent ) ; if ( finished >= numQueries ) { break ; } } catch ( Exception e ) { LocalLogger . printStackTrace ( e ) ; throw new IOException ( "(Join failure in dispatch threading : joined value was " + joined + ":: " + e . getClass ( ) . toString ( ) + " : " + e . getMessage ( ) + ")" ) ; } } if ( currentReq % threadLimit == 0 ) { iteration += 1 ; } } else { if ( currentReq < numQueries ) { try { DispatchQueryGroup currDQG = this . queryGroupColl . getGroupByUUID ( UUID . fromString ( queryGenTable . getCell ( currentReq , Utility . COL_NAME_UUID ) ) ) ; JSONObject configJSON = null ; JSONParser jParser = new JSONParser ( ) ; configJSON = ( JSONObject ) jParser . parse ( queryGenTable . getCell ( currentReq , Utility . COL_NAME_CONFIGJSON ) ) ; String currentJobId = this . getJobId ( ) + "_edc_" + String . valueOf ( currentReq ) ; LocalLogger . logToStdErr ( "config for current qry:" ) ; LocalLogger . logToStdErr ( configJSON . toString ( ) + " jobId: " + currentJobId ) ; QueryExecuteClient clnt = this . dispatchServiceMgr . getExecuteClient ( configJSON , currentJobId ) ; ResultsClient rClient = new ResultsClient ( this . resConfig ) ; queryExecClientName = clnt . getClass ( ) . getSimpleName ( ) ; JobTracker tracker = new JobTracker ( this . jobTrackerSei ) ; String query = queryGenTable . getColumn ( Utility . COL_NAME_QUERY ) [ currentReq ] ; spunup [ currentReq ] = new DispatcherWorkThread ( currDQG , query , clnt , tracker , rClient , threadExceptionArr , currentReq ) ; LocalLogger . logToStdErr ( "Starting EDC thread " + currentReq + " of a total of " + numQueries ) ; spunup [ currentReq ] . start ( ) ; } catch ( Exception EEE ) { throw new Exception ( "spin up of query thread failed. reported:" + EEE . getMessage ( ) ) ; } running += 1 ; currentReq += 1 ; } } } for ( int joined = 0 ; joined < running ; joined += 1 ) { try { spunup [ joined + ( iteration * threadLimit ) ] . join ( ) ; this . jobTracker . incrementPercentComplete ( this . jobID , Math . round ( increment ) , highestStatPercent ) ; continue ; } catch ( Exception e ) { LocalLogger . printStackTrace ( e ) ; throw new Exception ( "(Join failure in dispatch threading : joined value was " + joined + ":: " + e . getClass ( ) . toString ( ) + " : " + e . getMessage ( ) + ")" ) ; } } } catch ( Exception e ) { throw new Exception ( "Dispatcher threading failure: " + e . getMessage ( ) ) ; } for ( Exception e : threadExceptionArr ) { if ( e == null ) { numberThreadsCompletedSuccessfully += 1 ; } else { String message = e . getMessage ( ) ; int MAX_MESSAGE_LENGTH = 500 ; if ( message != null && message . length ( ) > MAX_MESSAGE_LENGTH ) { message = message . substring ( 0 , MAX_MESSAGE_LENGTH ) + "..." ; } throw new Exception ( queryExecClientName + " could not retrieve data: " + message , e ) ; } } return numberThreadsCompletedSuccessfully ; }
execute the threads to be completed.
execute the threads to be completed .
// note (for updates to the status service, we have a range of 50 percentage units to play with: 20% to 70% // array of entries to determine success count.  // create the group of threads  // the total we will try. // the array of them.  // nothing left to do... just wait for them to finish.  // get the right DispatchQueryGroup // get the JSON config from the generator return statement // TESTING ONLY // clnt.getConfig().setServiceServer("localhost"); // update the counters // make sure it all really closed. // determine if any threads failed  // at least one thread failed - throw a new Exception based on the first failure // this message will get displayed in a dialog, so limit its length