0
\$\begingroup\$

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.

\$\endgroup\$
2
  • 1
    \$\begingroup\$ 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? \$\endgroup\$
    – Flame239
    Commented Jan 20, 2023 at 14:54
  • \$\begingroup\$ @Flame239 +1. I will add description tomorrow; now tired. \$\endgroup\$
    – coderodde
    Commented Jan 20, 2023 at 16:14

1 Answer 1

1
\$\begingroup\$

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

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.