/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.jena.fuseki.async;

import static java.lang.String.format ;

import java.util.* ;
import java.util.concurrent.* ;

import org.apache.jena.fuseki.Fuseki ;
import org.apache.jena.fuseki.server.DataService ;

/** The set of currently active async tasks */
public class AsyncPool
{
    private static int nMaxThreads = 4 ;
    private static int MAX_FINISHED = 20 ;
    
    // See Executors.newCachedThreadPool and Executors.newFixedThreadPool 
    private ExecutorService executor = new ThreadPoolExecutor(0, nMaxThreads,
                                                              120L, TimeUnit.SECONDS,
                                                              new LinkedBlockingQueue<Runnable>()) ;
    private final Object mutex = new Object() ; 
    private long counter = 0 ;
    private Map<String, AsyncTask> runningTasks = new LinkedHashMap<>() ; 
    private Map<String, AsyncTask> finishedTasks = new LinkedHashMap<>() ;
    
    private static AsyncPool instance = new AsyncPool() ;
    public static AsyncPool get() 
    { return instance ; }

    private AsyncPool() { }
    
    public AsyncTask submit(Runnable task, String displayName, DataService dataService, long requestId) { 
        synchronized(mutex) {
            String taskId = Long.toString(++counter) ;
            Fuseki.serverLog.info(format("Task : %s : %s",taskId, displayName)) ;
            Callable<Object> c = ()->{
                try { task.run(); } 
                catch (Throwable th) {
                    Fuseki.serverLog.error(format("Exception in task %s execution", taskId), th);
                }
                return null; 
            };
            AsyncTask asyncTask = new AsyncTask(c, this, taskId, displayName, dataService, requestId) ;
            /* Future<Object> future = */ executor.submit(asyncTask);
            runningTasks.put(taskId, asyncTask) ;
            return asyncTask ;
        }
    }
    
    public Collection<AsyncTask> tasks() {
        synchronized(mutex) {
            List<AsyncTask> x = new ArrayList<>(runningTasks.size()+finishedTasks.size()) ;
            x.addAll(runningTasks.values()) ;
            x.addAll(finishedTasks.values()) ;
            return x ;
        }
    }
    
    public void finished(AsyncTask task) { 
        synchronized(mutex) {
            String id = task.getTaskId() ;
            runningTasks.remove(id) ;
            while ( finishedTasks.size() >= MAX_FINISHED )
                finishedTasks.remove(task.getTaskId()) ;
            finishedTasks.put(id, task) ;
        }
    }

    public AsyncTask getRunningTask(String taskId) {
        synchronized(mutex) {
            return runningTasks.get(taskId) ;
        }
    }

    /** Get for any state */
    public AsyncTask getTask(String taskId) {
        synchronized(mutex) {
            AsyncTask task = runningTasks.get(taskId) ;
            if ( task != null )
                return task ;
            return finishedTasks.get(taskId) ;
        }
    }
}


@@ -1063,18 +1063,36 @@
 ve a
-sync
+nd recently completed
  tasks
+.
  */%0A
@@ -1116,16 +1116,45 @@
 cPool%0A%7B%0A
+    // Max concurrent tasks.%0A
     priv
@@ -1186,16 +1186,54 @@
 s = 4 ;%0A
+    // Number of finished tasks kept.%0A
     priv
@@ -1761,25 +1761,24 @@
 ashMap%3C%3E() ;
- 
 %0A    private
@@ -1837,24 +1837,137 @@
 shMap%3C%3E() ;%0A
+    // Finite FIFO of finished tasks. %0A    private LinkedList%3CAsyncTask%3E finishedTasksList = new LinkedList%3C%3E();%0A
     %0A    pri
@@ -2013,24 +2013,25 @@
 yncPool() ;%0A
+%0A
     public s
@@ -2052,21 +2052,16 @@
 l get() 
-%0A    
 %7B return
@@ -3387,24 +3387,61 @@
 emove(id) ;%0A
+            // Reduce old tasks list%0A
             
@@ -3457,24 +3457,28 @@
 inishedTasks
+List
 .size() %3E= M
@@ -3490,16 +3490,18 @@
 NISHED )
+ %7B
 %0A       
@@ -3500,32 +3500,52 @@
 %0A               
+ AsyncTask oldTask =
  finishedTasks.r
@@ -3546,17 +3546,69 @@
 asks
+List
 .remove
-(t
+First();%0A                finishedTasks.remove(oldT
 ask.
@@ -3619,18 +3619,31 @@
 askId())
- 
 ;
+%0A            %7D
 %0A       
@@ -3673,24 +3673,65 @@
 id, task) ;%0A
+            finishedTasksList.add(task);%0A
         %7D%0A  
@@ -3892,26 +3892,183 @@
  /**
- Get for any state
+%0A     * Get a task - whether running or finished. Return null for no such task and also%0A     * retired tasks. Only a limited number of old, finished tasks are remembered.%0A    
  */%0A

Merge commit 'refs/pull/348/head' of github.com:apache/jena