Delayed, concurrent event stack in Java - follow-up

Question

I have slightly refactored the Delayed, concurrent event stack in Java. Now it looks like this:

DelayedEventStack.java

package com.github.coderodde.cconcurrent.eventstack;

import java.util.Deque;
import java.util.Objects;
import java.util.concurrent.ConcurrentLinkedDeque;

/**
 * This class implements an delayed event stack.
 * 
 * @author Rodion "rodde" Efremov
 * @version 1.61 (Jan 14, 2023)
 * @since 1.6 (Jan 12, 2023)
 */
public final class DelayedEventStack implements AutoCloseable {

    /**
     * This static inner class implements the event stack entry.
     */
    private static final class DelayedEventStackEntry {
        private final Runnable onExpired;
        private final long expirationMillis;

        DelayedEventStackEntry(Runnable onExpired, long expirationMillis) {
            this.onExpired = 
                    Objects.requireNonNull(onExpired, "onExpired is null.");

            this.expirationMillis = expirationMillis;
        }
    }

    /**
     * The event loop flag. 
     */
    private volatile boolean doRunFlag = true;

    /**
     * The actual event stack.
     */
    private final Deque<DelayedEventStackEntry> delayedEventStack = 
            new ConcurrentLinkedDeque<>();

    /**
     * This flag specifies whether the stack runs the leftover events in the 
     * stack upon closing the stack.
     */
    private final boolean dischargeRemainingEventsOnClose;

    /** 
     * The worker thread.
     */
    private final Thread workerThread = new Thread() {
        @Override
        public void run() {
            while (doRunFlag) {
                // Ask delayedEventStack for a topmost event entry. Gets null if 
                // the stack is empty, in which case we sleep a millisecond and 
                // ask one more time.
                DelayedEventStackEntry topmostEventStackEntry = 
                        delayedEventStack.peekLast();

                if (topmostEventStackEntry == null
                        || System.currentTimeMillis() < 
                           topmostEventStackEntry.expirationMillis) {
                    // Once here, nothing to do.
                    Utils.sleep(1L);
                } else {
                    delayedEventStack.removeLast().onExpired.run();
                }
            }
        }
    };

    public DelayedEventStack() {
        this(true);
    }

    public DelayedEventStack(boolean dischargeRemainingEventsOnClose) {
        this.dischargeRemainingEventsOnClose = dischargeRemainingEventsOnClose;
        workerThread.start();
    }

    public void add(Runnable onAdd, Runnable onExpired, long durationMillis) {
        delayedEventStack.addLast(
                new DelayedEventStackEntry(
                        onExpired, 
                        System.currentTimeMillis() + durationMillis));

        onAdd.run();
    }

    @Override
    public void close() {
        doRunFlag = false;

        if (dischargeRemainingEventsOnClose) {
            while (!delayedEventStack.isEmpty()) {
                delayedEventStack.removeLast().onExpired.run();
            }
        }
    }
}

Utils.java

package com.github.coderodde.cconcurrent.eventstack;

import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * This class is supposed to hold miscellaneous utility methods.
 * 
 * @author Rodion "rodde" Efremov
 * @version 1.6 (Jan 14, 2023)
 * @since 1.6 (Jan 14, 2023)
 */
public final class Utils {

    private static final Logger LOGGER = 
            Logger.getLogger(Utils.class.getName());

    private Utils() {

    }

    public static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException ex) {
            LOGGER.log(Level.WARNING, "Interrupted while sleeping.");
        }
    }
}

Demo.java

package com.github.coderodde.cconcurrent.eventstack;

import static com.github.coderodde.cconcurrent.eventstack.Utils.sleep;
import java.util.logging.Level;
import java.util.logging.Logger;

public final class Demo {

    public static void main(String[] args) {
        System.out.println("Simulation started.");

        DelayedEventStack eventStack = new DelayedEventStack();

        eventStack.add(() -> { System.out.println("Event 1 start"); },
                       () -> { System.out.println("Event 1 end");},
                       10_000L);

        sleep(2_000L);

        eventStack.add(() -> { System.out.println("Event 2 start"); },
                       () -> { System.out.println("Event 2 end");},
                       3000L);

        sleep(7_000L);

        eventStack.add(() -> { System.out.println("Leftover event start"); },
                       () -> { System.out.println("Leftover event end"); },
                       10_000L);

        CloseThread closeThread = new CloseThread(eventStack);
        closeThread.start();
    }
}

/**
 * This class is responsible for closing the event stack.
 */
class CloseThread extends Thread {

