[teiid-commits] teiid SVN: r1760 - in trunk/engine/src/main/java: org/teiid/dqp/internal/process and 1 other directory.

[teiid-commits at lists.jboss.org]

Author: shawkins
Date: 2010-01-20 15:22:54 -0500 (Wed, 20 Jan 2010)
New Revision: 1760

Modified:
   trunk/engine/src/main/java/com/metamatrix/query/processor/QueryProcessor.java
   trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
Log:
TEIID-936 fixing the timeslicing mechanism to prevent query hangs

Modified: trunk/engine/src/main/java/com/metamatrix/query/processor/QueryProcessor.java
===================================================================
--- trunk/engine/src/main/java/com/metamatrix/query/processor/QueryProcessor.java	2010-01-20 18:29:36 UTC (rev 1759)
+++ trunk/engine/src/main/java/com/metamatrix/query/processor/QueryProcessor.java	2010-01-20 20:22:54 UTC (rev 1760)
@@ -32,6 +32,7 @@
 import com.metamatrix.common.buffer.TupleBatch;
 import com.metamatrix.common.buffer.BufferManager.TupleSourceType;
 import com.metamatrix.common.log.LogManager;
+import com.metamatrix.core.MetaMatrixRuntimeException;
 import com.metamatrix.core.log.MessageLevel;
 import com.metamatrix.core.util.Assertion;
 import com.metamatrix.dqp.util.LogConstants;
@@ -41,6 +42,12 @@
 
 public class QueryProcessor implements BatchProducer {
 	
+	public static class ExpiredTimeSliceException extends MetaMatrixRuntimeException {
+		
+	}
+	
+	private static ExpiredTimeSliceException EXPIRED_TIME_SLICE = new ExpiredTimeSliceException();
+	
 	public interface ProcessorFactory {
 		QueryProcessor createQueryProcessor(String query, String recursionGroup, CommandContext commandContext) throws MetaMatrixProcessingException, MetaMatrixComponentException;
 	}
@@ -92,16 +99,19 @@
 	    while (true) {
 	    	try {
 	    		return nextBatchDirect();
+	    	} catch (ExpiredTimeSliceException e) {
+	    		if (!nonBlocking) {
+	    			throw e;
+	    		}
 	    	} catch (BlockedException e) {
-	    		if (nonBlocking) {
-	    			try {
-	                    Thread.sleep(DEFAULT_WAIT);
-	                } catch (InterruptedException err) {
-	                    throw new MetaMatrixComponentException(err);
-	                }
-	    			continue;
+	    		if (!nonBlocking) {
+	    			throw e;
 	    		}
-	    		throw e;
+	    		try {
+	                Thread.sleep(DEFAULT_WAIT);
+	            } catch (InterruptedException err) {
+	                throw new MetaMatrixComponentException(err);
+	            }
 	    	}
 	    }
 	}
@@ -162,7 +172,7 @@
 			closeProcessing();
 		} 
 	    if (result == null) {
-	    	throw BlockedException.INSTANCE; //expired timeslice
+	    	throw EXPIRED_TIME_SLICE;
 	    }
 		return result;
 	}

Modified: trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java
===================================================================
--- trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java	2010-01-20 18:29:36 UTC (rev 1759)
+++ trunk/engine/src/main/java/org/teiid/dqp/internal/process/RequestWorkItem.java	2010-01-20 20:22:54 UTC (rev 1760)
@@ -187,6 +187,9 @@
             }                  	            
         } catch (BlockedException e) {
             LogManager.logDetail(LogConstants.CTX_DQP, "############# PW EXITING on", requestID, "- processor blocked ###########"); //$NON-NLS-1$ //$NON-NLS-2$
+        } catch (QueryProcessor.ExpiredTimeSliceException e) {
+            LogManager.logDetail(LogConstants.CTX_DQP, "############# PW reenqueueing ", requestID, "- time slice expired ###########"); //$NON-NLS-1$ //$NON-NLS-2$
+            this.moreWork();
         } catch (Throwable e) {
         	LogManager.logDetail(LogConstants.CTX_DQP, e, "############# PW EXITING on", requestID, "- error occurred ###########"); //$NON-NLS-1$ //$NON-NLS-2$
             
@@ -490,13 +493,10 @@
         resultsReceiver.receiveResults(response);
     }
 
-    private static List getParameterInfo(StoredProcedure procedure) {
-        List params = procedure.getParameters();
-        List paramInfos = new ArrayList(params.size());
+    private static List<ParameterInfo> getParameterInfo(StoredProcedure procedure) {
+        List<ParameterInfo> paramInfos = new ArrayList<ParameterInfo>();
         
-        Iterator iter = params.iterator();
-        while(iter.hasNext()) {
-            SPParameter param = (SPParameter) iter.next();
+        for (SPParameter param : procedure.getParameters()) {
             ParameterInfo info = new ParameterInfo(param.getParameterType(), param.getResultSetColumns().size());
             paramInfos.add(info);
         }