Add support for ahead-of-time compilation

graalvm/generate-config.sh

@@ -43,9 +43,9 @@ if [ -f "$TARGET_DIR"/native/agent-output/test/resource-config.json ]; then
 fi
 
 # Run through multiple different execution paths, so that the tracing agent can generate complete configuration files.
-"$GRAALVM_HOME"/bin/java -agentlib:native-image-agent=config-merge-dir=src/main/resources/META-INF/native-image/ -jar "$TARGET_DIR"/iguana.jar --help      > /dev/null
-"$GRAALVM_HOME"/bin/java -agentlib:native-image-agent=config-merge-dir=src/main/resources/META-INF/native-image/ -jar "$TARGET_DIR"/iguana.jar -is "$SUITE" > /dev/null
-"$GRAALVM_HOME"/bin/java -agentlib:native-image-agent=config-merge-dir=src/main/resources/META-INF/native-image/ -jar "$TARGET_DIR"/iguana.jar "$SUITE"     > /dev/null
+"$GRAALVM_HOME"/bin/java -agentlib:native-image-agent=config-merge-dir=src/main/resources/META-INF/native-image/ -jar "$TARGET_DIR"/iguana.jar --help             > /dev/null
+"$GRAALVM_HOME"/bin/java -agentlib:native-image-agent=config-merge-dir=src/main/resources/META-INF/native-image/ -jar "$TARGET_DIR"/iguana.jar --dry-run -is "$SUITE" > /dev/null
+"$GRAALVM_HOME"/bin/java -agentlib:native-image-agent=config-merge-dir=src/main/resources/META-INF/native-image/ -jar "$TARGET_DIR"/iguana.jar --dry-run "$SUITE"     > /dev/null
 
 # there is a bug in the tracing agent that outputs wrong formatted lines in the resource-config.json file (https://github.com/oracle/graal/issues/7985)
 sed 's/\\\\E//g' src/main/resources/META-INF/native-image/resource-config.json | sed 's/\\\\Q//g' > src/main/resources/META-INF/native-image/resource-config.json.tmp

pom.xml

@@ -313,6 +313,7 @@
                                 -march=x86-64-v3
                                 --no-fallback
                                 -O3
+                                -H:-UseCompressedReferences
                             </buildArgs>
                             <metadataRepository>
                                 <enabled>true</enabled>

src/main/java/org/aksw/iguana/cc/controller/MainController.java

@@ -28,14 +28,16 @@ public Path convert(String value) {
         @Parameter(names = {"--ignore-schema", "-is"}, description = "Do not check the schema before parsing the suite file.")
         private boolean ignoreShema = false;
 
+        @Parameter(names = {"--dry-run", "-d"}, hidden = true)
+        public static boolean dryRun = false;
+
         @Parameter(names = "--help", help = true)
         private boolean help;
 
         @Parameter(description = "suite file {yml,yaml,json}", arity = 1, required = true, converter = PathConverter.class)
         private Path suitePath;
     }
 
