/*
 * 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.tdb2.lib;

import java.util.concurrent.* ;

import org.apache.jena.tdb2.TDBException;

/** Asynchronous operations support
 *  This class provides a simple framework for asynchronous operations.
 *  There is a thread pool and a pending queue. 
 *  <p>
 *  Default settings are thread pool of one and at most two pending operations.
 *  When a new async operation is added, the pending queue is checked, and the
 *  call blocks until the pending queue is below the threashold.
 *  <p>
 *  With the default setting of a thread pool of one, operations are
 *  executed in the order submitted.
 *  <p>
 *  If the far end is the bottleneck, a longer queeue is no help.
 */
public final class Async {
    // Currently specific to ThriftRunnable
    private static final int THREAD_POOL_SIZE = 1 ;
    private static final int PENDING_MAX = 2 ;
    private static final int BlockingQueueSize = PENDING_MAX+2 ;
    
    private final int pendingQueueLimit ;
    private final int blockingQueueSize ;
    private final ExecutorService executorService; 
    private final BlockingQueue<Future<Void>> outstanding ; 
    
    public Async() {
        this(THREAD_POOL_SIZE, PENDING_MAX) ;
    }

    public Async(int threadPoolSize, int pendingQueueLimit) {
        this.pendingQueueLimit = pendingQueueLimit ;
        this.blockingQueueSize = this.pendingQueueLimit+1 ;
        this.executorService = Executors.newFixedThreadPool(threadPoolSize) ;
        this.outstanding = new ArrayBlockingQueue<>(BlockingQueueSize) ;
    }
    
    /** Block until all pending operations has been completed */
    public void completeAsyncOperations() {
        reduceAsyncQueue(0) ;
    }
    
    /** Block until the pending operations queue is below the given size. */
    public void reduceAsyncQueue(int reduceSize) {
        System.out.println("Reduce: "+outstanding.size()) ;
        while ( outstanding.size() > reduceSize ) {
            try { outstanding.take().get() ; }
            catch (Exception ex) {
                throw new TDBException("Exception taking from async queue", ex) ;
            } 
        }
    }
    
    public void execAsync(Object lock, Runnable action) {
        reduceAsyncQueue(pendingQueueLimit) ;
        Future<Void> task = executorService.submit(()-> {
            try { action.run() ; } 
            catch (Exception ex)    { throw new TDBException("Unexpected exception: "+ex.getMessage(), ex) ; }
            return null ;
        }) ;
        outstanding.add(task) ;
    }
}