    private static final Logger logger = 
            Logger.getLogger(CloseThread.class.getName());

    private final DelayedEventStack delayedEventStack;

    CloseThread(DelayedEventStack delayedEventStack) {
        this.delayedEventStack = delayedEventStack;
    }

    @Override
    public void run() {
        Utils.sleep(10_000L);

        try {
            delayedEventStack.close();
        } catch (Exception ex) {
            logger.log(
                    Level.SEVERE,
                    "An exception was thrown upon closing the event stack: {0}", 
                    ex.getMessage());
        }
    }
}

Critique request

Since I am not good at writing concurrent code, I need your, guys, help to make it mature.

Could you please describe what this class is supposed to do? It somewhat looks like a DelayQueue. Should it execute evens in order, i.e. earlier events should be executed first even if later ones already expires? — Flame239
– Flame239, Commented Jan 20, 2023 at 14:54

Flame239 · Accepted Answer · 2023-01-20 16:28:50Z

Biggest problem I see is busy waiting.

Waiting even 1ms in a loop is not a good practice, because you are wasting CPU anyways. Better option would be to wait indefinitely if stack is empty or wait for the time remaining until "top" event expiration.

If we add new event, we should add it to the stack first and then notify waiting thread, which will re-check the top of the stack and act accordingly.

close() method is a bit trickier and depends what you want to do here. One good approach is to forbid adding new events and finish when stack is empty, i.e. don't wait if no elements, but just return. Anyways we should account that working thread might be waiting at the time we call close(), so we might need to notify it.

Code tips:

use push(), pop() and peek() instead of ..Last() for better readability
rename class methods to push() and pop() (class is called ...Stack afterall)

give worker thread a name

   public final class DelayedEventStack implements AutoCloseable {

      private volatile boolean doRunFlag = true;
      // no need for synchronized collection, since we operate it under the lock (max 1 thread is modifying/reading it at the time)
      private final Deque<DelayedEventStackEntry> delayedEventStack = new ArrayDeque<>();
      private final boolean dischargeRemainingEventsOnClose;
      private final ReentrantLock lock;
      private final Condition available;

      private final Thread workerThread = new Thread(() -> {
          while (doRunFlag) {
              try {
                  // note: here we don't hold a lock while running
                  pop().onExpired.run();
              } catch (InterruptedException e) {
                  // process exception
              }
          }
      }, "DelayedEventStackThread");

      public DelayedEventStack(boolean dischargeRemainingEventsOnClose) {
          this.dischargeRemainingEventsOnClose = dischargeRemainingEventsOnClose;
          lock = new ReentrantLock();
          available = lock.newCondition();
          workerThread.start();
      }

      private DelayedEventStackEntry pop() throws InterruptedException {
          // we should be holding the lock while calling `await()`
          lock.lock();
          try {
              // need a loop here to re-check the top
              while (true) {
                  DelayedEventStackEntry top = delayedEventStack.peek();
                  if (top == null) {
                      if (!doRunFlag) return new DelayedEventStackEntry(() -> {}, 0);

                      // if empty - wait indefinitely i.e. until we got signalled
                      available.await();
                  } else {
                      long delay = top.expirationMillis - System.currentTimeMillis();
                      if (delay <= 0) {
                          // event is ready to be executed
                          return delayedEventStack.pop();
                      }
                      // wait until event start time
                      available.awaitNanos(delay * 1000000);
                  }
              }
          } finally {
              lock.unlock();
          }
      }

      public void push(Runnable onExpired, long durationMillis) {
          if (!doRunFlag)
              throw new UnsupportedOperationException("Event stack is closed; adding new events is impossible");

          lock.lock();
          try {
              delayedEventStack.push(new DelayedEventStackEntry(onExpired, System.currentTimeMillis() + durationMillis));
              // signal waiting thread to re-check
              available.signal();
          } finally {
              lock.unlock();
          }
      }

      @Override
      public void close() {
          doRunFlag = false;
          lock.lock();
          try {
              // if we don't want to run reminaing events - just remove them from the stack
              if (!dischargeRemainingEventsOnClose) {
                  delayedEventStack.clear();
              }
              // wake up waiting thread to re-check top
              available.notify();
          } finally {
              lock.unlock();
          }
      }
  }

NB: code is not tested

Stack Exchange Network

Delayed, concurrent event stack in Java - follow-up

1 Answer 1

You must log in to answer this question.

Linked

Hot Network Questions

Delayed, concurrent event stack in Java - follow-up

1 Answer 1

You must log in to answer this question.

Linked

Related

Hot Network Questions