Use connection pooling to avoid using up all connections during massi…

…ve ingestions with URILookup

sparqlGraphLibrary/src/main/java/com/ge/research/semtk/sparqlX/BlazegraphSparqlEndpointInterface.java

@@ -25,6 +25,7 @@
 import org.apache.http.HttpHost;
 import org.apache.http.HttpResponse;
 import org.apache.http.NameValuePair;
+import org.apache.http.client.HttpClient;
 import org.apache.http.client.entity.UrlEncodedFormEntity;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.ByteArrayEntity;
@@ -154,7 +155,7 @@ public JSONObject executeUpload(byte[] owl, String filename) throws Authorizatio
 		this.authorizeUpload();
 
 		HttpHost targetHost = this.buildHttpHost();
-        CloseableHttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
+        HttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
 		BasicHttpContext localcontext = this.buildHttpContext(targetHost);
 		HttpPost httppost = new HttpPost(this.getUploadURL());
 

sparqlGraphLibrary/src/main/java/com/ge/research/semtk/sparqlX/FusekiSparqlEndpointInterface.java

@@ -25,6 +25,7 @@
 import org.apache.http.HttpHost;
 import org.apache.http.HttpResponse;
 import org.apache.http.NameValuePair;
+import org.apache.http.client.HttpClient;
 import org.apache.http.client.entity.UrlEncodedFormEntity;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.mime.MultipartEntityBuilder;
@@ -150,7 +151,7 @@ public JSONObject executeUpload(byte[] owl, String filename) throws Authorizatio
 		this.authorizeUpload();
 
 		HttpHost targetHost = this.buildHttpHost();
-        CloseableHttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
+        HttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
 		BasicHttpContext localcontext = this.buildHttpContext(targetHost);
 		HttpPost httppost = new HttpPost(this.getUploadURL());
 

sparqlGraphLibrary/src/main/java/com/ge/research/semtk/sparqlX/SparqlEndpointInterface.java

@@ -49,15 +49,17 @@
 import org.apache.http.auth.UsernamePasswordCredentials;
 import org.apache.http.client.AuthCache;
 import org.apache.http.client.CredentialsProvider;
+import org.apache.http.client.HttpClient;
 import org.apache.http.client.entity.UrlEncodedFormEntity;
 import org.apache.http.client.methods.HttpGet;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.client.protocol.HttpClientContext;
+import org.apache.http.conn.HttpClientConnectionManager;
 import org.apache.http.impl.auth.DigestScheme;
 import org.apache.http.impl.client.BasicAuthCache;
 import org.apache.http.impl.client.BasicCredentialsProvider;
-import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.apache.http.message.BasicNameValuePair;
 import org.apache.http.protocol.BasicHttpContext;
 import org.apache.http.util.EntityUtils;
@@ -67,16 +69,12 @@
 
 import com.ge.research.semtk.auth.AuthorizationException;
 import com.ge.research.semtk.auth.AuthorizationManager;
-import com.ge.research.semtk.belmont.ValueConstraint;
-import com.ge.research.semtk.belmont.XSDSupportedType;
 import com.ge.research.semtk.ontologyTools.OntologyInfo;
 import com.ge.research.semtk.resultSet.GeneralResultSet;
 import com.ge.research.semtk.resultSet.NodeGroupResultSet;
 import com.ge.research.semtk.resultSet.SimpleResultSet;
 import com.ge.research.semtk.resultSet.Table;
 import com.ge.research.semtk.resultSet.TableResultSet;
-import com.ge.research.semtk.sparqlX.FusekiSparqlEndpointInterface;
-import com.ge.research.semtk.sparqlX.VirtuosoSparqlEndpointInterface;
 import com.ge.research.semtk.utility.LocalLogger;
 import com.ge.research.semtk.utility.Utility;
 
