ARTEMIS-353 retrying after interrupts on the native layer

artemis-journal/src/main/java/org/apache/activemq/artemis/core/io/aio/AIOSequentialFileFactory.java

@@ -19,12 +19,7 @@
 import java.io.File;
 import java.io.IOException;
 import java.nio.ByteBuffer;
-import java.security.AccessController;
-import java.security.PrivilegedAction;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import org.apache.activemq.artemis.ArtemisConstants;
@@ -38,7 +33,6 @@
 import org.apache.activemq.artemis.jlibaio.SubmitInfo;
 import org.apache.activemq.artemis.jlibaio.util.CallbackCache;
 import org.apache.activemq.artemis.journal.ActiveMQJournalLogger;
-import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
 
 public final class AIOSequentialFileFactory extends AbstractSequentialFileFactory {
 
@@ -48,7 +42,7 @@ public final class AIOSequentialFileFactory extends AbstractSequentialFileFactor
 
    private volatile boolean reuseBuffers = true;
 
-   private ExecutorService pollerExecutor;
+   private Thread pollerThread;
 
    volatile LibaioContext<AIOSequentialCallback> libaioContext;
 
@@ -195,14 +189,8 @@ public void start() {
 
          this.running.set(true);
 
-         pollerExecutor = Executors.newCachedThreadPool(AccessController.doPrivileged(new PrivilegedAction<ActiveMQThreadFactory>() {
-            @Override
-            public ActiveMQThreadFactory run() {
-               return new ActiveMQThreadFactory("ActiveMQ-AIO-poller-pool" + System.identityHashCode(this), true, AIOSequentialFileFactory.class.getClassLoader());
-            }
-         }));
-
-         pollerExecutor.execute(new PollerRunnable());
+         pollerThread = new PollerThread();
+         pollerThread.start();
       }
 
    }
@@ -215,11 +203,11 @@ public void stop() {
          libaioContext.close();
          libaioContext = null;
 
-         if (pollerExecutor != null) {
-            pollerExecutor.shutdown();
-
+         if (pollerThread != null) {
             try {
-               if (!pollerExecutor.awaitTermination(AbstractSequentialFileFactory.EXECUTOR_TIMEOUT, TimeUnit.SECONDS)) {
+               pollerThread.join(AbstractSequentialFileFactory.EXECUTOR_TIMEOUT * 1000);
+
+               if (pollerThread.isAlive()) {
                   ActiveMQJournalLogger.LOGGER.timeoutOnPollerShutdown(new Exception("trace"));
                }
             }
@@ -232,11 +220,6 @@ public void stop() {
       }
    }
 
-   @Override
-   protected void finalize() {
-      stop();
-   }
-
    /**
     * The same callback is used for Runnable executor.
     * This way we can save some memory over the pool.
@@ -348,11 +331,22 @@ public void sequentialDone() {
       }
    }
 
-   private class PollerRunnable implements Runnable {
+   private class PollerThread extends Thread {
+
+      public PollerThread() {
+         super("Apache ActiveMQ Artemis libaio poller");
+      }
 
       @Override
       public void run() {
-         libaioContext.poll();
+         while (running.get()) {
+            try {
+               libaioContext.poll();
+            }
+            catch (Throwable e) {
+               ActiveMQJournalLogger.LOGGER.warn(e.getMessage(), e);
+            }
+         }
       }
    }
 

artemis-native/src/main/c/org_apache_activemq_artemis_jlibaio_LibaioContext.c

@@ -557,9 +557,16 @@ JNIEXPORT void JNICALL Java_org_apache_activemq_artemis_jlibaio_LibaioContext_bl
 
         int result = io_getevents(theControl->ioContext, 1, max, theControl->events, 0);
 
+        if (result == -EINTR)
+        {
+           // ARTEMIS-353: jmap will issue some weird interrupt signal what would break the execution here
+           // we need to ignore such calls here
+           continue;
+        }
+
         if (result < 0)
         {
-            throwIOExceptionErrorNo(env, "Error while submitting IO: ", -result);
+            throwIOExceptionErrorNo(env, "Error while calling io_getevents IO: ", -result);
             break;
         }
         #ifdef DEBUG

artemis-native/src/main/java/org/apache/activemq/artemis/jlibaio/LibaioContext.java

@@ -49,7 +49,7 @@ public class LibaioContext<Callback extends SubmitInfo> implements Closeable {
     * <br>
     * Or else the native module won't be loaded because of version mismatches
     */
-   private static final int EXPECTED_NATIVE_VERSION = 5;
+   private static final int EXPECTED_NATIVE_VERSION = 6;
 
    private static boolean loaded = false;