527 lines
17 KiB
Java
527 lines
17 KiB
Java
/* KqueueSelectorImpl.java -- Selector for systems with kqueue event notification.
|
|
Copyright (C) 2006 Free Software Foundation, Inc.
|
|
|
|
This file is part of GNU Classpath.
|
|
|
|
GNU Classpath is free software; you can redistribute it and/or modify
|
|
it under the terms of the GNU General Public License as published by
|
|
the Free Software Foundation; either version 2, or (at your option)
|
|
any later version.
|
|
|
|
GNU Classpath 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 for more details.
|
|
|
|
You should have received a copy of the GNU General Public License
|
|
along with GNU Classpath; see the file COPYING. If not, write to the
|
|
Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
|
|
02110-1301 USA.
|
|
|
|
Linking this library statically or dynamically with other modules is
|
|
making a combined work based on this library. Thus, the terms and
|
|
conditions of the GNU General Public License cover the whole
|
|
combination.
|
|
|
|
As a special exception, the copyright holders of this library give you
|
|
permission to link this library with independent modules to produce an
|
|
executable, regardless of the license terms of these independent
|
|
modules, and to copy and distribute the resulting executable under
|
|
terms of your choice, provided that you also meet, for each linked
|
|
independent module, the terms and conditions of the license of that
|
|
module. An independent module is a module which is not derived from
|
|
or based on this library. If you modify this library, you may extend
|
|
this exception to your version of the library, but you are not
|
|
obligated to do so. If you do not wish to do so, delete this
|
|
exception statement from your version. */
|
|
|
|
|
|
package gnu.java.nio;
|
|
|
|
|
|
import java.io.IOException;
|
|
import java.nio.ByteBuffer;
|
|
import java.nio.ByteOrder;
|
|
import java.nio.channels.ClosedSelectorException;
|
|
import java.nio.channels.SelectableChannel;
|
|
import java.nio.channels.SelectionKey;
|
|
import java.nio.channels.Selector;
|
|
import java.nio.channels.spi.AbstractSelectableChannel;
|
|
import java.nio.channels.spi.AbstractSelector;
|
|
import java.nio.channels.spi.SelectorProvider;
|
|
import java.util.HashMap;
|
|
import java.util.HashSet;
|
|
import java.util.Iterator;
|
|
import java.util.Map;
|
|
import java.util.Set;
|
|
|
|
/**
|
|
* A {@link Selector} implementation that uses the <code>kqueue</code>
|
|
* event notification facility.
|
|
*
|
|
* @author Casey Marshall (csm@gnu.org)
|
|
*/
|
|
public class KqueueSelectorImpl extends AbstractSelector
|
|
{
|
|
// Prepended underscore to field name to make it distinct
|
|
// from the method with the similar name.
|
|
private static final int _sizeof_struct_kevent;
|
|
|
|
private static final int MAX_DOUBLING_CAPACITY = 16384;
|
|
private static final int CAP_INCREMENT = 1024;
|
|
private static final int INITIAL_CAPACITY;
|
|
|
|
static
|
|
{
|
|
try
|
|
{
|
|
System.loadLibrary("javanio");
|
|
}
|
|
catch (Exception x)
|
|
{
|
|
x.printStackTrace();
|
|
}
|
|
|
|
if (kqueue_supported ())
|
|
_sizeof_struct_kevent = sizeof_struct_kevent();
|
|
else
|
|
_sizeof_struct_kevent = -1;
|
|
INITIAL_CAPACITY = 16 * _sizeof_struct_kevent;
|
|
}
|
|
|
|
/**
|
|
* Tell if kqueue-based selectors are supported on this system.
|
|
*
|
|
* @return True if this system has kqueue support, and support for it was
|
|
* compiled in to Classpath.
|
|
*/
|
|
public static native boolean kqueue_supported();
|
|
|
|
/* Our native file descriptor. */
|
|
private int kq;
|
|
|
|
private HashMap/*<Integer,KqueueSelectionKeyImpl>*/ keys;
|
|
private HashSet/*<KqueueSelectionKeyImpl>*/ selected;
|
|
private Thread blockedThread;
|
|
private ByteBuffer events;
|
|
|
|
private static final int OP_ACCEPT = SelectionKey.OP_ACCEPT;
|
|
private static final int OP_CONNECT = SelectionKey.OP_CONNECT;
|
|
private static final int OP_READ = SelectionKey.OP_READ;
|
|
private static final int OP_WRITE = SelectionKey.OP_WRITE;
|
|
|
|
public KqueueSelectorImpl(SelectorProvider provider) throws IOException
|
|
{
|
|
super(provider);
|
|
kq = implOpen();
|
|
keys = new HashMap/*<KqueueSelectionKeyImpl>*/();
|
|
events = ByteBuffer.allocateDirect(INITIAL_CAPACITY);
|
|
}
|
|
|
|
protected void implCloseSelector() throws IOException
|
|
{
|
|
implClose(kq);
|
|
kq = -1;
|
|
}
|
|
|
|
/* (non-Javadoc)
|
|
* @see java.nio.channels.Selector#keys()
|
|
*/
|
|
public Set keys()
|
|
{
|
|
if (!isOpen())
|
|
throw new ClosedSelectorException();
|
|
|
|
return new HashSet(keys.values());
|
|
}
|
|
|
|
/* (non-Javadoc)
|
|
* @see java.nio.channels.Selector#select()
|
|
*/
|
|
public int select() throws IOException
|
|
{
|
|
return doSelect(-1);
|
|
}
|
|
|
|
/* (non-Javadoc)
|
|
* @see java.nio.channels.Selector#select(long)
|
|
*/
|
|
public int select(long timeout) throws IOException
|
|
{
|
|
if (timeout == 0)
|
|
timeout = -1;
|
|
return doSelect(timeout);
|
|
}
|
|
|
|
/* (non-Javadoc)
|
|
* @see java.nio.channels.Selector#selectedKeys()
|
|
*/
|
|
public Set selectedKeys()
|
|
{
|
|
if (!isOpen())
|
|
throw new ClosedSelectorException();
|
|
|
|
return selected;
|
|
}
|
|
|
|
/* (non-Javadoc)
|
|
* @see java.nio.channels.Selector#selectNow()
|
|
*/
|
|
public int selectNow() throws IOException
|
|
{
|
|
return doSelect(0);
|
|
}
|
|
|
|
/* (non-Javadoc)
|
|
* @see java.nio.channels.Selector#wakeup()
|
|
*/
|
|
public Selector wakeup()
|
|
{
|
|
if (blockedThread != null)
|
|
blockedThread.interrupt();
|
|
return this;
|
|
}
|
|
|
|
public String toString()
|
|
{
|
|
return super.toString() + " [ fd: " + kq + " ]";
|
|
}
|
|
|
|
public boolean equals(Object o)
|
|
{
|
|
if (!(o instanceof KqueueSelectorImpl))
|
|
return false;
|
|
|
|
return ((KqueueSelectorImpl) o).kq == kq;
|
|
}
|
|
|
|
int doSelect(long timeout) throws IOException
|
|
{
|
|
Set cancelled = cancelledKeys();
|
|
synchronized (cancelled)
|
|
{
|
|
synchronized (keys)
|
|
{
|
|
for (Iterator it = cancelled.iterator(); it.hasNext(); )
|
|
{
|
|
KqueueSelectionKeyImpl key = (KqueueSelectionKeyImpl) it.next();
|
|
key.interestOps = 0;
|
|
}
|
|
|
|
int events_size = (2 * _sizeof_struct_kevent) * keys.size();
|
|
int num_events = 0;
|
|
|
|
for (Iterator it = keys.entrySet().iterator(); it.hasNext(); )
|
|
{
|
|
Map.Entry e = (Map.Entry) it.next();
|
|
KqueueSelectionKeyImpl key = (KqueueSelectionKeyImpl) e.getValue();
|
|
|
|
SelectableChannel ch = key.channel();
|
|
if (ch instanceof VMChannelOwner)
|
|
{
|
|
if (!((VMChannelOwner) ch).getVMChannel().getState().isValid())
|
|
{
|
|
// closed channel; removed from kqueue automatically.
|
|
it.remove();
|
|
continue;
|
|
}
|
|
}
|
|
|
|
// If this key is registering a read filter, add it to the buffer.
|
|
if (key.needCommitRead())
|
|
{
|
|
kevent_set(events, num_events, key.fd,
|
|
key.interestOps & (OP_READ | OP_ACCEPT),
|
|
key.activeOps & (OP_READ | OP_ACCEPT), key.key);
|
|
num_events++;
|
|
}
|
|
|
|
// If this key is registering a write filter, add it to the buffer.
|
|
if (key.needCommitWrite())
|
|
{
|
|
kevent_set(events, num_events, key.fd,
|
|
key.interestOps & (OP_WRITE | OP_CONNECT),
|
|
key.activeOps & (OP_WRITE | OP_CONNECT), key.key);
|
|
num_events++;
|
|
}
|
|
}
|
|
events.rewind().limit(events.capacity());
|
|
|
|
//System.out.println("dump of keys to select:");
|
|
//dump_selection_keys(events.duplicate());
|
|
|
|
int n = 0;
|
|
try
|
|
{
|
|
//System.out.println("[" + kq + "] kevent enter selecting from " + keys.size());
|
|
begin();
|
|
blockedThread = Thread.currentThread();
|
|
if (blockedThread.isInterrupted())
|
|
timeout = 0;
|
|
n = kevent(kq, events, num_events,
|
|
events.capacity() / _sizeof_struct_kevent, timeout);
|
|
}
|
|
finally
|
|
{
|
|
end();
|
|
blockedThread = null;
|
|
Thread.interrupted();
|
|
//System.out.println("[" + kq + "kevent exit selected " + n);
|
|
}
|
|
|
|
//System.out.println("dump of keys selected:");
|
|
//dump_selection_keys((ByteBuffer) events.duplicate().limit(n * _sizeof_struct_kevent));
|
|
|
|
// Commit the operations we've just added in the call to kevent.
|
|
for (Iterator it = keys.values().iterator(); it.hasNext(); )
|
|
{
|
|
KqueueSelectionKeyImpl key = (KqueueSelectionKeyImpl) it.next();
|
|
key.activeOps = key.interestOps;
|
|
}
|
|
|
|
selected = new HashSet/*<KqueueSelectionKeyImpl>*/(n);
|
|
int x = 0;
|
|
for (int i = 0; i < n; i++)
|
|
{
|
|
events.position(x).limit(x + _sizeof_struct_kevent);
|
|
x += _sizeof_struct_kevent;
|
|
int y = fetch_key(events.slice());
|
|
KqueueSelectionKeyImpl key =
|
|
(KqueueSelectionKeyImpl) keys.get(new Integer(y));
|
|
|
|
if (key == null)
|
|
{
|
|
System.out.println("WARNING! no key found for selected key " + y);
|
|
continue;
|
|
}
|
|
// Keys that have been cancelled may be returned here; don't
|
|
// add them to the selected set.
|
|
if (!key.isValid())
|
|
continue;
|
|
key.readyOps = ready_ops(events.slice(), key.interestOps);
|
|
selected.add(key);
|
|
}
|
|
|
|
// Finally, remove the cancelled keys.
|
|
for (Iterator it = cancelled.iterator(); it.hasNext(); )
|
|
{
|
|
KqueueSelectionKeyImpl key = (KqueueSelectionKeyImpl) it.next();
|
|
keys.remove(new Integer(key.key));
|
|
deregister(key);
|
|
it.remove();
|
|
}
|
|
|
|
reallocateBuffer();
|
|
|
|
return selected.size();
|
|
}
|
|
}
|
|
}
|
|
|
|
protected SelectionKey register(AbstractSelectableChannel channel,
|
|
int interestOps,
|
|
Object attachment)
|
|
{
|
|
int native_fd = -1;
|
|
try
|
|
{
|
|
if (channel instanceof VMChannelOwner)
|
|
native_fd = ((VMChannelOwner) channel).getVMChannel()
|
|
.getState().getNativeFD();
|
|
else
|
|
throw new IllegalArgumentException("cannot handle channel type " +
|
|
channel.getClass().getName());
|
|
}
|
|
catch (IOException ioe)
|
|
{
|
|
throw new IllegalArgumentException("channel is closed or invalid");
|
|
}
|
|
|
|
KqueueSelectionKeyImpl result = new KqueueSelectionKeyImpl(this, channel);
|
|
result.interestOps = interestOps;
|
|
result.attach(attachment);
|
|
result.fd = native_fd;
|
|
result.key = System.identityHashCode(result);
|
|
synchronized (keys)
|
|
{
|
|
while (keys.containsKey(new Integer(result.key)))
|
|
result.key++;
|
|
keys.put(new Integer(result.key), result);
|
|
reallocateBuffer();
|
|
}
|
|
return result;
|
|
}
|
|
|
|
void setInterestOps(KqueueSelectionKeyImpl key, int ops)
|
|
{
|
|
synchronized (keys)
|
|
{
|
|
key.interestOps = ops;
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Reallocate the events buffer. This is the destination buffer for
|
|
* events returned by kevent. This method will:
|
|
*
|
|
* * Grow the buffer if there is insufficent space for all registered
|
|
* events.
|
|
* * Shrink the buffer if it is more than twice the size needed.
|
|
*
|
|
*/
|
|
private void reallocateBuffer()
|
|
{
|
|
synchronized (keys)
|
|
{
|
|
if (events.capacity() < (2 * _sizeof_struct_kevent) * keys.size())
|
|
{
|
|
int cap = events.capacity();
|
|
if (cap >= MAX_DOUBLING_CAPACITY)
|
|
cap += CAP_INCREMENT;
|
|
else
|
|
cap = cap << 1;
|
|
|
|
events = ByteBuffer.allocateDirect(cap);
|
|
}
|
|
else if (events.capacity() > 4 * (_sizeof_struct_kevent) * keys.size() + 1
|
|
&& events.capacity() > INITIAL_CAPACITY)
|
|
{
|
|
int cap = events.capacity();
|
|
cap = cap >>> 1;
|
|
events = ByteBuffer.allocateDirect(cap);
|
|
}
|
|
}
|
|
}
|
|
|
|
//synchronized void updateOps(KqueueSelectionKeyImpl key, int interestOps)
|
|
//{
|
|
// updateOps(key, interestOps, 0, false);
|
|
//}
|
|
|
|
/*void updateOps(KqueueSelectionKeyImpl key, int interestOps,
|
|
int activeOps, int fd)
|
|
{
|
|
//System.out.println(">> updating kqueue selection key:");
|
|
//dump_selection_keys(key.nstate.duplicate());
|
|
//System.out.println("<<");
|
|
synchronized (keys)
|
|
{
|
|
kevent_set(key.nstate, fd, interestOps, activeOps, key.key);
|
|
}
|
|
//System.out.println(">> updated kqueue selection key:");
|
|
//dump_selection_keys(key.nstate.duplicate());
|
|
//System.out.println("<<");
|
|
}*/
|
|
|
|
private void dump_selection_keys(ByteBuffer keys)
|
|
{
|
|
// WARNING! This method is not guaranteed to be portable! This works
|
|
// on darwin/x86, but the sizeof and offsetof these fields may be
|
|
// different on other platforms!
|
|
int i = 0;
|
|
keys.order(ByteOrder.nativeOrder());
|
|
while (keys.hasRemaining())
|
|
{
|
|
System.out.println("struct kevent { ident: "
|
|
+ Integer.toString(keys.getInt())
|
|
+ " filter: "
|
|
+ Integer.toHexString(keys.getShort() & 0xFFFF)
|
|
+ " flags: "
|
|
+ Integer.toHexString(keys.getShort() & 0xFFFF)
|
|
+ " fflags: "
|
|
+ Integer.toHexString(keys.getInt())
|
|
+ " data: "
|
|
+ Integer.toHexString(keys.getInt())
|
|
+ " udata: "
|
|
+ Integer.toHexString(keys.getInt())
|
|
+ " }");
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Return the size of a <code>struct kevent</code> on this system.
|
|
*
|
|
* @return The size of <code>struct kevent</code>.
|
|
*/
|
|
private static native int sizeof_struct_kevent();
|
|
|
|
/**
|
|
* Opens a kqueue descriptor.
|
|
*
|
|
* @return The new kqueue descriptor.
|
|
* @throws IOException If opening fails.
|
|
*/
|
|
private static native int implOpen() throws IOException;
|
|
|
|
/**
|
|
* Closes the kqueue file descriptor.
|
|
*
|
|
* @param kq The kqueue file descriptor.
|
|
* @throws IOException
|
|
*/
|
|
private static native void implClose(int kq) throws IOException;
|
|
|
|
/**
|
|
* Initialize the specified native state for the given interest ops.
|
|
*
|
|
* @param nstate The native state structures; in this buffer should be
|
|
* the <code>struct kevent</code>s created for a key.
|
|
* @param fd The file descriptor. If 0, the native FD is unmodified.
|
|
* @param interestOps The operations to enable.
|
|
* @param key A unique key that will reference the associated key later.
|
|
* @param delete Set to true if this event should be deleted from the
|
|
* kqueue (if false, this event is added/updated).
|
|
*/
|
|
private static native void kevent_set(ByteBuffer nstate, int i, int fd,
|
|
int interestOps, int activeOps, int key);
|
|
|
|
/**
|
|
* Poll for events. The source events are stored in <code>events</code>,
|
|
* which is also where polled events will be placed.
|
|
*
|
|
* @param events The events to poll. This buffer is also the destination
|
|
* for events read from the queue.
|
|
* @param nevents The number of events to poll (that is, the number of
|
|
* events in the <code>events</code> buffer).
|
|
* @param nout The maximum number of events that may be returned.
|
|
* @param timeout The timeout. A timeout of -1 returns immediately; a timeout
|
|
* of 0 waits indefinitely.
|
|
* @return The number of events read.
|
|
*/
|
|
private static native int kevent(int kq, ByteBuffer events, int nevents,
|
|
int nout, long timeout);
|
|
|
|
/**
|
|
* Fetch a polled key from a native state buffer. For each kevent key we
|
|
* create, we put the native state info (one or more <code>struct
|
|
* kevent</code>s) in that key's {@link KqueueSelectionKeyImpl#nstate}
|
|
* buffer, and place the pointer of the key in the <code>udata</code> field
|
|
* of that structure. This method fetches that pointer from the given
|
|
* buffer (assumed to be a <code>struct kqueue</code>) and returns it.
|
|
*
|
|
* @param nstate The buffer containing the <code>struct kqueue</code> to read.
|
|
* @return The key object.
|
|
*/
|
|
private static native int fetch_key(ByteBuffer nstate);
|
|
|
|
/**
|
|
* Fetch the ready ops of the associated native state. That is, this
|
|
* inspects the first argument as a <code>struct kevent</code>, looking
|
|
* at its operation (the input is assumed to have been returned via a
|
|
* previous call to <code>kevent</code>), and translating that to the
|
|
* appropriate Java bit set, based on the second argument.
|
|
*
|
|
* @param nstate The native state.
|
|
* @param interestOps The enabled operations for the key.
|
|
* @return The bit set representing the ready operations.
|
|
*/
|
|
private static native int ready_ops(ByteBuffer nstate, int interestOps);
|
|
|
|
/**
|
|
* Check if kevent returned EV_EOF for a selection key.
|
|
*
|
|
* @param nstate The native state.
|
|
* @return True if the kevent call returned EOF.
|
|
*/
|
|
private static native boolean check_eof(ByteBuffer nstate);
|
|
}
|