implementing PriorityQueue on ThreadPoolExecutor

Question

Been struggling with this for over 2 days now.

implemented the answer I saw on here Specify task order execution in Java

public class PriorityExecutor extends ThreadPoolExecutor {

public PriorityExecutor(int corePoolSize, int maximumPoolSize,
                        long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {
    super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}
//Utitlity method to create thread pool easily
public static ExecutorService newFixedThreadPool(int nThreads) {
    return new PriorityExecutor(nThreads, nThreads, 0L,
            TimeUnit.MILLISECONDS, new PriorityBlockingQueue<Runnable>());
}
//Submit with New comparable task
public Future<?> submit(Runnable task, int priority) {
    return super.submit(new ComparableFutureTask(task, null, priority));
}
//execute with New comparable task
public void execute(Runnable command, int priority) {
    super.execute(new ComparableFutureTask(command, null, priority));
}
}

public class ComparableFutureTask<T> extends FutureTask<T>
    implements
    Comparable<ComparableFutureTask<T>> {

volatile int priority = 0;

public ComparableFutureTask(Runnable runnable, T result, int priority) {
    super(runnable, result);
    this.priority = priority;
}
public ComparableFutureTask(Callable<T> callable, int priority) {
    super(callable);
    this.priority = priority;
}

@Override
public int compareTo(ComparableFutureTask<T> o) {
    return Integer.valueOf(priority).compareTo(o.priority);
}
}

The Runnable I use: MyTask

public class MyTask implements Runnable{

 public MyTask(File file, Context context, int requestId) {
    this._file = file;
    this.context = context;
    this.requestId = requestId;
}

@Override
public void run() {
      // some work
    } catch (IOException e) {
        Log.e("Callable try", post.toString());

    }
}

My service: MediaDownloadService

public class MediaDownloadService extends Service {

private DBHelper helper;
Notification notification;
HashMap<Integer,Future> futureTasks = new HashMap<Integer, Future>();
final int _notificationId=1;
File file;

@Override
public IBinder onBind(Intent intent) {
    return sharonsBinder;
}


@Override
public int onStartCommand(Intent intent, int flags, int startId) {
    helper = new DBHelper(getApplicationContext());
    PriorityExecutor executor = (PriorityExecutor) PriorityExecutor.newFixedThreadPool(3);
    Log.e("requestsExists", helper.requestsExists() + "");
   if(helper.requestsExists()){
        // map of the index of the request and the string of the absolute path of the request
        Map<Integer,String> requestMap = helper.getRequestsToExcute(0);
        Set<Integer> keySet = requestMap.keySet();
        Iterator<Integer> iterator = keySet.iterator();
        Log.e("MAP",requestMap.toString());
        //checks if the DB requests exists
        if(!requestMap.isEmpty()){
            //execute them and delete the DB entry
            while(iterator.hasNext()){
                int iteratorNext = iterator.next();
                Log.e("ITREATOR", iteratorNext + "");
                file = new File(requestMap.get(iteratorNext));
                Log.e("file", file.toString());
                Log.e("thread Opened", "Thread" + iteratorNext);
                Future future = executor.submit(new MyTask(file, this, iteratorNext),10);
                futureTasks.put(iteratorNext, future);
                helper.requestTaken(iteratorNext);
            }
            Log.e("The priority queue",executor.getQueue().toString());
        }else{

            Log.e("stopself", "stop self after this");
            this.stopSelf();
        }
    }
    return START_STICKY;
}

keep getting an error on this line : Future future = executor.submit(new MyTask(file, this, iteratorNext),10);

Even tho an executor.submit(); suppose to return a future object i keep getting

Caused by: java.lang.ClassCastException: java.util.concurrent.FutureTask cannot be cast to java.lang.Comparable
        at java.util.concurrent.PriorityBlockingQueue.siftUpComparable(PriorityBlockingQueue.java:318)
        at java.util.concurrent.PriorityBlockingQueue.offer(PriorityBlockingQueue.java:450)
        at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1331)
        at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:81)
        at com.vit.infibond.test.PriorityExecutor.submit(PriorityExecutor.java:26)
        at com.vit.infibond.test.MediaDownloadService.onStartCommand(MediaDownloadService.java:65)

Can anyone save me from this nightmare?

I tried doing as this answer suggest as well Testing PriorityBlockingQueue in ThreadPoolExecutor

by adding the forNewTask override only to get casting execption again but this time for RunnableFuture.

I understand something basic is missing on my understanding and would appreciate a depth explanation...

Thanks sharon, I have extended on your comments below. rather an intense topic — V H, Sep 23 '16 at 14:06

score 2 · Accepted Answer · answered Jun 01 '15 at 13:39

2

By looking at the source code for java.util.concurrent.ThreadPoolExecutor it seems to be a real pain to get this working when submitting futures. You have to override protected methods that feels internal and do some nasty casts.

I suggest you simply use the execute method instead. There is no wrapping of the Runnable going on there so it should work.

