ExecutorService that interrupts tasks after a timeout

Question

I m looking for an ExecutorService implementation that can be provided with a timeout  Tasks that are submitted to the ExecutorService are interrupted if they take longer than the timeout to run  Implementing such a beast isn t such a difficult task  but I m wondering if anybody knows of an existing implementation   Here s what I came up with based on some of the discussion below  Any comments   import java util List  import java util concurrent     public class TimeoutThreadPoolExecutor extends ThreadPoolExecutor       private final long timeout      private final TimeUnit timeoutUnit       private final ScheduledExecutorService timeoutExecutor   Executors newSingleThreadScheduledExecutor        private final ConcurrentMap lt Runnable  ScheduledFuture gt  runningTasks   new ConcurrentHashMap lt Runnable  ScheduledFuture gt          public TimeoutThreadPoolExecutor int corePoolSize  int maximumPoolSize  long keepAliveTime  TimeUnit unit  BlockingQueue lt Runnable gt  workQueue  long timeout  TimeUnit timeoutUnit            super corePoolSize  maximumPoolSize  keepAliveTime  unit  workQueue           this timeout   timeout          this timeoutUnit   timeoutUnit             public TimeoutThreadPoolExecutor int corePoolSize  int maximumPoolSize  long keepAliveTime  TimeUnit unit  BlockingQueue lt Runnable gt  workQueue  ThreadFactory threadFactory  long timeout  TimeUnit timeoutUnit            super corePoolSize  maximumPoolSize  keepAliveTime  unit  workQueue  threadFactory           this timeout   timeout          this timeoutUnit   timeoutUnit             public TimeoutThreadPoolExecutor int corePoolSize  int maximumPoolSize  long keepAliveTime  TimeUnit unit  BlockingQueue lt Runnable gt  workQueue  RejectedExecutionHandler handler  long timeout  TimeUnit timeoutUnit            super corePoolSize  maximumPoolSize  keepAliveTime  unit  workQueue  handler           this timeout   timeout          this timeoutUnit   timeoutUnit             public TimeoutThreadPoolExecutor int corePoolSize  int maximumPoolSize  long keepAliveTime  TimeUnit unit  BlockingQueue lt Runnable gt  workQueue  ThreadFactory threadFactory  RejectedExecutionHandler handler  long timeout  TimeUnit timeoutUnit            super corePoolSize  maximumPoolSize  keepAliveTime  unit  workQueue  threadFactory  handler           this timeout   timeout          this timeoutUnit   timeoutUnit              Override     public void shutdown             timeoutExecutor shutdown            super shutdown                Override     public List lt Runnable gt  shutdownNow             timeoutExecutor shutdownNow            return super shutdownNow                Override     protected void beforeExecute Thread t  Runnable r            if timeout  gt  0                final ScheduledFuture lt   gt  scheduled   timeoutExecutor schedule new TimeoutTask t   timeout  timeoutUnit               runningTasks put r  scheduled                         Override     protected void afterExecute Runnable r  Throwable t            ScheduledFuture timeoutTask   runningTasks remove r           if timeoutTask    null                timeoutTask cancel false                        class TimeoutTask implements Runnable           private final Thread thread           public TimeoutTask Thread thread                this thread   thread                      Override         public void run                 thread interrupt

User · Answer

Unfortunately the solution is flawed  There is a sort of bug with ScheduledThreadPoolExecutor  also reported in this question  cancelling a submitted task does not fully release the memory resources associated with the task  the resources are released only when the task expires   If you therefore create a TimeoutThreadPoolExecutor with a fairly long expiration time  a typical usage   and submit tasks fast enough  you end up filling the memory - even though the tasks actually completed successfully   You can see the problem with the following  very crude  test program   public static void main String   args  throws InterruptedException       ExecutorService service   new TimeoutThreadPoolExecutor 1  1  10  TimeUnit SECONDS               new LinkedBlockingQueue lt Runnable gt     10  TimeUnit MINUTES         ExecutorService service   Executors newFixedThreadPool 1       try           final AtomicInteger counter   new AtomicInteger            for  long i   0  i  lt  10000000  i                  service submit new Runnable                      Override                 public void run                         counter incrementAndGet                                                  if  i   10000    0                    System out println i         counter get                     while  i  gt  counter get                          Thread sleep 10                                                   finally           service shutdown              The program exhausts the available memory  although it waits for the spawned Runnables to complete   I though about this for a while  but unfortunately I could not come up with a good solution   EDIT  I found out this issue was reported as JDK bug 6602600  and appears to have been fixed very recently

User · Answer

How about using the ExecutorService shutDownNow   method as described in http   docs oracle com javase 7 docs api java util concurrent ExecutorService html  It seems to be the simplest solution

