fixes

Author: clebertsuconic

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/actors/Actor.java

@@ -18,7 +18,7 @@
 
 import java.util.concurrent.Executor;
 
-public class Actor<T> extends ProcessorBase<T> {
+public class Actor<T> extends TaskProcessorBase<T> {
 
    private final ActorListener<T> listener;
 
@@ -33,7 +33,7 @@ protected final void doTask(T task) {
    }
 
    public final void act(T message) {
-      task(message);
+      addTask(message);
    }
 
 }

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/actors/OrderedExecutor.java

@@ -29,7 +29,7 @@
  * More specifically, any call B to the {@link #execute(Runnable)} method that happens-after another call A to the
  * same method, will result in B's task running after A's.
  */
-public class OrderedExecutor extends ProcessorBase<Runnable> implements ArtemisExecutor {
+public class OrderedExecutor extends TaskProcessorBase<Runnable> implements ArtemisExecutor {
 
    public OrderedExecutor(Executor delegate) {
       super(delegate);
@@ -69,7 +69,7 @@ protected final void doTask(Runnable task) {
 
    @Override
    public final void execute(Runnable run) {
-      task(run);
+      addTask(run);
    }
 
    @Override

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/actors/ProcessorBase.java

@@ -47,11 +47,11 @@ public abstract class ProcessorBase<T> extends HandlerBase {
    @SuppressWarnings("unused")
    private volatile int state = STATE_NOT_RUNNING;
    // Request of forced shutdown
-   private volatile boolean requestedForcedShutdown = false;
+   protected volatile boolean requestedForcedShutdown = false;
    // Request of educated shutdown:
    protected volatile boolean requestedShutdown = false;
    // Request to yield to another thread
-   private volatile boolean yielded = false;
+   protected volatile boolean yielded = false;
 
    private static final AtomicIntegerFieldUpdater<ProcessorBase> stateUpdater = AtomicIntegerFieldUpdater.newUpdater(ProcessorBase.class, "state");
 
@@ -61,12 +61,7 @@ private void executePendingTasks() {
          if (stateUpdater.compareAndSet(this, STATE_NOT_RUNNING, STATE_RUNNING)) {
             enter();
             try {
-               T task;
-               //while the queue is not empty we process in order:
-               //if requestedForcedShutdown==true than no new tasks will be drained from the tasks q.
-               while (!yielded && !requestedForcedShutdown && (task = tasks.poll()) != null) {
-                  doTask(task);
-               }
+               processQueue();
             } finally {
                leave();
                //set state back to not running if possible: shutdownNow could be called by doTask(task).
@@ -90,6 +85,8 @@ private void executePendingTasks() {
       }
    }
 
+   protected abstract void processQueue();
+
    /**
     * It will shutdown and wait 30 seconds for timeout.
     */
@@ -135,8 +132,6 @@ public int shutdownNow(Consumer<? super T> onPendingItem, int timeout, TimeUnit
       return pendingItems;
    }
 
-   protected abstract void doTask(T task);
-
    public ProcessorBase(Executor parent) {
       this.delegate = parent;
    }
@@ -174,7 +169,7 @@ public final boolean flush(long timeout, TimeUnit unit) {
       return this.state == STATE_NOT_RUNNING;
    }
 
-   protected void task(T command) {
+   protected void addTask(T command) {
       if (requestedShutdown) {
          logAddOnShutdown();
          return;
@@ -190,7 +185,7 @@ protected void task(T command) {
 
    /**
     * This has to be called on the assumption that state!=STATE_RUNNING.
-    * It is packed separately from {@link #task(Object)} just for performance reasons: it
+    * It is packed separately from {@link #addTask(Object)} just for performance reasons: it
     * handles the uncommon execution cases for bursty scenarios i.e. the slowest execution path.
     */
    private void onAddedTaskIfNotRunning(int state) {

artemis-commons/src/main/java/org/apache/activemq/artemis/utils/actors/ThresholdActor.java

@@ -24,7 +24,7 @@
 import org.slf4j.LoggerFactory;
 import java.lang.invoke.MethodHandles;
 
-public class ThresholdActor<T> extends ProcessorBase<Object> {
+public class ThresholdActor<T> extends TaskProcessorBase<Object> {
 
    private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
@@ -85,13 +85,13 @@ public void act(T message) {
       } else {
          logger.debug("element {} returned an invalid size over the Actor", message);
       }
-      task(message);
+      addTask(message);
    }
 
    public void flush() {
       if (SCHEDULED_FLUSH_UPDATER.compareAndSet(this, 0, 1)) {
          overThreshold.run();
-         task(FLUSH);
+         addTask(FLUSH);
       }
    }
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java

@@ -114,11 +114,6 @@ default PagingStore enforceAddressFullMessagePolicy(AddressFullMessagePolicy enf
     * It's ok to look for this with an estimate on starting a task or not, but you will need to recheck on actual paging operations. */
    boolean isPaging();
 
-   /**
-    * Schedules sync to the file storage.
-    */
-   void addSyncPoint(OperationContext context) throws Exception;
-
    /**
     * Performs a real sync on the current IO file.
     */
@@ -265,6 +260,10 @@ default void addSize(int size) {
 
    void unblock();
 
+   default boolean hasPendingIO() {
+      return false;
+   }
+
    default StorageManager getStorageManager() {
       return null;
    }

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java

@@ -209,6 +209,25 @@ public Future<Boolean> scheduleCleanup() {
 
       return future;
    }
+   public Future<Boolean> forceScheduleCleanup() {
+      final SimpleFutureImpl<Boolean> future = new SimpleFutureImpl<>();
+
+      pagingStore.execute(() -> {
+         storageManager.setContext(storageManager.newSingleThreadContext());
+         try {
+            System.out.println("Forcing cleanup");
+            if (cleanupEnabled) {
+               cleanup();
+            }
+         } finally {
+            storageManager.clearContext();
+            future.set(true);
+         }
+      });
+
+      return future;
+   }
+
 
    /**
     * Delete everything associated with any queue on this address.
@@ -271,6 +290,11 @@ protected void cleanup() {
          return;
       }
 
+      if (pagingStore.hasPendingIO()) {
+         forceScheduleCleanup();
+         return;
+      }
+
       ArrayList<Page> depagedPages = new ArrayList<>();
       LongHashSet depagedPagesSet = new LongHashSet();
 

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageSyncTimer.java

@@ -16,84 +16,121 @@
  */
 package org.apache.activemq.artemis.core.paging.impl;
 
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
+import org.apache.activemq.artemis.core.paging.PagedMessage;
 import org.apache.activemq.artemis.core.paging.PagingStore;
 import org.apache.activemq.artemis.core.persistence.OperationContext;
 import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.RouteContextList;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
+ *
+ * TODO: remove this
  * This will batch multiple calls waiting to perform a sync in a single call.
  */
 final class PageSyncTimer extends ActiveMQScheduledComponent {
 
+   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
 
+   private final PagingStoreImpl store;
 
-   private final PagingStore store;
 
-   private final ScheduledExecutorService scheduledExecutor;
+   protected final List<PageEvent> pageEvents = new ArrayList<>();
 
-   private boolean pendingSync;
+   public boolean hasPendingIO() {
+      return !pageEvents.isEmpty();
+   }
 
-   private final long timeSync;
 
-   private final Runnable runnable = this::tick;
+   public static class PageEvent {
+      PageEvent(OperationContext context, PagedMessage message, Transaction tx, RouteContextList listCtx) {
+         this.context = context;
+         this.message = message;
+         this.listCtx = listCtx;
+         this.tx = tx;
+      }
 
-   private final List<OperationContext> syncOperations = new LinkedList<>();
+      PagedMessage message;
+      OperationContext context;
+      RouteContextList listCtx;
+      Transaction tx;
+   }
 
 
-   PageSyncTimer(PagingStore store, ScheduledExecutorService scheduledExecutor, Executor executor, long timeSync) {
+   PageSyncTimer(PagingStoreImpl store, ScheduledExecutorService scheduledExecutor, Executor executor, long timeSync) {
       super(scheduledExecutor, executor, timeSync, TimeUnit.NANOSECONDS, true);
       this.store = store;
-      this.scheduledExecutor = scheduledExecutor;
-      this.timeSync = timeSync;
    }
 
 
-   synchronized void addSync(OperationContext ctx) {
-      ctx.pageSyncLineUp();
-      if (!pendingSync) {
-         pendingSync = true;
-
-         delay();
-      }
-      syncOperations.add(ctx);
+   public synchronized void addTask(OperationContext context, PagedMessage message, Transaction tx, RouteContextList listCtx) {
+      PageEvent event = new PageEvent(context, message, tx, listCtx);
+      context.storeLineUp();
+      this.pageEvents.add(event);
+      delay();
    }
 
+
    @Override
    public void run() {
       tick();
    }
 
-   private void tick() {
-      OperationContext[] pendingSyncsArray;
+   private PageEvent[] extractPendingEvents() {
       synchronized (this) {
+         if (pageEvents.isEmpty()) {
+            return null;
+         }
+         PageEvent[] pendingsWrites = new PageEvent[pageEvents.size()];
+         pendingsWrites = pageEvents.toArray(pendingsWrites);
+         pageEvents.clear();
+         return pendingsWrites;
+      }
+   }
 
-         pendingSync = false;
-         pendingSyncsArray = new OperationContext[syncOperations.size()];
-         pendingSyncsArray = syncOperations.toArray(pendingSyncsArray);
-         syncOperations.clear();
+   private void tick() {
+      PageEvent[] pendingEvents = extractPendingEvents();
+      if (pendingEvents == null) {
+         return;
       }
 
       try {
-         if (pendingSyncsArray.length != 0) {
+
+         for (PageEvent event : pendingEvents) {
+            store.writePageCallback(event.message, event.tx, event.listCtx);
+         }
+
+         try {
             store.ioSync();
+         } catch (Exception e) {
+            // TODO - do not merge this TODO - critical error here, server should shutdown
+            logger.warn(e.getMessage(), e);
          }
+
+
       } catch (Exception e) {
-         for (OperationContext ctx : pendingSyncsArray) {
-            ctx.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getClass() + " during ioSync for paging on " + store.getStoreName() + ": " + e.getMessage());
+         for (PageEvent event : pendingEvents) {
+            event.context.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getClass() + " during ioSync for paging on " + store.getStoreName() + ": " + e.getMessage());
          }
       } finally {
          // In case of failure, The context should propagate an exception to the client
          // We send an exception to the client even on the case of a failure
          // to avoid possible locks and the client not getting the exception back
-         for (OperationContext ctx : pendingSyncsArray) {
-            ctx.pageSyncDone();
+         for (PageEvent event : pendingEvents) {
+            event.context.done();
          }
       }
    }

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryDatabase.java

@@ -153,7 +153,7 @@ public PageCursorProvider newCursorProvider(PagingStore store,
    @Override
    public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) {
 
-      return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), ioExecutorFactory.getExecutor(), syncNonTransactional);
+      return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), syncNonTransactional);
    }
 
    @Override
@@ -233,7 +233,7 @@ public synchronized List<PagingStore> reloadStores(final HierarchicalRepository<
 
          AddressSettings settings = addressSettingsRepository.getMatch(address.toString());
 
-         PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor().setFair(true), ioExecutorFactory.getExecutor(), syncNonTransactional);
+         PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor().setFair(true), syncNonTransactional);
 
          storesReturn.add(store);
       }

artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java

@@ -153,7 +153,7 @@ public PageCursorProvider newCursorProvider(PagingStore store,
    @Override
    public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) {
 
-      return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), ioExecutorFactory.getExecutor(), syncNonTransactional);
+      return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), syncNonTransactional);
    }
 
    @Override
@@ -230,7 +230,7 @@ public List<PagingStore> reloadStores(final HierarchicalRepository<AddressSettin
 
             AddressSettings settings = addressSettingsRepository.getMatch(address.toString());
 
-            PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor(), executorFactory.getExecutor(), syncNonTransactional);
+            PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor(), syncNonTransactional);
 
             storesReturn.add(store);
          }