If you need to wait for the results of your jobs I suggest implementing that on your own to avoid having to mess around with the ThreadPoolExecutor internals.

answered Jun 01 '15 at 13:39

K Erlandsson

13,408
6
51
67

My problem is not waiting for my jobs to complete but I want to mange them as they are in queue. For example i got items on queue and i need right now to improve one of their priorities, so i need to have a reference to that future object so i can cancel that task and reinsert it with an higher priority. And ill appreciate if explain a bit more about implementing it on my own, you mean to create a thread pool by myself? – sharon gur Jun 01 '15 at 13:44
To change the priority you need access to the `ComparableFutureTask` since that is what is on the queue, not the `Future`. Hence, for that purpose you can still use the `executor.execute()` instead of `submit()`. – K Erlandsson Jun 01 '15 at 13:55
executor.execute() will return a ComparableFutureTask? If not how will I reach the task I want to cancel..? I thought that to change priority I must cancel and reinsert the task, and cancel is done through the future object.. or is it? – sharon gur Jun 01 '15 at 13:58
@sharongur No it doesn't, but since you have implemented the PriorityExecutor you can make a custom method to return it when executing. – K Erlandsson Jun 01 '15 at 14:03
Brilliant! Absolutley brilliant! I changed the execute method to return a ComparableFutureTask and added a method to increase priority with an id for my hashmap that contains the Comparable objects, then just change its priority Thanks a lot! – sharon gur Jun 01 '15 at 14:22

score 2 · Answer 2 · answered Sep 23 '16 at 13:57

What sharon gur has suggest at the very bottom is to change

//execute with New comparable task
public void execute(Runnable command, int priority) {
    super.execute(new ComparableFutureTask(command, null, priority));
}

to

//execute with New comparable task
public ComparableFutureTask  execute(Runnable command, int priority) {
    ComparableFutureTask task = new ComparableFutureTask(command, null, priority);
    super.execute(task);
    return task;
}

Then in your caller:

CurrentTask currentTask = new CurrentTask(priority,queue)
RunnableFuture task = enhancedExecutor.execute(currentTask,priority.value)
task?.get()

I had a problem where

RunnableFuture task = myExecutor.submit(currentTask)
task?.get()

caused an issue where currentTask was then converted to FutureTask and had no understanding of my objects within CurrentTask. As .execute alone everything was fine. This hack appears to semi / near enough work.

so as it worked perfectly but no file was produced

RunnableFuture task = myExecutor.execuute(currentTask)
    task?.get()

So this is how I made it work (priority is being handled twice) doesn't feel right but works...

CurrentTask::

class CurrentTask implements Runnable {
    private Priority priority
    private MyQueue queue

    public int getPriority() {
        return priority.value
    }

    public CurrentTask(Priority priority,ReportsQueue queue){
        this.priority = priority
        this.queue=queue
    }

    @Override
    public void run() {
...
}
}

Priority:

public enum Priority {

    HIGHEST(0),
    HIGH(1),
    MEDIUM(2),
    LOW(3),
    LOWEST(4)

    int value

    Priority(int val) {
        this.value = val
    }

    public int getValue(){
        return value
    }
}

Then your executor call

public YourExecutor() {

    public YourExecutor() {
        super(maxPoolSize,maxPoolSize,timeout,TimeUnit.SECONDS, new PriorityBlockingQueue<Runnable>(1000,new ReverseComparator()))
    }

So before changing to new method submit hit the comparator below and being TaskExecutor would not understand .priority?.value which as default .execute currentTask was what hit this and it all worked

public int compare(final Runnable lhs, final Runnable rhs) {

    if(lhs instanceof Runnable && rhs instanceof Runnable){
      // Favour a higher priority
        println "${lhs} vs ${lhs.getClass()}"
      if(((Runnable)lhs)?.priority?.value<((Runnable)rhs)?.priority?.value){
 ...
}

}

so with above hack and below changes it appears to be working

class  ReverseComparator implements Comparator<ComparableFutureTask>{

  @Override
  public int compare(final ComparableFutureTask lhs, final ComparableFutureTask rhs) {

    if(lhs instanceof ComparableFutureTask && rhs instanceof ComparableFutureTask){

        // run higher priority (lower numbers before higher numbers)
        println "${lhs} vs ${lhs.getClass()} ::: ${lhs.priority}"
      if(((Runnable)lhs)?.priority<((Runnable)rhs)?.priority){
          println "-returning -1"
        return -1;
      } else if (((Runnable)lhs)?.priority>((Runnable)rhs)?.priority){
      println "-returning @@@1"
        return 1;
      } 


    }
    println "-returning ==0 "
    return 0;
  }

Simply because we have passed in override ComparableFutureTask that has a priority which extends FutureTask

Hope it makes sense been going round for a day and a bit now :)

working implementation can be found here https://github.com/vahidhedayati/grails-queuekit-plugin — V H, Oct 21 '16 at 16:28

implementing PriorityQueue on ThreadPoolExecutor

2 Answers2

Linked