User · Answer

What about this alternative idea      two have two executors      one for     submitting the task  without caring about the timeout of the task adding the Future resulted and the time when it should end to an internal structure  one for executing an internal job which is checking the internal structure  if some tasks are timeout and if they have to be cancelled      Small sample is here     public class AlternativeExecutorService     private final CopyOnWriteArrayList lt ListenableFutureTask gt  futureQueue         new CopyOnWriteArrayList    private final ScheduledThreadPoolExecutor                scheduledExecutor   new ScheduledThreadPoolExecutor 1      used for internal cleaning job private final ListeningExecutorService                   threadExecutor      MoreExecutors listeningDecorator Executors newFixedThreadPool 5       used for private ScheduledFuture scheduledFuture  private static final long INTERNAL JOB CLEANUP FREQUENCY   1000L   public AlternativeExecutorService         scheduledFuture   scheduledExecutor scheduleAtFixedRate new TimeoutManagerJob    0  INTERNAL JOB CLEANUP FREQUENCY  TimeUnit MILLISECONDS      public void pushTask OwnTask task        ListenableFuture lt Void gt  future   threadExecutor submit task       - gt  create your Callable     futureQueue add new ListenableFutureTask future  task  getCurrentMillisecondsTime         - gt  store the time when the task should end    public void shutdownInternalScheduledExecutor         scheduledFuture cancel true       scheduledExecutor shutdownNow       long getCurrentMillisecondsTime         return Calendar getInstance   get Calendar MILLISECOND      class ListenableFutureTask       private final ListenableFuture lt Void gt  future      private final OwnTask                task      private final long                   milliSecEndTime       private ListenableFutureTask ListenableFuture lt Void gt  future  OwnTask task  long milliSecStartTime                this future   future          this task   task          this milliSecEndTime   milliSecStartTime   task getTimeUnit   convert task getTimeoutDuration    TimeUnit MILLISECONDS              ListenableFuture lt Void gt  getFuture                 return future             OwnTask getTask                 return task             long getMilliSecEndTime                 return milliSecEndTime           class TimeoutManagerJob implements Runnable       CopyOnWriteArrayList lt ListenableFutureTask gt  getCopyOnWriteArrayList                 return futureQueue              Override     public void run                 long currentMileSecValue   getCurrentMillisecondsTime            for  ListenableFutureTask futureTask   futureQueue                        consumeFuture futureTask  currentMileSecValue                        private void consumeFuture ListenableFutureTask futureTask  long currentMileSecValue                ListenableFuture lt Void gt  future   futureTask getFuture            boolean isTimeout   futureTask getMilliSecEndTime    gt   currentMileSecValue          if  isTimeout                        if   future isDone                                  future cancel true                             futureQueue remove futureTask                      class OwnTask implements Callable lt Void gt        private long     timeoutDuration      private TimeUnit timeUnit       OwnTask long timeoutDuration  TimeUnit timeUnit                this timeoutDuration   timeoutDuration          this timeUnit   timeUnit              Override     public Void call   throws Exception                  do logic         return null             public long getTimeoutDuration                 return timeoutDuration             public TimeUnit getTimeUnit                 return timeUnit

User · Answer

You can use a ScheduledExecutorService for this  First you would submit it only once to begin immediately and retain the future that is created  After that you can submit a new task that would cancel the retained future after some period of time    ScheduledExecutorService executor   Executors newScheduledThreadPool 2     final Future handler   executor submit new Callable             executor schedule new Runnable         public void run             handler cancel                     10000  TimeUnit MILLISECONDS     This will execute your handler  main functionality to be interrupted  for 10 seconds  then will cancel  i e  interrupt  that specific task

User · Answer

