DelayedQueue

• Interfaces to be implemented

• Single thread

• One thread for each task

• PriorityQueue + A background thread

Interfaces to be implemented

public interface Scheduler
{
    void schedule( Task t, long delayMs );
}

public interface Task
{
    void run();
}

Single thread

• Main thread is in Timedwaiting state for delayMs for each call of schedule()

• Only one thread, very low CPU utilization

• Also, this is not working as later call

• How about sleeping in other threads

public class SchedulerImpl implements Scheduler
{
    public void schedule( Task t, long delayMs )
    {
        try
        {
            // sleep for delayMs, and then execute the task
            Thread.sleep( delayMs );
            t.run();
        }
        catch ( InterruptedException e )
        {
            // ignore
        }
    }

    public static void main( String[] args )
    {
        Scheduler scheduler = new SchedulerImpl();
        Task t1 = new TaskImpl( 1 );
        Task t2 = new TaskImpl( 2 );

        // main thread in timedwaiting state for 10000 ms
        scheduler.schedule( t1, 10000 );
        scheduler.schedule( t2, 1 );
    }
}

One thread for each task

• No blocking when calling schedule

• What happens if we call schedule many times

  • A lot of thread creation overhead

• Call be alleviated by using a thread pool, but still not ideal

public class SchedulerImpl implements Scheduler
{
    public void schedule( Task t, long delayMs )
    {
        Thread t = new Thread( new Runnable() {
            public void run()
            {
                try 
                {
                    Thread.sleep( delayMs );
                    t.run();
                }
                catch ( InterruptedException e )
                {
                    // ignore;
                }
            }
        } );
        t.start();
    }
}

PriorityQueue + A background thread

package designThreadSafeEntity.delayedTaskScheduler;

import java.util.PriorityQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class Scheduler
{
    // order task by time to run
    private PriorityQueue<Task> tasks;

    // 
    private final Thread taskRunnerThread;

    // State indicating the scheduler is running
    // Why volatile? As long as main thread stops, runner needs to has visibility.
    private volatile boolean running;

    // Task id to assign to submitted tasks
    // AtomicInteger: Threadsafe. Do not need to add locks when assigning task Ids
    // Final: Reference of atomicInteger could not be changed
    private final AtomicInteger taskId;

    public Scheduler()
    {
        tasks = new PriorityQueue<>();
        taskRunnerThread = new Thread( new TaskRunner() );
        running = true;
        taskId = new AtomicInteger( 0 );

        // start task runner thread
        taskRunnerThread.start();
    }

    public void schedule( Task task, long delayMs )
    {
        // Set time to run and assign task id
        long timeToRun = System.currentTimeMillis() + delayMs;
        task.setTimeToRun( timeToRun );
        task.setId( taskId.incrementAndGet() );

        // Put the task in queue
        synchronized ( this )
        {
            tasks.offer( task );
            this.notify(); // only a single background thread waiting
        }
    }

    public void stop( ) throws InterruptedException
    {
        // Notify the task runner as it may be in wait()
        synchronized ( this )
        {
            running = false;
            this.notify();
        }

        // Wait for the task runner to terminate
        taskRunnerThread.join();
    }

    private class TaskRunner implements Runnable
    {
        @Override
        public void run()
        {
            while ( running )
            {
                // Need to synchronize with main thread
                synchronized( Scheduler.this )
                {
                    try 
                    {
                        // task runner is blocked when no tasks in queue
                        while ( running && tasks.isEmpty() )
                        {
                            Scheduler.this.wait();
                        }

                        // check the first task in queue
                        long now = System.currentTimeMillis();
                        Task t = tasks.peek();

                        // delay exhausted, execute task
                        if ( t.getTimeToRun() < now )
                        {
                            tasks.poll();
                            t.run();
                        }            
                        else
                        {
                            // no task executable, wait
                            Scheduler.this.wait( t.getTimeToRun() - now );
                        }
                    }
                    catch ( InterruptedException e )
                    {
                        Thread.currentThread().interrupt();    
                    }
                }
            }
        }
    }

    public static void main( String[] args ) throws InterruptedException
    {
        Scheduler scheduler = new Scheduler();
        scheduler.schedule( new Task(), 1000000 );
        scheduler.schedule( new Task(), 1000 );
        Thread.sleep( 7000 );
        scheduler.stop();
    }
}

class Task implements Comparable<Task> 
{
    // When the task will be run
    private long timeToRun;
    private int id;

    public void setId( int id )
    {
        this.id = id;
    }

    public void setTimeToRun( long timeToRun )
    {
        this.timeToRun = timeToRun;
    }

    public void run()
    {
        System.out.println( "Running task " + id );
    }

    public int compareTo( Task other )
    {
        return (int) ( timeToRun - other.getTimeToRun() );
    }

    public long getTimeToRun()
    {
        return timeToRun;
    }
}

package designThreadSafeEntity.delayedTaskScheduler;

import java.util.PriorityQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class Scheduler
{
    private PriorityQueue<Task> tasks;
    private final Thread taskRunnerThread;
    private volatile boolean running;
    private final AtomicInteger taskId;

    public Scheduler()
    {
        tasks = new PriorityQueue<>();
        taskRunnerThread = new Thread( new TaskRunner() );
        running = true;
        taskId = new AtomicInteger( 0 );
        taskRunnerThread.start();
    }

    public void schedule( Task task, long delayMs )
    {
        long timeToRun = System.currentTimeMillis() + delayMs;
        task.setTimeToRun( timeToRun );
        task.setId( taskId.incrementAndGet() );
        synchronized ( this )
        {
            tasks.offer( task );
            this.notify();
        }
    }

    public void stop( ) throws InterruptedException
    {
        synchronized ( this )
        {
            running = false;
            this.notify();
        }
        taskRunnerThread.join();
    }

    private class TaskRunner implements Runnable
    {
        @Override
        public void run()
        {
            while ( running )
            {
                synchronized( Scheduler.this )
                {
                    try 
                    {
                        while ( running && tasks.isEmpty() )
                        {
                            Scheduler.this.wait();
                        }
                        long now = System.currentTimeMillis();
                        Task t = tasks.peek();
                        if ( t.getTimeToRun() < now )
                        {
                            tasks.poll();
                            t.run();
                        }            
                        else
                        {
                            Scheduler.this.wait( t.getTimeToRun() - now );
                        }
                    }
                    catch ( InterruptedException e )
                    {
                        Thread.currentThread().interrupt();    
                    }
                }
            }
        }
    }

    public static void main( String[] args ) throws InterruptedException
    {
        Scheduler scheduler = new Scheduler();
        scheduler.schedule( new Task(), 1000000 );
        scheduler.schedule( new Task(), 1000 );
        Thread.sleep( 7000 );
        scheduler.stop();
    }
}

class Task implements Comparable<Task> 
{
    private long timeToRun;
    private int id;

    public void setId( int id )
    {
        this.id = id;
    }

    public void setTimeToRun( long timeToRun )
    {
        this.timeToRun = timeToRun;
    }

    public void run()
    {
        System.out.println( "Running task " + id );
    }

    public int compareTo( Task other )
    {
        return (int) ( timeToRun - other.getTimeToRun() );
    }

    public long getTimeToRun()
    {
        return timeToRun;
    }
}