/*
    * 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.river.jeri.internal.runtime;
  
  import org.apache.river.logging.Levels;
  import org.apache.river.thread.Executor;
  import org.apache.river.thread.GetThreadPoolAction;
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.nio.channels.CancelledKeyException;
  import java.nio.channels.ClosedChannelException;
  import java.nio.channels.IllegalBlockingModeException;
  import java.nio.channels.Pipe;
  import java.nio.channels.SelectableChannel;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.security.AccessController;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Set;
  import java.util.WeakHashMap;
  import java.util.logging.Level;
  import java.util.logging.Logger;
  
  /**
   * A SelectionManager provides an event dispatching layer on top of the
   * java.nio.Selector and java.nio.SelectableChannel abstractions; it manages
   * one-shot registrations of interest in I/O readiness events and
   * dispatching notifications of such events to registered callback objects.
   *
   * SelectionManager is designed to support multiple select/dispatch threads
   * to allow for improved I/O concurrency on symmetric multiprocessor systems.
   *
   * If interest in a particular I/O event is (re-)registered while there is a
   * blocking select operation in progress, that select operation must be woken
   * up so that it can take the new interest into account; this wakeup is
   * achieved by writing a byte to an internal pipe that is registered with the
   * underlying selector.
   *
   * A current limitation of this API is that it does not allow an executing
   * callback to yield control in such a way that it will be scheduled to
   * execute again without another I/O readiness event occurring.  Therefore,
   * data that gets read during a callback must also get processed, unless such
   * processing depends on more data that has not yet been read (like the
   * remainder of a partial message).
   *
   * <p>This implementation uses the {@link Logger} named
   * <code>org.apache.river.jeri.internal.runtime.SelectionManager</code> to
   * log information at the following levels:
   *
   * <p><table summary="Describes what is logged by SelectionManager at
   * various logging levels" border=1 cellpadding=5>
   *
   * <tr> <th> Level <th> Description
   *
   * <tr> <td> {@link Levels#HANDLED HANDLED} <td> I/O exception caught
   * from select operation
   *
   * </table>
   *
   * @author Sun Microsystems, Inc.
   **/
  public final class SelectionManager {
  
      /** number of concurrent I/O processing threads */
      private static final int concurrency = 1;	// REMIND: get from property?
  
      private static final Logger logger = Logger.getLogger(
  	"org.apache.river.jeri.internal.runtime.SelectionManager");
  
      /** pool of threads for executing tasks in system thread group */
      private static final Executor systemThreadPool = (Executor)
  	AccessController.doPrivileged(new GetThreadPoolAction(false));
  
      /** shared Selector used by this SelectionManager */
      private final Selector selector;
  
      /** internal pipe used to wake up a blocked select operation */
      private final Pipe.SinkChannel wakeupPipeSink;
      private final Pipe.SourceChannel wakeupPipeSource;
      private final SelectionKey wakeupPipeKey;
      private final ByteBuffer wakeupBuffer = ByteBuffer.allocate(2);
 
     /** set of registered channels, to detect duplicate registrations */
     private final Map registeredChannels = 
 	Collections.synchronizedMap(new WeakHashMap());
 
     /**
      * lock guarding selectingThread, wakeupPending, renewQueue, readyQueue,
      * renewMaskRef, and mutable state of all Key instances.
      */
     private final Object lock = new Object();
 
     /** thread with exclusive right to perform a select operation, if any */
     private Thread selectingThread = null;
 
     /** true if a wakeup has been requested but not yet processed */
     private boolean wakeupPending = false;
 
     /*
      * The following two queues of Key objects are implemented as LIFO
      * linked lists with internally threaded links for fast addition and
      * removal of elements (no memory allocation overhead) and so that
      * multiple entries for the same channel get implicitly combined.
      * LIFO ordering is OK because the entire queue always gets drained
      * and processed at once, and in between such processing, addition
      * order is arbitrary anyway.
      */
 
     /** queue of keys that need to have interest masks updated */
     private Key renewQueue = null;
 
     /** queue of keys that have I/O operations ready to be handled */
     private Key readyQueue = null;
 
     /** holder used for pass-by-reference invocations */
     private final int[] renewMaskRef = new int[1];
 
     /**
      * Creates a new SelectionManager.
      *
      * REMIND: Is this necessary, or should we just provide access to
      * a singleton instance?
      */
     public SelectionManager() throws IOException {
 
 	// REMIND: create threads and other resources lazily?
 
 	selector = Selector.open();
 
 	Pipe pipe = Pipe.open();
 	wakeupPipeSink = pipe.sink();
 	wakeupPipeSource = pipe.source();
 	wakeupPipeSource.configureBlocking(false);
 	wakeupPipeKey = wakeupPipeSource.register(selector,
 						  SelectionKey.OP_READ);
 
 	for (int i = 0; i < concurrency; i++) {
 	    systemThreadPool.execute(new SelectLoop(),
 				     "I/O SelectionManager-" + i);
 	}
 
 	// REMIND: How do these threads and other resources get cleaned up?
 	// REMIND: Should there be an explicit close method?
     }
 
     /**
      * Registers the given SelectableChannel with this SelectionManager.
      * After registration, the returned Key's renewInterestMask method may
      * be used to register one-shot interest in particular I/O events.
      */
     public Key register(SelectableChannel channel, SelectionHandler handler) {
 	if (registeredChannels.containsKey(channel)) {
 	    throw new IllegalStateException("channel already registered");
 	}
 	Key key = new Key(channel, handler);
 	registeredChannels.put(channel, null);
 	return key;
     }
 
     /**
      * SelectionHandler is the callback interface for an object that will
      * process an I/O readiness event that has been detected by a
      * SelectionManager.
      */
     public interface SelectionHandler {
 	void handleSelection(int readyMask, Key key);
     }
 
     /**
      * A Key represents a given SelectableChannel's registration with this
      * SelectionManager.  Externally, this object is used to re-register
      * interest in I/O readiness events that have been previously detected
      * and dispatched.
      */
     public final class Key {
 
 	/** the channel that this Key represents a registration for */
 	final SelectableChannel channel;
 	/** the supplied callback object for dispatching I/O events */
 	final SelectionHandler handler;
 
 	// mutable instance state guarded by enclosing SelectionManager's lock:
 
 	/**
 	 * the SelectionKey representing this Key's registration with the
 	 * internal Selector, or null if it hasn't yet been registered
 	 */
 	SelectionKey selectionKey = null;
 
 	/** the current interest mask established with the SelectionKey */
 	int interestMask = 0;
 
 	boolean onRenewQueue = false;	// invariant: == (renewMask != 0)
 	Key renewQueueNext = null;	// null if !onRenewQueue
 	int renewMask = 0;
 
 	boolean onReadyQueue = false;	// invariant: == (readyMask != 0)
 	Key readyQueueNext = null;	// null if !onReadyQueue
 	int readyMask = 0;
 
 	/*
 	 * other invariants:
 	 *
 	 * (renewMask & interestMask) == 0
 	 * (interestMask & readyMask) == 0
 	 * (renewMask & readyMask) == 0
 	 */
 
 	Key(SelectableChannel channel, SelectionHandler handler) {
 	    this.channel = channel;
 	    this.handler = handler;
 	}
 
 	/**
 	 * Renews interest in receiving notifications when the I/O operations
 	 * identified by the specified mask are ready for the associated
 	 * SelectableChannel.  The specified mask identifies I/O operations
 	 * with the same bit values as would a java.nio.SelectionKey for the
 	 * same SelectableChannel.
 	 *
 	 * Some time after one of the operations specified in the mask is
 	 * detected to be ready, the previously-registered SelectionHandler
 	 * callback object will be invoked to handle the readiness event.
 	 *
 	 * An event for each operation specified will only be dispatched to the
 	 * callback handler once for the invocation of this method; to
 	 * re-register interest in subsequent readiness of the same operation
 	 * for the given channel, this method must be invoked again.
 	 */
 	public void renewInterestMask(int mask)
 	    throws ClosedChannelException
 	{
 	    if (!channel.isOpen()) {
 		throw new ClosedChannelException();
 	    }
 	    if ((mask & ~channel.validOps()) != 0) {
 		throw new IllegalArgumentException(
 		    "invalid mask " + mask +
 		    " (valid mask " + channel.validOps() + ")");
 	    }
 	    if (channel.isBlocking()) {
 		throw new IllegalBlockingModeException();
 	    }
 	    synchronized (lock) {
 		int delta = mask & ~(renewMask | interestMask | readyMask);
 		if (delta != 0) {
 		    addOrUpdateRenewQueue(this, delta);
 		    if (selectingThread != null && !wakeupPending) {
 			wakeupSelector();
 			wakeupPending = true;
 		    }
 		}
 	    }
 	}
     }
 
     /**
      * SelectLoop provides the main loop for each I/O processing thread.
      */
     private class SelectLoop implements Runnable {
 
 	private long lastExceptionTime = 0L;	// local to select thread
 	private int recentExceptionCount;	// local to select thread
 
 	public void run() {
 	    int[] readyMaskRef = new int[1];
 	    while (true) {
 		try {
 		    Key readyKey = waitForReadyKey(readyMaskRef);
 		    readyKey.handler.handleSelection(readyMaskRef[0],
 						     readyKey);
 		} catch (Throwable t) {
 		    try {
 			logger.log(Level.WARNING, "select loop throws", t);
 		    } catch (Throwable tt) {
 		    }
 		    throttleLoopOnException();
 		}
 	    }
 	}
 
 	/**
 	 * Throttles the select loop after an exception has been
 	 * caught: if a burst of 10 exceptions in 5 seconds occurs,
 	 * then wait for 10 seconds to curb busy CPU usage.
 	 **/
 	private void throttleLoopOnException() {
 	    long now = System.currentTimeMillis();
 	    if (lastExceptionTime == 0L || (now - lastExceptionTime) > 5000) {
 		// last exception was long ago (or this is the first)
 		lastExceptionTime = now;
 		recentExceptionCount = 0;
 	    } else {
 		// exception burst window was started recently
 		if (++recentExceptionCount >= 10) {
 		    try {
 			Thread.sleep(10000);
 		    } catch (InterruptedException ignore) {
 		    }
 		}
 	    }
 	}
     }
 
     /**
      * Waits until one of the registered channels is ready for one or more
      * I/O operations.  The Key for the ready channel is returned, and the
      * first element of the supplied array is set to the mask of the
      * channel's ready operations.
      *
      * If there is a ready channel available, then its key is returned.
      * If another thread is already performing a select operation, then the
      * current thread waits for that thread to complete and then begins
      * again.  Otherwise, the current thread assumes the responsibility of
      * performing the next select operation.
      */
     private Key waitForReadyKey(int[] readyMaskOut)
 	throws InterruptedException
     {
 	assert !Thread.holdsLock(lock);
 	assert readyMaskOut != null && readyMaskOut.length == 1;
 
 	boolean needToClearSelectingThread = false;
 	Set selectedKeys = selector.selectedKeys();
 
 	try {
 	    synchronized (lock) {
 		while (isReadyQueueEmpty() && selectingThread != null) {
 		    lock.wait();
 		}
 		if (!isReadyQueueEmpty()) {
 		    Key readyKey = removeFromReadyQueue(readyMaskOut);
 		    lock.notify();
 		    return readyKey;
 		}
 
 		assert selectingThread == null;
 		selectingThread = Thread.currentThread();
 		needToClearSelectingThread = true;
 
 		processRenewQueue();
 	    }						// wakeup allowed
 
 	    while (true) {
 		try {
 		    int n = selector.select();
 		    if (Thread.interrupted()) {
 			throw new InterruptedException();
 		    }
 		} catch (Error e) {
 		    String message = e.getMessage();
 		    if (message != null && message.startsWith("POLLNVAL")) {
 //			Thread.yield();
                         Thread.sleep(100L);
 			continue;		// work around 4458268
 		    } else {
 			throw e;
 		    }
 		} catch (CancelledKeyException e) {
 		    continue;			// work around 4458268
 		} catch (NullPointerException e) {
 		    continue;			// work around 4729342
 		} catch (IOException e) {
 		    logger.log(Levels.HANDLED,
 			       "thrown by select, continuing", e);
 		    continue;			// work around 4504001
 		}
 
 		synchronized (lock) {
 		    if (wakeupPending &&
 			selectedKeys.contains(wakeupPipeKey))
 		    {
 			drainWakeupPipe();		// clear wakeup state
 			wakeupPending = false;
 			selectedKeys.remove(wakeupPipeKey);
 		    }
 		    if (selectedKeys.isEmpty()) {
 			processRenewQueue();
 			continue;
 		    }
 
 		    selectingThread = null;
 		    needToClearSelectingThread = false;
 		    lock.notify();
 
 		    Iterator iter = selectedKeys.iterator();
 		    assert iter.hasNext();	// there must be at least one
 		    while (iter.hasNext()) {
 			SelectionKey selectionKey = (SelectionKey) iter.next();
 			Key key = (Key) selectionKey.attachment();
 
 			int readyMask = 0;
 			try {
 			    readyMask = selectionKey.readyOps();
 			    assert readyMask != 0;
 			    assert (key.interestMask & readyMask) == readyMask;
 
 			    /*
 			     * Remove interest in I/O events detected to be
 			     * ready; interest must be renewed after each
 			     * notification.
 			     */
 			    int newInterestMask =
 				key.interestMask & ~readyMask;
 			    assert key.interestMask ==
 				selectionKey.interestOps();
 			    key.selectionKey.interestOps(newInterestMask);
 			    key.interestMask = newInterestMask;
 			} catch (CancelledKeyException e) {
 			    /*
 			     * If channel is closed, then all interested events
 			     * become considered ready immediately.
 			     */
 			    readyMask |= key.interestMask;
 			    key.interestMask = 0;
 			}
 			addOrUpdateReadyQueue(key, readyMask);
 
 			iter.remove();
 		    }
 
 		    return removeFromReadyQueue(readyMaskOut);
 		}					// wakeup NOT allowed
 	    }
 	} finally {
 	    if (needToClearSelectingThread) {
 		synchronized (lock) {
 		    if (wakeupPending &&
 			selectedKeys.contains(wakeupPipeKey))
 		    {
 			drainWakeupPipe();		// clear wakeup state
 			wakeupPending = false;
 			selectedKeys.remove(wakeupPipeKey);
 		    }
 		    selectingThread = null;
 		    needToClearSelectingThread = false;
 		    lock.notify();
 		}					// wakeup NOT allowed
 	    }
 	}
     }
 
     private void wakeupSelector() {
 	assert Thread.holdsLock(lock);
 	assert wakeupPending == false;
 
 	wakeupBuffer.clear().limit(1);
 	try {
 	    wakeupPipeSink.write(wakeupBuffer);
 	} catch (IOException e) {
 	    // REMIND: what if thread was interrupted?
 	    throw new AssertionError("unexpected I/O exception", e);
 	}
     }
 
     private void drainWakeupPipe() {
 	assert Thread.holdsLock(lock);
 	assert selectingThread != null;
 
 	do {
 	    wakeupBuffer.clear();
 	    try {
 		wakeupPipeSource.read(wakeupBuffer);
 	    } catch (IOException e) {
 		// REMIND: what if thread was interrupted?
 		throw new AssertionError("unexpected I/O exception", e);
 	    }
 	} while (!wakeupBuffer.hasRemaining());
     }
 
     /**
      * In preparation for performing a select operation, process all new
      * and renewed interest registrations so that current SelectionKey
      * interest masks are up to date.
      *
      * This method must not be invoked while there is a select operation in
      * progress (because otherwise it could block indefinitely); therefore,
      * it must be invoked only by a thread that has the exclusive right to
      * perform a select operation.
      */
     private void processRenewQueue() {
 	assert Thread.holdsLock(lock);
 	assert selectingThread != null;
 
 	while (!isRenewQueueEmpty()) {
 	    Key key = removeFromRenewQueue(renewMaskRef);
 	    int renewMask = renewMaskRef[0];
 	    assert renewMask != 0;
 
 	    if (key.selectionKey == null) {
 		assert key.interestMask == 0 && key.readyMask == 0;
 
 		try {
 		    key.selectionKey = key.channel.register(selector,
 							    renewMask);
 		    key.selectionKey.attach(key);
 		    key.interestMask = renewMask;
 		} catch (ClosedChannelException e) {
 		    addOrUpdateReadyQueue(key, renewMask);
 		} catch (IllegalBlockingModeException e) {
 		    addOrUpdateReadyQueue(key, renewMask);
 		}
 	    } else {
 		assert (key.interestMask & renewMask) == 0;
 
 		int newInterestMask = key.interestMask | renewMask;
 		try {
 		    assert key.interestMask == key.selectionKey.interestOps();
 		    key.selectionKey.interestOps(newInterestMask);
 		    key.interestMask = newInterestMask;
 		} catch (CancelledKeyException e) {
 		    addOrUpdateReadyQueue(key, newInterestMask);
 		    key.interestMask = 0;
 		}
 
 		assert (key.interestMask & key.readyMask) == 0;
 	    }
 	}
     }
 
     /*
      * Queue manipulation utilities:
      */
 
     private boolean isRenewQueueEmpty() {
 	assert Thread.holdsLock(lock);
 	return renewQueue == null;
     }
 
     private Key removeFromRenewQueue(int[] renewMaskOut) {
 	assert renewMaskOut != null && renewMaskOut.length == 1;
 	assert Thread.holdsLock(lock);
 
 	Key key = renewQueue;
 	assert key != null;
 
 	assert key.onRenewQueue;
 	assert key.renewMask != 0;
 	renewMaskOut[0] = key.renewMask;
 	key.renewMask = 0;
 	renewQueue = key.renewQueueNext;
 	key.renewQueueNext = null;
 	key.onRenewQueue = false;
 	return key;
     }
 
     private void addOrUpdateRenewQueue(Key key, int newRenewMask) {
 	assert newRenewMask != 0;
 	assert Thread.holdsLock(lock);
 
 	if (!key.onRenewQueue) {
 	    assert key.renewMask == 0;
 	    assert key.renewQueueNext == null;
 	    key.renewMask = newRenewMask;
 	    key.renewQueueNext = renewQueue;
 	    renewQueue = key;
 	    key.onRenewQueue = true;
 	} else {
 	    assert key.renewMask != 0;
 	    assert (key.renewMask & newRenewMask) == 0;
 	    key.renewMask |= newRenewMask;
 	}
     }
 
     private boolean isReadyQueueEmpty() {
 	assert Thread.holdsLock(lock);
 	return readyQueue == null;
     }
 
     private Key removeFromReadyQueue(int[] readyMaskOut) {
 	assert readyMaskOut != null && readyMaskOut.length == 1;
 	assert Thread.holdsLock(lock);
 
 	Key key = readyQueue;
 	assert key != null;
 
 	assert key.onReadyQueue;
 	assert key.readyMask != 0;
 	readyMaskOut[0] = key.readyMask;
 	key.readyMask = 0;
 	readyQueue = key.readyQueueNext;
 	key.readyQueueNext = null;
 	key.onReadyQueue = false;
 	return key;
     }
 
     private void addOrUpdateReadyQueue(Key key, int newReadyMask) {
 	assert newReadyMask != 0;
 	assert Thread.holdsLock(lock);
 
 	if (!key.onReadyQueue) {
 	    assert key.readyMask == 0;
 	    assert key.readyQueueNext == null;
 	    key.readyMask = newReadyMask;
 	    key.readyQueueNext = readyQueue;
 	    readyQueue = key;
 	    key.onReadyQueue = true;
 	} else {
 	    assert key.readyMask != 0;
 	    assert (key.readyMask & newReadyMask) == 0;
 	    key.readyMask |= newReadyMask;
 	}
     }
 }