check if this works for you       public  lt T S K V gt  ResponseObject lt Collection lt ResponseObject lt T gt  gt  gt  runOnScheduler ThreadPoolExecutor threadPoolExecutor        int parallelismLevel  TimeUnit timeUnit  int timeToCompleteEachTask  Collection lt S gt  collection        Map lt K V gt  context  Task lt T S K V gt  someTask       if threadPoolExecutor  null         return ResponseObject  lt Collection lt ResponseObject lt T gt  gt  gt builder   errorCode  500   errorMessage  threadPoolExecutor can not be null   build              if someTask  null         return ResponseObject  lt Collection lt ResponseObject lt T gt  gt  gt builder   errorCode  500   errorMessage  Task can not be null   build              if CollectionUtils isEmpty collection          return ResponseObject  lt Collection lt ResponseObject lt T gt  gt  gt builder   errorCode  500   errorMessage  input collection can not be empty   build               LinkedBlockingQueue lt Callable lt T gt  gt  callableLinkedBlockingQueue   new LinkedBlockingQueue lt  gt  collection size         collection forEach value - gt          callableLinkedBlockingQueue offer   - gt someTask perform value context      pass some values in callable  which can be anything              LinkedBlockingQueue lt Future lt T gt  gt  futures   new LinkedBlockingQueue lt  gt          int count   0       while count lt parallelismLevel  amp  amp  count  lt  callableLinkedBlockingQueue size           Future lt T gt  f   threadPoolExecutor submit callableLinkedBlockingQueue poll           futures offer f         count               Collection lt ResponseObject lt T gt  gt  responseCollection   new ArrayList lt  gt          while futures size   gt 0         Future lt T gt  future   futures poll          ResponseObject lt T gt  responseObject   null          try             T response   future get timeToCompleteEachTask  timeUnit             responseObject   ResponseObject  lt T gt builder   data response  build              catch  InterruptedException e              future cancel true             catch  ExecutionException e              future cancel true             catch  TimeoutException e              future cancel true             finally             if  Objects nonNull responseObject                 responseCollection add responseObject                         futures remove future    remove this           Callable lt T gt  callable   getRemainingCallables callableLinkedBlockingQueue             if null  callable               Future lt T gt  f   threadPoolExecutor submit callable               futures add f                                    return ResponseObject  lt Collection lt ResponseObject lt T gt  gt  gt builder   data responseCollection  build           private  lt T gt  Callable lt T gt  getRemainingCallables LinkedBlockingQueue lt Callable lt T gt  gt  callableLinkedBlockingQueue       if callableLinkedBlockingQueue size   gt 0         return callableLinkedBlockingQueue poll              return null        you can restrict the no of thread uses from scheduler as well as put timeout on the task

User · Answer

After ton of time to survey  Finally  I use invokeAll method of ExecutorService to solve this problem  That will strictly interrupt the task while task running  Here is example    ExecutorService executorService   Executors newCachedThreadPool     try       List lt Callable lt Object gt  gt  callables   new ArrayList lt  gt            Add your long time task  callable      callables add new VaryLongTimeTask            Assign tasks for specific execution timeout  e g  2 sec      List lt Future lt Object gt  gt  futures   executorService invokeAll callables  2000  TimeUnit MILLISECONDS       for  Future lt Object gt  future   futures               Getting result         catch  InterruptedException e        e printStackTrace       executorService shutdown        The pro is you can also submit ListenableFuture at the same ExecutorService  Just slightly change the first line of code   ListeningExecutorService executorService   MoreExecutors listeningDecorator Executors newCachedThreadPool       ListeningExecutorService is the Listening feature of ExecutorService at google guava project  com google guava

User · Answer

Wrap the task in FutureTask and you can specify timeout for the FutureTask  Look at the example in my answer to this question   java native Process timeout

User · Answer

It seems problem is not in JDK bug 6602600   it was solved at 2010-05-22   but in  incorrect call of sleep 10  in circle  Addition note  that the main Thread must give directly CHANCE to other threads to realize thier tasks by invoke SLEEP 0  in  EVERY branch of outer circle  It is better  I think  to use Thread yield   instead of Thread sleep 0   The result corrected part of previous problem code is such like this                                                    Thread yield              if  i   1000   0    System out println i         counter get        service toString                                             while  i  gt  counter get                            Thread sleep 10                          It works correctly with amount of outer counter up to 150 000 000 tested circles

User · Answer

You can use this implementation that ExecutorService provides invokeAll Collection lt   extends Callable lt T gt  gt  tasks long timeout  TimeUnit unit  as  executor invokeAll Arrays asList task   2   TimeUnit SECONDS    However  in my case  I could not as Arrays asList took extra 20ms

User · Answer

Using John W answer I created an implementation that correctly begin the timeout when the task starts its execution  I even write a unit test for it     However  it does not suit my needs since some IO operations do not interrupt when Future cancel   is called  ie when Thread interrupt   is called   Some examples of IO operation that may not be interrupted when Thread interrupt   is called are Socket connect and Socket read  and I suspect most of IO operation implemented in java io   All IO operations in java nio should be interruptible when Thread interrupt   is called  For example  that is the case for SocketChannel open and SocketChannel read   Anyway if anyone is interested  I created a gist for a thread pool executor that allows tasks to timeout  if they are using interruptible operations      https   gist github com amanteaux 64c54a913c1ae34ad7b86db109cbc0bf

[java] ExecutorService that interrupts tasks after a timeout

Examples related to java

Examples related to multithreading

Examples related to concurrency

Examples related to executorservice