Deferred Callback Executor

📌 Problem Statement

Design and implement a thread-safe system that lets users register callbacks which are automatically executed after a specified delay.

✅ Requirements:

• Register callbacks with delays (in seconds)
• Safe for multiple threads
• Must execute callbacks on time
• Should be efficient (no busy-waiting)
• Reliable execution

❗ Constraints

1. Thread Safety: Must handle concurrent registration and execution safely
2. Timing Accuracy: Callbacks should execute as close to the specified time as possible
3. Memory Efficiency: Should not consume excessive memory for long-running systems
4. Scalability: Should handle a large number of callbacks efficiently
5. Resource Management: Must properly manage threads and prevent resource leaks

📥 Input / Output

Input:

• Callback functions/tasks to be executed
• Delay time in seconds for each callback
• Multiple threads registering callbacks simultaneously

Output:

• Callbacks executed at their scheduled times
• Thread-safe execution without race conditions
• Proper resource cleanup

💡 Intuition About Solving This Problem

The key insight is that we need a priority queue to store callbacks sorted by their execution time, combined with proper synchronization to handle concurrent access. The main thread continuously checks the queue for ready callbacks while other threads can safely register new ones.

🚀 Steps to Take to Solve

1. Design the data structure: Use a priority queue to store callbacks sorted by execution time
2. Implement thread safety: Use locks and conditions to synchronize access
3. Create the executor thread: Main thread that continuously processes the queue
4. Handle timing: Calculate wait times and use condition variables for efficient waiting
5. Manage registration: Thread-safe method to add new callbacks
6. Implement shutdown: Graceful termination of the executor

⚠️ Edge Cases to Consider

1. Empty queue: Handle when no callbacks are available
2. Shutdown during execution: Proper cleanup when stopping the executor
3. Exception handling: What happens if a callback throws an exception
4. Memory overflow: Handle cases with many callbacks
5. Clock drift: System time changes affecting execution timing
6. Interrupted threads: Handle ThreadInterruptedException properly

💻 Code (with comments)

Approach 1: Custom Deferred Executor

import java.util.PriorityQueue;
import java.util.Comparator;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.Condition;

public class DeferredCallbackExecutor {
    private final PriorityQueue<Callback> queue = new PriorityQueue<>(
        Comparator.comparingLong(callback -> callback.executeAt)
    );
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition newCallbackArrived = lock.newCondition();
    private volatile boolean running = true;

    public void start() throws InterruptedException {
        lock.lock();
        try {
            while (running) {
                while (queue.isEmpty() && running) {
                    newCallbackArrived.await(); // Wait for new callbacks
                }
                if (!running) break;

                Callback nextCallback = queue.peek();
                long waitTime = nextCallback.executeAt - System.currentTimeMillis();
                while (waitTime > 0 && running) {
                    newCallbackArrived.awaitNanos(waitTime * 1_000_000); // Wait until execution time
                    waitTime = nextCallback.executeAt - System.currentTimeMillis();
                }

                if (!running) break;
                nextCallback = queue.poll();
                if (nextCallback != null) {
                    System.out.println("Executing: " + nextCallback.message);
                }
            }
        } finally {
            lock.unlock();
        }
    }

    public void registerCallback(Callback callback) {
        lock.lock();
        try {
            queue.add(callback);
            newCallbackArrived.signal(); // Notify waiting executor thread
        } finally {
            lock.unlock();
        }
    }

    public void stop() {
        running = false;
        lock.lock();
        try {
            newCallbackArrived.signal(); // Wake up executor thread to check running flag
        } finally {
            lock.unlock();
        }
    }

    static class Callback {
        final long executeAt;
        final String message;

        public Callback(long executeAfter, String message) {
            this.executeAt = System.currentTimeMillis() + executeAfter * 1000;
            this.message = message;
        }
    }
}

Approach 2: ScheduledExecutorService

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class DeferredCallbackScheduler {
    private final ScheduledExecutorService scheduler;
    private volatile boolean running = true;

    public DeferredCallbackScheduler(int poolSize) {
        this.scheduler = Executors.newScheduledThreadPool(poolSize);
    }

    public DeferredCallbackScheduler() {
        this(1);
    }

    public void schedule(Runnable callback, long delaySeconds) {
        if (callback == null) throw new IllegalArgumentException("Callback cannot be null");
        if (delaySeconds < 0) throw new IllegalArgumentException("Delay must be non-negative");
        if (!running) throw new IllegalStateException("Scheduler has been stopped");

        scheduler.schedule(() -> {
            try {
                callback.run();
            } catch (Exception e) {
                System.err.println("Callback execution failed: " + e.getMessage());
            }
        }, delaySeconds, TimeUnit.SECONDS);
    }

    public void stop() {
        running = false;
        scheduler.shutdown();
        try {
            if (!scheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                scheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            scheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

🔍 Dry Run Example

Scenario: Registering 3 callbacks with different delays

Initial State:

• Empty priority queue
• Executor thread waiting for callbacks

Step 1: Register callback with 2-second delay

• Callback added to queue with executeAt = currentTime + 2000ms
• Executor thread notified via signal()
• Executor checks queue, finds callback, calculates wait time

Step 2: Register callback with 3-second delay

• Callback added to queue with executeAt = currentTime + 3000ms
• Queue automatically sorted by executeAt time
• Executor continues waiting for first callback

Step 3: Register callback with 1-second delay

• Callback added to queue with executeAt = currentTime + 1000ms
• Queue re-sorted: [1s, 2s, 3s]
• Executor notified, recalculates wait time for 1s callback

Step 4: Execution

• After 1 second: First callback executes
• After 2 seconds: Second callback executes
• After 3 seconds: Third callback executes

Result: All callbacks execute in correct order at their scheduled times.

⏱️ Time Complexity / Space Complexity

Time Complexity:

• Registration: O(log n) - inserting into priority queue
• Execution: O(log n) - removing from priority queue
• Queue operations: O(log n) for add/remove operations

Space Complexity:

• Storage: O(n) where n is the number of callbacks
• Thread overhead: O(1) - single executor thread
• Lock overhead: O(1) - constant space for locks and conditions

---

🧪 Usage Example (Custom):

public static void main(String[] args) throws InterruptedException {
    DeferredCallbackExecutor executor = new DeferredCallbackExecutor();

    Thread executorThread = new Thread(() -> {
        try {
            executor.start();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
    executorThread.start();

    executor.registerCallback(new Callback(2, "First callback"));
    executor.registerCallback(new Callback(3, "Second callback"));
    executor.registerCallback(new Callback(1, "Third callback"));

    Thread.sleep(5000);

    executor.stop();
    executorThread.join();
}

---

🧪 Usage Example (ScheduledExecutorService):

public static void main(String[] args) throws InterruptedException {
    DeferredCallbackScheduler scheduler = new DeferredCallbackScheduler();

    scheduler.schedule(() -> System.out.println("Task 1"), 3);
    scheduler.schedule(() -> System.out.println("Task 2"), 1);
    scheduler.schedule(() -> System.out.println("Task 3"), 5);

    Thread.sleep(7000);

    scheduler.stop();
}

---

🧠 Summary

Feature	Custom Executor	ScheduledExecutorService
Code Simplicity	❌ Complex	✅ Simple
Thread Safety	✅ Lock-based	✅ Built-in
Delay Precision	✅ Manual control	✅ Built-in
Reusability	Moderate	High
Use Case	Education, Custom behavior	Production ready

---

🟢 Choose ScheduledExecutorService when ease and robustness matter.
🧪 Use Custom Executor to learn threading, locking, and task scheduling internals.