@@ -134,6 +132,14 @@ public abstract class SparqlEndpointInterface {
 
 	protected int retries = 0;
 
+	private static HttpClientConnectionManager manager = buildConnectionManager();
+
+	private static HttpClientConnectionManager buildConnectionManager() {
+		PoolingHttpClientConnectionManager ret =  new PoolingHttpClientConnectionManager();
+		ret.setDefaultMaxPerRoute(10);
+		return ret;
+	}
+
 	/**
 	 * Constructor
 	 * @param serverAndPort e.g. "http://localhost:2420"
@@ -708,7 +714,7 @@ public JSONObject executeQueryPost(String query, SparqlResultTypes resultType) t
 
         // get client, adding userName/password credentials if any exist
 		HttpHost targetHost = this.buildHttpHost();
-        CloseableHttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
+        HttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
 		// get context with digest auth if there is a userName, else null context
 		BasicHttpContext localcontext = this.buildHttpContext(targetHost);
 
@@ -742,9 +748,8 @@ public JSONObject executeQueryPost(String query, SparqlResultTypes resultType) t
 			// parse response
 			return this.parseResponse(resultType, responseTxt);
 		} finally {
-			httpclient.close();
 			if (entity != null) {
-				entity.getContent().close();
+				EntityUtils.consume(entity);
 			}
 		}
 	}
@@ -828,13 +833,14 @@ public void dropGraph() throws Exception {
 	}
 
 	/**
-	 * Get an CloseableHttpClient, handling credentials and HTTPS if needed
+	 * Get an HttpClient, handling credentials and HTTPS if needed
 	 * NOTE: for HTTPS connections, does not validate certificate chain.
 	 * @schemeName http or https
 	 */
-	protected CloseableHttpClient buildHttpClient(String schemeName) throws Exception {
+	protected HttpClient buildHttpClient(String schemeName) throws Exception {
 
 		HttpClientBuilder clientBuilder = HttpClientBuilder.create();
+		clientBuilder.setConnectionManager(SparqlEndpointInterface.manager);
 
 		// add userName and password, if any
 		if (this.isAuth()) {
@@ -1098,7 +1104,7 @@ private JSONObject executeQueryAuthGet(String queryAndUrl, SparqlResultTypes res
 		}
 
 		HttpHost targetHost = this.buildHttpHost();
-		CloseableHttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
+		HttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
 		BasicHttpContext localcontext = this.buildHttpContext(targetHost);
 
 		LocalLogger.logToStdErr(queryAndUrl);
@@ -1118,8 +1124,7 @@ private JSONObject executeQueryAuthGet(String queryAndUrl, SparqlResultTypes res
 		try {
 			return this.parseResponse(resultType, responseTxt);
 		} finally {
-			httpclient.close();
-			entity.getContent().close();
+			EntityUtils.consume(entity);
 		}
 	}
 

sparqlGraphLibrary/src/main/java/com/ge/research/semtk/sparqlX/VirtuosoSparqlEndpointInterface.java

@@ -23,16 +23,14 @@
 import org.apache.http.HttpEntity;
 import org.apache.http.HttpHost;
 import org.apache.http.HttpResponse;
+import org.apache.http.client.HttpClient;
 import org.apache.http.client.methods.HttpPost;
 import org.apache.http.entity.mime.MultipartEntityBuilder;
-import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.protocol.BasicHttpContext;
 import org.apache.http.util.EntityUtils;
 import org.json.simple.JSONObject;
 import com.ge.research.semtk.resultSet.SimpleResultSet;
 import com.ge.research.semtk.auth.AuthorizationException;
-import com.ge.research.semtk.auth.AuthorizationManager;
-import com.ge.research.semtk.sparqlX.SparqlEndpointInterface;
 
 /**
  * Interface to Virtuoso SPARQL endpoint
@@ -144,7 +142,7 @@ public JSONObject executeUpload(byte[] owl) throws AuthorizationException, Excep
 		this.authorizeUpload();
 
 		HttpHost targetHost = this.buildHttpHost();
-        CloseableHttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
+        HttpClient httpclient = this.buildHttpClient(targetHost.getSchemeName());
 		BasicHttpContext localcontext = this.buildHttpContext(targetHost);
 		HttpPost httppost = new HttpPost(this.getUploadURL());
 

standaloneExecutables/src/main/java/com/ge/research/semtk/standaloneExecutables/PerformanceTest.java

@@ -63,7 +63,8 @@ public static void main(String[] args) throws Exception {
 		try {
 
 			//addSimpleRows(10, 10000); 
-			addBatteryDescriptions(1000, 500000);  
+			//addBatteryDescriptions(1000, 500000);  
+			addBatteryDescriptions(40000, 75000);  
 			//addBatteryDescriptionsVaryingThreadsAndSize(0);
 			//addSimpleBiggerRows(10, 50000);
 
@@ -234,7 +235,7 @@ private static void addBatteryDescriptions(int rows_per_pass, int max_triples) t
 				Dataset ds1 = new CSVDataset(content1.toString(), true);
 				DataLoader dl1 = new DataLoader(sgJson1, ds1, "dba", "dba");
 				startTask("addBatteryDescriptions load simple, rows, " + rows_per_pass + ",total rows," + total_rows);
-				dl1.importData(NO_PRECHECK);
+				dl1.importData(PRECHECK);
 				lastSec[0] = endTask();
 			}
 
@@ -244,7 +245,7 @@ private static void addBatteryDescriptions(int rows_per_pass, int max_triples) t
 				DataLoader dl2 = new DataLoader(sgJson2, ds2, "dba", "dba");
 				dl2.setLogPerformance(LOG_QUERY_PERFORMANCE);
 				startTask("addBatteryDescriptions load lookup class, rows," + rows_per_pass/2 + ",total rows," + total_rows);
-				dl2.importData(NO_PRECHECK);
+				dl2.importData(PRECHECK);
 				lastSec[1] = endTask();
 			}
 
@@ -254,7 +255,7 @@ private static void addBatteryDescriptions(int rows_per_pass, int max_triples) t
 				DataLoader dl3 = new DataLoader(sgJson3, ds3, "dba", "dba");
 				dl3.setLogPerformance(LOG_QUERY_PERFORMANCE);
 				startTask("addBatteryDescriptions load lookup superclass, rows," + rows_per_pass /2 + ",total rows," + total_rows);
-				dl3.importData(NO_PRECHECK);
+				dl3.importData(PRECHECK);
 				lastSec[2] = endTask();
 			}