-
     private static final Logger LOGGER = LoggerFactory.getLogger(MainController.class);
 
     /**

src/main/java/org/aksw/iguana/cc/storage/impl/TriplestoreStorage.java

@@ -2,6 +2,7 @@
 
 import com.fasterxml.jackson.annotation.JsonProperty;
 import org.aksw.iguana.cc.config.elements.StorageConfig;
+import org.aksw.iguana.cc.controller.MainController;
 import org.aksw.iguana.cc.storage.Storage;
 import org.apache.http.auth.AuthScope;
 import org.apache.http.auth.Credentials;
@@ -17,6 +18,7 @@
 import org.apache.jena.update.UpdateFactory;
 import org.apache.jena.update.UpdateProcessor;
 import org.apache.jena.update.UpdateRequest;
+import org.mortbay.jetty.Main;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -79,10 +81,14 @@ public void storeResult(Model data) {
 		//submit Block to Triple Store
 		UpdateProcessor processor = UpdateExecutionFactory
 				.createRemote(blockRequest, endpoint, createHttpClient());
-		try {
+		if (MainController.Args.dryRun) {
+			try {
+				processor.execute();
+			} catch (Exception e) {
+				logger.error("Error while storing data in triplestore: " + e.getMessage());
+			}
+		} else {
 			processor.execute();
-		} catch (Exception e) {
-			logger.error("Error while storing data in triplestore: " + e.getMessage());
 		}
 		blockRequest = new UpdateRequest();
 	}

src/main/java/org/aksw/iguana/cc/worker/ResponseBodyProcessor.java

@@ -1,11 +1,10 @@
 package org.aksw.iguana.cc.worker;
 
 import org.aksw.iguana.cc.lang.LanguageProcessor;
-import org.aksw.iguana.commons.io.BigByteArrayInputStream;
-import org.aksw.iguana.commons.io.BigByteArrayOutputStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.InputStream;
 import java.text.MessageFormat;
 import java.time.Duration;
 import java.util.ArrayList;
@@ -46,18 +45,18 @@ public ResponseBodyProcessor(String contentType) {
     private final ThreadPoolExecutor executor;
     private final ScheduledExecutorService executorHandler = Executors.newScheduledThreadPool(1);
 
-    public boolean add(long contentLength, long xxh64, BigByteArrayOutputStream bbaos) {
+    public boolean add(long contentLength, long xxh64, InputStream responseBodyStream) {
         final var key = new Key(contentLength, xxh64);
         if (seenResponseBodies.add(key)) {
-            submit(key, bbaos);
+            submit(key, responseBodyStream);
             return true;
         }
         return false;
     }
 
-    private void submit(Key key, BigByteArrayOutputStream bigByteArrayOutputStream) {
+    private void submit(Key key, InputStream responseBodyStream) {
         final var future = executor.submit(() -> {
-            var processingResult = languageProcessor.process(new BigByteArrayInputStream(bigByteArrayOutputStream), key.xxh64);
+            var processingResult = languageProcessor.process(responseBodyStream, key.xxh64);
             responseDataMetrics.add(processingResult);
         });
         executorHandler.schedule(() -> {

src/main/java/org/aksw/iguana/cc/worker/impl/SPARQLProtocolWorker.java

@@ -10,10 +10,11 @@
 import org.aksw.iguana.cc.worker.ResponseBodyProcessor;
 import org.aksw.iguana.cc.worker.HttpWorker;
 import org.aksw.iguana.commons.io.BigByteArrayOutputStream;
+import org.aksw.iguana.commons.io.ByteArrayListOutputStream;
+import org.aksw.iguana.commons.io.ReversibleOutputStream;
 import org.apache.hc.client5.http.async.methods.AbstractBinResponseConsumer;
 import org.apache.hc.client5.http.config.RequestConfig;
 import org.apache.hc.client5.http.impl.DefaultConnectionKeepAliveStrategy;
-import org.apache.hc.client5.http.impl.DefaultHttpRequestRetryStrategy;
 import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
 import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
 import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
@@ -22,11 +23,7 @@
 import org.apache.hc.core5.http.HttpException;
 import org.apache.hc.core5.http.HttpResponse;
 import org.apache.hc.core5.http.nio.AsyncRequestProducer;
-import org.apache.hc.core5.pool.PoolConcurrencyPolicy;
-import org.apache.hc.core5.pool.PoolReusePolicy;
 import org.apache.hc.core5.reactor.IOReactorConfig;
-import org.apache.hc.core5.util.TimeValue;
-import org.apache.hc.core5.util.Timeout;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.helpers.MessageFormatter;
@@ -77,7 +74,7 @@ record HttpExecutionResult(
             Optional<HttpResponse> response,
             Instant requestStart,
             Duration duration,
-            Optional<BigByteArrayOutputStream> outputStream,
+            Optional<ReversibleOutputStream> outputStream,
             OptionalLong actualContentLength,
             OptionalLong hash,
             Optional<Exception> exception
@@ -131,28 +128,19 @@ public static void initHttpClient(int threadCount) {
         connectionManager = PoolingAsyncClientConnectionManagerBuilder.create()
                 .setMaxConnTotal(threadCount * 1000)
                 .setMaxConnPerRoute(threadCount * 1000)
-                .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.LAX)
-                .setConnPoolPolicy(PoolReusePolicy.FIFO)
-                .setDefaultConnectionConfig(org.apache.hc.client5.http.config.ConnectionConfig.custom()
-                        .setConnectTimeout(Timeout.ofSeconds(5))
-                        .build())
                 .build();
         final var ioReactorConfig = IOReactorConfig.custom()
                 .setTcpNoDelay(true)
-                .setIoThreadCount(threadCount + 1)
-                .setSelectInterval(TimeValue.ofMilliseconds(100))
                 .setSoKeepAlive(true)
                 .build();
         httpClient = HttpAsyncClients.custom()
                 .setConnectionManager(connectionManager)
                 .setIOReactorConfig(ioReactorConfig)
                 .setKeepAliveStrategy(new DefaultConnectionKeepAliveStrategy())
-                .setConnectionManagerShared(false)
                 .setDefaultRequestConfig(RequestConfig.custom()
                         .setContentCompressionEnabled(false)
                         .setHardCancellationEnabled(true)
                         .build())
-                .setRetryStrategy(DefaultHttpRequestRetryStrategy.INSTANCE)
                 .build();
         httpClient.start();
     }
@@ -264,7 +252,7 @@ private ExecutionStats executeQuery(Duration timeout, boolean discardOnFailure)
             }
 
             // process result
-            responseBodyProcessor.add(result.actualContentLength().orElse(-1), result.hash().orElse(-1), result.outputStream().orElse(new BigByteArrayOutputStream()));
+            responseBodyProcessor.add(result.actualContentLength().getAsLong(), result.hash().getAsLong(), result.outputStream.get().toInputStream());
         }
 
         if (!result.successful() && discardOnFailure) {
@@ -328,7 +316,7 @@ private HttpExecutionResult executeHttpRequest(Duration timeout) {
             private final StreamingXXHash64 hasher = hasherFactory.newStreamingHash64(0);
             private long responseSize = 0; // will be used if parseResults is false
             private long responseEnd = 0;  // time in nanos
-            private BigByteArrayOutputStream responseBodybbaos = null;
+            private ReversibleOutputStream responseBody = null;
 
             @Override
             public void releaseResources() {} // nothing to release
@@ -349,22 +337,22 @@ protected void data(ByteBuffer src, boolean endOfStream) throws IOException {
                 if (endOfStream)
                     responseEnd = System.nanoTime();
 
-                if (responseBodybbaos == null)
-                    responseBodybbaos = new BigByteArrayOutputStream();
+                if (responseBody == null)
+                    responseBody = new ByteArrayListOutputStream();
 
                 responseSize += src.remaining();
                 if (config.parseResults()) {
                     // if the buffer uses an array, use the array directly
                     if (src.hasArray()) {
                         hasher.update(src.array(), src.position() + src.arrayOffset(), src.remaining());
-                        responseBodybbaos.write(src.array(), src.position() + src.arrayOffset(), src.remaining());
+                        responseBody.write(src.array(), src.position() + src.arrayOffset(), src.remaining());
                     } else { // otherwise, copy the buffer to an array
                         int readCount;
                         while (src.hasRemaining()) {
                             readCount = Math.min(BUFFER_SIZE, src.remaining());
                             src.get(buffer, 0, readCount);
                             hasher.update(buffer, 0, readCount);
-                            responseBodybbaos.write(buffer, 0, readCount);
+                            responseBody.write(buffer, 0, readCount);
                         }
                     }
                 }
@@ -383,8 +371,8 @@ protected void start(HttpResponse response, ContentType contentType) {
                 final var contentLengthHeader = response.getFirstHeader("Content-Length");
                 Long contentLength = contentLengthHeader != null ? Long.parseLong(contentLengthHeader.getValue()) : null;
                 // if the content length is known, create a BigByteArrayOutputStream with the known length
-                if (contentLength != null && responseBodybbaos == null && config.parseResults()) {
-                    responseBodybbaos = new BigByteArrayOutputStream(contentLength);
+                if (contentLength != null && responseBody == null && config.parseResults()) {
+                    responseBody = new BigByteArrayOutputStream(contentLength);
                 }
             }
 
@@ -412,10 +400,10 @@ protected HttpExecutionResult buildResult() {
                 Long contentLength = contentLengthHeader != null ? Long.parseLong(contentLengthHeader.getValue()) : null;
                 if (contentLength != null) {
                     if ((!config.parseResults() && responseSize != contentLength) // if parseResults is false, the responseSize will be used
-                            || (config.parseResults() && responseBodybbaos.size() != contentLength)) { // if parseResults is true, the size of the bbaos will be used
-                        if (responseSize != responseBodybbaos.size())
-                            LOGGER.error("Error during copying the response data. (expected written data size = {}, actual written data size = {}, Content-Length-Header = {})", responseSize, responseBodybbaos.size(), contentLengthHeader.getValue());
-                        final var exception = new HttpException(String.format("Content-Length header value doesn't match actual content length. (Content-Length-Header = %s, written data size = %s)", contentLength, config.parseResults() ? responseBodybbaos.size() : responseSize));
+                            || (config.parseResults() && responseBody.size() != contentLength)) { // if parseResults is true, the size of the bbaos will be used
+                        if (responseSize != responseBody.size())
+                            LOGGER.error("Error during copying the response data. (expected written data size = {}, actual written data size = {}, Content-Length-Header = {})", responseSize, responseBody.size(), contentLengthHeader.getValue());
+                        final var exception = new HttpException(String.format("Content-Length header value doesn't match actual content length. (Content-Length-Header = %s, written data size = %s)", contentLength, config.parseResults() ? responseBody.size() : responseSize));
                         return createFailedResultDuringResponse(queryIndex, response, timeStamp, duration, exception);
                     }
                 }
@@ -431,8 +419,8 @@ protected HttpExecutionResult buildResult() {
                         Optional.of(response),
                         timeStamp,
                         Duration.ofNanos(responseEnd - requestStart),
-                        Optional.of(responseBodybbaos),
-                        OptionalLong.of(config.parseResults() ? responseBodybbaos.size() : responseSize),
+                        Optional.of(responseBody),
+                        OptionalLong.of(config.parseResults() ? responseBody.size() : responseSize),
                         OptionalLong.of(config.parseResults() ? hasher.getValue() : 0),
                         Optional.empty()
                 );
@@ -451,7 +439,7 @@ protected HttpExecutionResult buildResult() {
             return createFailedResultBeforeRequest(queryIndex, e);
         } catch (TimeoutException e) {
             if (future.isDone()) {
-                LOGGER.warn("Request was already done after timeout.");
+                LOGGER.warn("Request finished immediately after timeout but will still be counted as timed out.");
                 try {
                     return future.get();
                 } catch (InterruptedException | ExecutionException ex) {

src/main/java/org/aksw/iguana/commons/io/BigByteArrayOutputStream.java

@@ -3,7 +3,7 @@
 import org.apache.hadoop.hbase.io.ByteArrayOutputStream;
 
 import java.io.IOException;
-import java.io.OutputStream;
+import java.io.InputStream;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
@@ -22,7 +22,7 @@
  * the stream is cleared, all the internal ByteArrayOutputStreams are cleared and a new one is
  * added to the list.
  */
-public class BigByteArrayOutputStream extends OutputStream {
+public class BigByteArrayOutputStream extends ReversibleOutputStream {
 
     /**
      * The maximum size limit for an array. This is no limit to the amount of bytes {@code BigByteArrayOutputStream} can consume.
@@ -102,6 +102,7 @@ public void write(BigByteArrayOutputStream bbaos) throws IOException {
         write(bbaos.toByteArray());
     }
 
+    @Override
     public long size() {
         return baosList.stream().mapToLong(ByteArrayOutputStream::size).sum();
     }
@@ -201,4 +202,9 @@ public void clear() throws IOException {
     public void close() throws IOException {
         this.closed = true;
     }
+
+    @Override
+    public InputStream toInputStream() {
+        return new BigByteArrayInputStream(this);
+    }
 }