ojluni/src/main/java/sun/nio/ch/KQueuePort.java - platform/libcore.git - Git at Google

 /*
  * Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.
  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
  * This code is free software; you can redistribute it and/or modify it
  * under the terms of the GNU General Public License version 2 only, as
  * published by the Free Software Foundation.  Oracle designates this
  * particular file as subject to the "Classpath" exception as provided
  * by Oracle in the LICENSE file that accompanied this code.
  *
  * This code is distributed in the hope that it will be useful, but WITHOUT
  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  * version 2 for more details (a copy is included in the LICENSE file that
  * accompanied this code).
  *
  * You should have received a copy of the GNU General Public License version
  * 2 along with this work; if not, write to the Free Software Foundation,
  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
  *
  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
  * or visit www.oracle.com if you need additional information or have any
  * questions.
  */

 package sun.nio.ch;

 import java.nio.channels.spi.AsynchronousChannelProvider;
 import java.io.IOException;
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.atomic.AtomicInteger;
 import static sun.nio.ch.KQueue.*;

 /**
  * AsynchronousChannelGroup implementation based on the BSD kqueue facility.
  */

 final class KQueuePort
     extends Port
 {
     // maximum number of events to poll at a time
     private static final int MAX_KEVENTS_TO_POLL = 512;

     // kqueue file descriptor
     private final int kqfd;

     // true if kqueue closed
     private boolean closed;

     // socket pair used for wakeup
     private final int sp[];

     // number of wakeups pending
     private final AtomicInteger wakeupCount = new AtomicInteger();

     // address of the poll array passed to kqueue_wait
     private final long address;

     // encapsulates an event for a channel
     static class Event {
         final PollableChannel channel;
         final int events;

         Event(PollableChannel channel, int events) {
             this.channel = channel;
             this.events = events;
         }

         PollableChannel channel()   { return channel; }
         int events()                { return events; }
     }

     // queue of events for cases that a polling thread dequeues more than one
     // event
     private final ArrayBlockingQueue<Event> queue;
     private final Event NEED_TO_POLL = new Event(null, 0);
     private final Event EXECUTE_TASK_OR_SHUTDOWN = new Event(null, 0);

     KQueuePort(AsynchronousChannelProvider provider, ThreadPool pool)
         throws IOException
     {
         super(provider, pool);

         // open kqueue
         this.kqfd = kqueue();

         // create socket pair for wakeup mechanism
         int[] sv = new int[2];
         try {
             socketpair(sv);

             // register one end with kqueue
             keventRegister(kqfd, sv[0], EVFILT_READ, EV_ADD);
         } catch (IOException x) {
             close0(kqfd);
             throw x;
         }
         this.sp = sv;

         // allocate the poll array
         this.address = allocatePollArray(MAX_KEVENTS_TO_POLL);

         // create the queue and offer the special event to ensure that the first
         // threads polls
         this.queue = new ArrayBlockingQueue<Event>(MAX_KEVENTS_TO_POLL);
         this.queue.offer(NEED_TO_POLL);
     }

     KQueuePort start() {
         startThreads(new EventHandlerTask());
         return this;
     }

     /**
      * Release all resources
      */
     private void implClose() {
         synchronized (this) {
             if (closed)
                 return;
             closed = true;
         }
         freePollArray(address);
         close0(sp[0]);
         close0(sp[1]);
         close0(kqfd);
     }

     private void wakeup() {
         if (wakeupCount.incrementAndGet() == 1) {
             // write byte to socketpair to force wakeup
             try {
                 interrupt(sp[1]);
             } catch (IOException x) {
                 throw new AssertionError(x);
             }
         }
     }

     @Override
     void executeOnHandlerTask(Runnable task) {
         synchronized (this) {
             if (closed)
                 throw new RejectedExecutionException();
             offerTask(task);
             wakeup();
         }
     }

     @Override
     void shutdownHandlerTasks() {
         /*
          * If no tasks are running then just release resources; otherwise
          * write to the one end of the socketpair to wakeup any polling threads.
          */
         int nThreads = threadCount();
         if (nThreads == 0) {
             implClose();
         } else {
             // send interrupt to each thread
             while (nThreads-- > 0) {
                 wakeup();
             }
         }
     }

     // invoked by clients to register a file descriptor
     @Override
     void startPoll(int fd, int events) {
         // We use a separate filter for read and write events.
         // TBD: Measure cost of EV_ONESHOT vs. EV_CLEAR, either will do here.
         int err = 0;
         int flags = (EV_ADD|EV_ONESHOT);
         if ((events & Port.POLLIN) > 0)
             err = keventRegister(kqfd, fd, EVFILT_READ, flags);
         if (err == 0 && (events & Port.POLLOUT) > 0)
             err = keventRegister(kqfd, fd, EVFILT_WRITE, flags);
         if (err != 0)
             throw new InternalError("kevent failed: " + err);  // should not happen
     }

     /*
      * Task to process events from kqueue and dispatch to the channel's
      * onEvent handler.
      *
      * Events are retreived from kqueue in batch and offered to a BlockingQueue
      * where they are consumed by handler threads. A special "NEED_TO_POLL"
      * event is used to signal one consumer to re-poll when all events have
      * been consumed.
      */
     private class EventHandlerTask implements Runnable {
         private Event poll() throws IOException {
             try {
                 for (;;) {
                     int n = keventPoll(kqfd, address, MAX_KEVENTS_TO_POLL);
                     /*
                      * 'n' events have been read. Here we map them to their
                      * corresponding channel in batch and queue n-1 so that
                      * they can be handled by other handler threads. The last
                      * event is handled by this thread (and so is not queued).
                      */
                     fdToChannelLock.readLock().lock();
                     try {
                         while (n-- > 0) {
                             long keventAddress = getEvent(address, n);
                             int fd = getDescriptor(keventAddress);

                             // wakeup
                             if (fd == sp[0]) {
                                 if (wakeupCount.decrementAndGet() == 0) {
                                     // no more wakeups so drain pipe
                                     drain1(sp[0]);
                                 }

                                 // queue special event if there are more events
                                 // to handle.
                                 if (n > 0) {
                                     queue.offer(EXECUTE_TASK_OR_SHUTDOWN);
                                     continue;
                                 }
                                 return EXECUTE_TASK_OR_SHUTDOWN;
                             }

                             PollableChannel channel = fdToChannel.get(fd);
                             if (channel != null) {
                                 int filter = getFilter(keventAddress);
                                 int events = 0;
                                 if (filter == EVFILT_READ)
                                     events = Port.POLLIN;
                                 else if (filter == EVFILT_WRITE)
                                     events = Port.POLLOUT;

                                 Event ev = new Event(channel, events);

                                 // n-1 events are queued; This thread handles
                                 // the last one except for the wakeup
                                 if (n > 0) {
                                     queue.offer(ev);
                                 } else {
                                     return ev;
                                 }
                             }
                         }
                     } finally {
                         fdToChannelLock.readLock().unlock();
                     }
                 }
             } finally {
                 // to ensure that some thread will poll when all events have
                 // been consumed
                 queue.offer(NEED_TO_POLL);
             }
         }

         public void run() {
             Invoker.GroupAndInvokeCount myGroupAndInvokeCount =
                 Invoker.getGroupAndInvokeCount();
             final boolean isPooledThread = (myGroupAndInvokeCount != null);
             boolean replaceMe = false;
             Event ev;
             try {
                 for (;;) {
                     // reset invoke count
                     if (isPooledThread)
                         myGroupAndInvokeCount.resetInvokeCount();

                     try {
                         replaceMe = false;
                         ev = queue.take();

                         // no events and this thread has been "selected" to
                         // poll for more.
                         if (ev == NEED_TO_POLL) {
                             try {
                                 ev = poll();
                             } catch (IOException x) {
                                 x.printStackTrace();
                                 return;
                             }
                         }
                     } catch (InterruptedException x) {
                         continue;
                     }

                     // handle wakeup to execute task or shutdown
                     if (ev == EXECUTE_TASK_OR_SHUTDOWN) {
                         Runnable task = pollTask();
                         if (task == null) {
                             // shutdown request
                             return;
                         }
                         // run task (may throw error/exception)
                         replaceMe = true;
                         task.run();
                         continue;
                     }

                     // process event
                     try {
                         ev.channel().onEvent(ev.events(), isPooledThread);
                     } catch (Error x) {
                         replaceMe = true; throw x;
                     } catch (RuntimeException x) {
                         replaceMe = true; throw x;
                     }
                 }
             } finally {
                 // last handler to exit when shutdown releases resources
                 int remaining = threadExit(this, replaceMe);
                 if (remaining == 0 && isShutdown()) {
                     implClose();
                 }
             }
         }
     }

     // -- Native methods --

     private static native void socketpair(int[] sv) throws IOException;

     private static native void interrupt(int fd) throws IOException;

     private static native void drain1(int fd) throws IOException;

     private static native void close0(int fd);

     static {
         Util.load();
     }
 }
	/*
	* Copyright (c) 2012, Oracle and/or its affiliates. All rights reserved.
	* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
	*
	* This code is free software; you can redistribute it and/or modify it
	* under the terms of the GNU General Public License version 2 only, as
	* published by the Free Software Foundation. Oracle designates this
	* particular file as subject to the "Classpath" exception as provided
	* by Oracle in the LICENSE file that accompanied this code.
	*
	* This code is distributed in the hope that it will be useful, but WITHOUT
	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
	* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
	* version 2 for more details (a copy is included in the LICENSE file that
	* accompanied this code).
	*
	* You should have received a copy of the GNU General Public License version
	* 2 along with this work; if not, write to the Free Software Foundation,
	* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
	*
	* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
	* or visit www.oracle.com if you need additional information or have any
	* questions.
	*/

	package sun.nio.ch;

	import java.nio.channels.spi.AsynchronousChannelProvider;
	import java.io.IOException;
	import java.util.concurrent.ArrayBlockingQueue;
	import java.util.concurrent.RejectedExecutionException;
	import java.util.concurrent.atomic.AtomicInteger;
	import static sun.nio.ch.KQueue.*;

	/**
	* AsynchronousChannelGroup implementation based on the BSD kqueue facility.
	*/

	final class KQueuePort
	extends Port
	{
	// maximum number of events to poll at a time
	private static final int MAX_KEVENTS_TO_POLL = 512;

	// kqueue file descriptor
	private final int kqfd;

	// true if kqueue closed
	private boolean closed;

	// socket pair used for wakeup
	private final int sp[];

	// number of wakeups pending
	private final AtomicInteger wakeupCount = new AtomicInteger();

	// address of the poll array passed to kqueue_wait
	private final long address;

	// encapsulates an event for a channel
	static class Event {
	final PollableChannel channel;
	final int events;

	Event(PollableChannel channel, int events) {
	this.channel = channel;
	this.events = events;
	}

	PollableChannel channel() { return channel; }
	int events() { return events; }
	}

	// queue of events for cases that a polling thread dequeues more than one
	// event
	private final ArrayBlockingQueue<Event> queue;
	private final Event NEED_TO_POLL = new Event(null, 0);
	private final Event EXECUTE_TASK_OR_SHUTDOWN = new Event(null, 0);

	KQueuePort(AsynchronousChannelProvider provider, ThreadPool pool)
	throws IOException
	{
	super(provider, pool);

	// open kqueue
	this.kqfd = kqueue();

	// create socket pair for wakeup mechanism
	int[] sv = new int[2];
	try {
	socketpair(sv);

	// register one end with kqueue
	keventRegister(kqfd, sv[0], EVFILT_READ, EV_ADD);
	} catch (IOException x) {
	close0(kqfd);
	throw x;
	}
	this.sp = sv;

	// allocate the poll array
	this.address = allocatePollArray(MAX_KEVENTS_TO_POLL);

	// create the queue and offer the special event to ensure that the first
	// threads polls
	this.queue = new ArrayBlockingQueue<Event>(MAX_KEVENTS_TO_POLL);
	this.queue.offer(NEED_TO_POLL);
	}

	KQueuePort start() {
	startThreads(new EventHandlerTask());
	return this;
	}

	/**
	* Release all resources
	*/
	private void implClose() {
	synchronized (this) {
	if (closed)
	return;
	closed = true;
	}
	freePollArray(address);
	close0(sp[0]);
	close0(sp[1]);
	close0(kqfd);
	}

	private void wakeup() {
	if (wakeupCount.incrementAndGet() == 1) {
	// write byte to socketpair to force wakeup
	try {
	interrupt(sp[1]);
	} catch (IOException x) {
	throw new AssertionError(x);
	}
	}
	}

	@Override
	void executeOnHandlerTask(Runnable task) {
	synchronized (this) {
	if (closed)
	throw new RejectedExecutionException();
	offerTask(task);
	wakeup();
	}
	}

	@Override
	void shutdownHandlerTasks() {
	/*
	* If no tasks are running then just release resources; otherwise
	* write to the one end of the socketpair to wakeup any polling threads.
	*/
	int nThreads = threadCount();
	if (nThreads == 0) {
	implClose();
	} else {
	// send interrupt to each thread
	while (nThreads-- > 0) {
	wakeup();
	}
	}
	}

	// invoked by clients to register a file descriptor
	@Override
	void startPoll(int fd, int events) {
	// We use a separate filter for read and write events.
	// TBD: Measure cost of EV_ONESHOT vs. EV_CLEAR, either will do here.
	int err = 0;
	int flags = (EV_ADD\|EV_ONESHOT);
	if ((events & Port.POLLIN) > 0)
	err = keventRegister(kqfd, fd, EVFILT_READ, flags);
	if (err == 0 && (events & Port.POLLOUT) > 0)
	err = keventRegister(kqfd, fd, EVFILT_WRITE, flags);
	if (err != 0)
	throw new InternalError("kevent failed: " + err); // should not happen
	}

	/*
	* Task to process events from kqueue and dispatch to the channel's
	* onEvent handler.
	*
	* Events are retreived from kqueue in batch and offered to a BlockingQueue
	* where they are consumed by handler threads. A special "NEED_TO_POLL"
	* event is used to signal one consumer to re-poll when all events have
	* been consumed.
	*/
	private class EventHandlerTask implements Runnable {
	private Event poll() throws IOException {
	try {
	for (;;) {
	int n = keventPoll(kqfd, address, MAX_KEVENTS_TO_POLL);
	/*
	* 'n' events have been read. Here we map them to their
	* corresponding channel in batch and queue n-1 so that
	* they can be handled by other handler threads. The last
	* event is handled by this thread (and so is not queued).
	*/
	fdToChannelLock.readLock().lock();
	try {
	while (n-- > 0) {
	long keventAddress = getEvent(address, n);
	int fd = getDescriptor(keventAddress);

	// wakeup
	if (fd == sp[0]) {
	if (wakeupCount.decrementAndGet() == 0) {
	// no more wakeups so drain pipe
	drain1(sp[0]);
	}

	// queue special event if there are more events
	// to handle.
	if (n > 0) {
	queue.offer(EXECUTE_TASK_OR_SHUTDOWN);
	continue;
	}
	return EXECUTE_TASK_OR_SHUTDOWN;
	}

	PollableChannel channel = fdToChannel.get(fd);
	if (channel != null) {
	int filter = getFilter(keventAddress);
	int events = 0;
	if (filter == EVFILT_READ)
	events = Port.POLLIN;
	else if (filter == EVFILT_WRITE)
	events = Port.POLLOUT;

	Event ev = new Event(channel, events);

	// n-1 events are queued; This thread handles
	// the last one except for the wakeup
	if (n > 0) {
	queue.offer(ev);
	} else {
	return ev;
	}
	}
	}
	} finally {
	fdToChannelLock.readLock().unlock();
	}
	}
	} finally {
	// to ensure that some thread will poll when all events have
	// been consumed
	queue.offer(NEED_TO_POLL);
	}
	}

	public void run() {
	Invoker.GroupAndInvokeCount myGroupAndInvokeCount =
	Invoker.getGroupAndInvokeCount();
	final boolean isPooledThread = (myGroupAndInvokeCount != null);
	boolean replaceMe = false;
	Event ev;
	try {
	for (;;) {
	// reset invoke count
	if (isPooledThread)
	myGroupAndInvokeCount.resetInvokeCount();

	try {
	replaceMe = false;
	ev = queue.take();

	// no events and this thread has been "selected" to
	// poll for more.
	if (ev == NEED_TO_POLL) {
	try {
	ev = poll();
	} catch (IOException x) {
	x.printStackTrace();
	return;
	}
	}
	} catch (InterruptedException x) {
	continue;
	}

	// handle wakeup to execute task or shutdown
	if (ev == EXECUTE_TASK_OR_SHUTDOWN) {
	Runnable task = pollTask();
	if (task == null) {
	// shutdown request
	return;
	}
	// run task (may throw error/exception)
	replaceMe = true;
	task.run();
	continue;
	}

	// process event
	try {
	ev.channel().onEvent(ev.events(), isPooledThread);
	} catch (Error x) {
	replaceMe = true; throw x;
	} catch (RuntimeException x) {
	replaceMe = true; throw x;
	}
	}
	} finally {
	// last handler to exit when shutdown releases resources
	int remaining = threadExit(this, replaceMe);
	if (remaining == 0 && isShutdown()) {
	implClose();
	}
	}
	}
	}

	// -- Native methods --

	private static native void socketpair(int[] sv) throws IOException;

	private static native void interrupt(int fd) throws IOException;

	private static native void drain1(int fd) throws IOException;

	private static native void close0(int fd);

	static {
	Util.load();
	}
	}