TEIID-2322 switching reusable executions to a pool paradigm rather than

relying on partid

Author: shawkins

engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorManager.java

@@ -33,10 +33,10 @@
 import org.teiid.dqp.message.AtomicRequestID;
 import org.teiid.dqp.message.AtomicRequestMessage;
 import org.teiid.logging.CommandLogMessage;
+import org.teiid.logging.CommandLogMessage.Event;
 import org.teiid.logging.LogConstants;
 import org.teiid.logging.LogManager;
 import org.teiid.logging.MessageLevel;
-import org.teiid.logging.CommandLogMessage.Event;
 import org.teiid.metadata.FunctionMethod;
 import org.teiid.query.QueryPlugin;
 import org.teiid.query.optimizer.capabilities.BasicSourceCapabilities;
@@ -57,6 +57,7 @@ public class ConnectorManager  {
 
 	private String translatorName;
 	private String connectionName;
+	private List<String> id;
 
     // known requests
     private ConcurrentHashMap<AtomicRequestID, ConnectorWorkItem> requestStates = new ConcurrentHashMap<AtomicRequestID, ConnectorWorkItem>();
@@ -69,6 +70,7 @@ public class ConnectorManager  {
     public ConnectorManager(String translatorName, String connectionName) {
     	this.translatorName = translatorName;
     	this.connectionName = connectionName;
+    	this.id = Arrays.asList(translatorName, connectionName);
     }
 
     public String getStausMessage() {
@@ -107,7 +109,7 @@ public SourceCapabilities getCapabilities() throws TeiidComponentException {
 
 		checkStatus();
 		ExecutionFactory<Object, Object> translator = getExecutionFactory();
-		BasicSourceCapabilities resultCaps = CapabilitiesConverter.convertCapabilities(translator, Arrays.asList(translatorName, connectionName));
+		BasicSourceCapabilities resultCaps = CapabilitiesConverter.convertCapabilities(translator, id);
 		cachedCapabilities = resultCaps;
 		return resultCaps;
     }
@@ -245,4 +247,8 @@ public String getConnectionName() {
     	return this.connectionName;
     }
 
+    public List<String> getId() {
+		return id;
+	}
+    
 }

engine/src/main/java/org/teiid/dqp/internal/datamgr/ConnectorWorkItem.java

@@ -156,7 +156,10 @@ public void close() {
         try {
 	        if (execution != null) {
 	            execution.close();
-	            LogManager.logDetail(LogConstants.CTX_CONNECTOR, new Object[] {this.id, "Closed execution"}); //$NON-NLS-1$                    
+	            LogManager.logDetail(LogConstants.CTX_CONNECTOR, new Object[] {this.id, "Closed execution"}); //$NON-NLS-1$
+	            if (execution instanceof ReusableExecution<?>) {
+		        	this.requestMsg.getCommandContext().putReusableExecution(this.manager.getId(), (ReusableExecution<?>) execution);
+		        }
 	        }	        
         } catch (Throwable e) {
             LogManager.logError(LogConstants.CTX_CONNECTOR, e, e.getMessage());
@@ -238,14 +241,11 @@ public void execute() throws TranslatorException {
 		        	this.expectedColumns = ((StoredProcedure)command).getResultSetColumns().size();
 		        }
 
-				Execution exec = this.requestMsg.getCommandContext().getReusableExecution(this.securityContext.getPartIdentifier());
+				Execution exec = this.requestMsg.getCommandContext().getReusableExecution(this.manager.getId());
 				if (exec != null) {
 					((ReusableExecution)exec).reset(translatedCommand, this.securityContext, connection);
 				} else {
 			        exec = connector.createExecution(translatedCommand, this.securityContext, queryMetadata, (unwrapped == null) ? this.connection:unwrapped);
-			        if (exec instanceof ReusableExecution<?>) {
-			        	this.requestMsg.getCommandContext().putReusableExecution(this.securityContext.getPartIdentifier(), (ReusableExecution<?>) exec);
-			        }
 				}
 		        setExecution(command, translatedCommand, exec);
 

engine/src/main/java/org/teiid/query/util/CommandContext.java

@@ -26,7 +26,6 @@
 import java.text.DecimalFormat;
 import java.text.SimpleDateFormat;
 import java.util.*;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.Executor;
 import java.util.concurrent.atomic.AtomicLong;
 
@@ -136,7 +135,7 @@ private static class GlobalState {
 		private TransactionService transactionService;
 		private SourceHint sourceHint;
 		private Executor executor = ExecutorUtils.getDirectExecutor();
-		Map<String, ReusableExecution<?>> reusableExecutions;
+		Map<Object, List<ReusableExecution<?>>> reusableExecutions;
 	    Set<CommandListener> commandListeners = null;
 	    private LRUCache<String, DecimalFormat> decimalFormatCache;
 		private LRUCache<String, SimpleDateFormat> dateFormatCache;
@@ -647,32 +646,43 @@ public void setExecutor(Executor e) {
 		this.globalState.executor = e;
 	}
 
-	public ReusableExecution<?> getReusableExecution(String nodeId) {
+	public ReusableExecution<?> getReusableExecution(Object key) {
 		synchronized (this.globalState) {
 			if (this.globalState.reusableExecutions == null) {
 				return null;
 			}
-			return this.globalState.reusableExecutions.get(nodeId);
+			List<ReusableExecution<?>> reusableExecutions = this.globalState.reusableExecutions.get(key);
+			if (reusableExecutions != null && !reusableExecutions.isEmpty()) {
+				return reusableExecutions.remove(0);
+			}
+			return null;
 		}
 	}
 
-	public void putReusableExecution(String nodeId, ReusableExecution<?> execution) {
+	public void putReusableExecution(Object key, ReusableExecution<?> execution) {
 		synchronized (this.globalState) {
 			if (this.globalState.reusableExecutions == null) {
-				this.globalState.reusableExecutions = new ConcurrentHashMap<String, ReusableExecution<?>>();
+				this.globalState.reusableExecutions = new HashMap<Object, List<ReusableExecution<?>>>();
+			}
+			List<ReusableExecution<?>> reusableExecutions = this.globalState.reusableExecutions.get(key);
+			if (reusableExecutions == null) {
+				reusableExecutions = new LinkedList<ReusableExecution<?>>();
+				this.globalState.reusableExecutions.put(key, reusableExecutions);
 			}
-			this.globalState.reusableExecutions.put(nodeId, execution);
+			reusableExecutions.add(execution);
 		}
 	}
 
 	public void close() {
 		synchronized (this.globalState) {
 			if (this.globalState.reusableExecutions != null) {
-				for (ReusableExecution<?> reusableExecution : this.globalState.reusableExecutions.values()) {
-					try {
-						reusableExecution.dispose();
-					} catch (Exception e) {
-						LogManager.logWarning(LogConstants.CTX_DQP, e, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30030));
+				for (List<ReusableExecution<?>> reusableExecutions : this.globalState.reusableExecutions.values()) {
+					for (ReusableExecution<?> reusableExecution : reusableExecutions) {
+						try {
+							reusableExecution.dispose();
+						} catch (Exception e) {
+							LogManager.logWarning(LogConstants.CTX_DQP, e, QueryPlugin.Util.gs(QueryPlugin.Event.TEIID30030));
+						}
 					}
 				}
 				this.globalState.reusableExecutions.clear();

test-integration/common/src/test/java/org/teiid/dqp/internal/process/TestExecutionReuse.java

@@ -21,8 +21,7 @@
  */
 package org.teiid.dqp.internal.process;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.*;
 
 import java.sql.Connection;
 import java.sql.ResultSet;
@@ -50,6 +49,7 @@
 import org.teiid.language.Command;
 import org.teiid.language.QueryExpression;
 import org.teiid.metadata.RuntimeMetadata;
+import org.teiid.query.util.CommandContext;
 import org.teiid.runtime.EmbeddedConfiguration;
 import org.teiid.translator.DataNotAvailableException;
 import org.teiid.translator.ExecutionContext;
@@ -203,5 +203,18 @@ public void onComplete(Statement stmt) {
 		Mockito.verify(execution, Mockito.times(EXEC_COUNT)).close();
 		Mockito.verify(execution, Mockito.times(EXEC_COUNT - 1)).reset((Command)Mockito.anyObject(), (ExecutionContext)Mockito.anyObject(), Mockito.anyObject());
 	}
+	
+	@Test public void testCommandContext() {
+		CommandContext cc = new CommandContext();
+		FakeReusableExecution fe = new FakeReusableExecution();
+		cc.putReusableExecution("a", fe);
+		cc.putReusableExecution("a", new FakeReusableExecution());
+		
+		ReusableExecution<?> re = cc.getReusableExecution("a");
+		ReusableExecution<?> re1 = cc.getReusableExecution("a");
+		assertSame(fe, re);
+		assertNotSame(fe, re1);
+		assertNull(cc.getReusableExecution("a"));
+	}
 
 }