archive/jdk
1
0
Fork 0
mirror of https://github.com/openjdk/jdk.git synced 2025-08-28 15:24:43 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

8200179: (se) More Selector cleanup

Browse source 
Reviewed-by: bpb
This commit is contained in:
Alan Bateman 2018-03-24 08:49:55 +00:00
parent 4b6cd06e3d
commit 187bf57418
19 changed files with 869 additions and 1247 deletions
Download patch file Download diff file Expand all files Collapse all files

278
src/java.base/solaris/classes/sun/nio/ch/DevPollArrayWrapper.java
View file

@ -26,12 +26,6 @@
package sun.nio.ch;
import java.io.IOException;
import java.security.AccessController;
import java.util.BitSet;
import java.util.Map;
import java.util.HashMap;
import sun.security.action.GetIntegerAction;
/**
* Manipulates a native array of pollfd structs on Solaris:
@ -48,43 +42,18 @@ import sun.security.action.GetIntegerAction;
class DevPollArrayWrapper {
// Event masks
static final short POLLIN = 0x0001;
static final short POLLPRI = 0x0002;
static final short POLLOUT = 0x0004;
static final short POLLRDNORM = 0x0040;
static final short POLLWRNORM = POLLOUT;
static final short POLLRDBAND = 0x0080;
static final short POLLWRBAND = 0x0100;
static final short POLLNORM = POLLRDNORM;
static final short POLLERR = 0x0008;
static final short POLLHUP = 0x0010;
static final short POLLNVAL = 0x0020;
// special event to remove a file descriptor from the driver
static final short POLLREMOVE = 0x0800;
static final short POLLCONN = POLLOUT;
// Miscellaneous constants
// struct pollfd constants
static final short SIZE_POLLFD = 8;
static final short FD_OFFSET = 0;
static final short EVENT_OFFSET = 4;
static final short REVENT_OFFSET = 6;
// Special value to indicate that an update should be ignored
static final byte IGNORE = (byte)-1;
// Maximum number of open file descriptors
static final int OPEN_MAX = IOUtil.fdLimit();
// Number of pollfd structures to create.
// dpwrite/ioctl(DP_POLL) allows up to OPEN_MAX-1
static final int NUM_POLLFDS = Math.min(OPEN_MAX-1, 8192);
// Initial size of arrays for fd registration changes
private static final int INITIAL_PENDING_UPDATE_SIZE = 64;
// maximum size of updatesLow
private static final int MAX_UPDATE_ARRAY_SIZE = AccessController.doPrivileged(
new GetIntegerAction("sun.nio.ch.maxUpdateArraySize", Math.min(OPEN_MAX, 64*1024)));
// maximum number of pollfd structure to poll or update at a time
// dpwrite/ioctl(DP_POLL) allows up to file descriptor limit minus 1
static final int NUM_POLLFDS = Math.min(IOUtil.fdLimit()-1, 1024);
// The pollfd array for results from devpoll driver
private final AllocatedNativeObject pollArray;
@ -95,122 +64,12 @@ class DevPollArrayWrapper {
// The fd of the devpoll driver
private int wfd;
// The fd of the interrupt line going out
private int outgoingInterruptFD;
// The fd of the interrupt line coming in
private int incomingInterruptFD;
// The index of the interrupt FD
private int interruptedIndex;
// Number of updated pollfd entries
int updated;
// object to synchronize fd registration changes
private final Object updateLock = new Object();
// number of file descriptors with registration changes pending
private int updateCount;
// file descriptors with registration changes pending
private int[] updateDescriptors = new int[INITIAL_PENDING_UPDATE_SIZE];
// events for file descriptors with registration changes pending, indexed
// by file descriptor and stored as bytes for efficiency reasons. For
// file descriptors higher than MAX_UPDATE_ARRAY_SIZE (unlimited case at
// least then the update is stored in a map.
private final byte[] eventsLow = new byte[MAX_UPDATE_ARRAY_SIZE];
private Map<Integer,Byte> eventsHigh;
// Used by release and updateRegistrations to track whether a file
// descriptor is registered with /dev/poll.
private final BitSet registered = new BitSet();
DevPollArrayWrapper() throws IOException {
this.wfd = init();
int allocationSize = NUM_POLLFDS * SIZE_POLLFD;
pollArray = new AllocatedNativeObject(allocationSize, true);
pollArrayAddress = pollArray.address();
wfd = init();
if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE)
eventsHigh = new HashMap<>();
}
void initInterrupt(int fd0, int fd1) throws IOException {
outgoingInterruptFD = fd1;
incomingInterruptFD = fd0;
register(wfd, fd0, POLLIN);
}
void putReventOps(int i, int revent) {
int offset = SIZE_POLLFD * i + REVENT_OFFSET;
pollArray.putShort(offset, (short)revent);
}
int getEventOps(int i) {
int offset = SIZE_POLLFD * i + EVENT_OFFSET;
return pollArray.getShort(offset);
}
int getReventOps(int i) {
int offset = SIZE_POLLFD * i + REVENT_OFFSET;
return pollArray.getShort(offset);
}
int getDescriptor(int i) {
int offset = SIZE_POLLFD * i + FD_OFFSET;
return pollArray.getInt(offset);
}
private void setUpdateEvents(int fd, byte events) {
if (fd < MAX_UPDATE_ARRAY_SIZE) {
eventsLow[fd] = events;
} else {
eventsHigh.put(Integer.valueOf(fd), Byte.valueOf(events));
}
}
private byte getUpdateEvents(int fd) {
if (fd < MAX_UPDATE_ARRAY_SIZE) {
return eventsLow[fd];
} else {
Byte result = eventsHigh.get(Integer.valueOf(fd));
// result should never be null
return result.byteValue();
}
}
void setInterest(int fd, int mask) {
synchronized (updateLock) {
// record the file descriptor and events, expanding the
// respective arrays first if necessary.
int oldCapacity = updateDescriptors.length;
if (updateCount == oldCapacity) {
int newCapacity = oldCapacity + INITIAL_PENDING_UPDATE_SIZE;
int[] newDescriptors = new int[newCapacity];
System.arraycopy(updateDescriptors, 0, newDescriptors, 0, oldCapacity);
updateDescriptors = newDescriptors;
}
updateDescriptors[updateCount++] = fd;
// events are stored as bytes for efficiency reasons
byte b = (byte)mask;
assert (b == mask) && (b != IGNORE);
setUpdateEvents(fd, b);
}
}
void release(int fd) throws IOException {
synchronized (updateLock) {
// ignore any pending update for this file descriptor
setUpdateEvents(fd, IGNORE);
// remove from /dev/poll
if (registered.get(fd)) {
register(wfd, fd, POLLREMOVE);
registered.clear(fd);
}
}
this.pollArray = new AllocatedNativeObject(allocationSize, true);
this.pollArrayAddress = pollArray.address();
}
void close() throws IOException {
@ -218,102 +77,41 @@ class DevPollArrayWrapper {
pollArray.free();
}
void register(int fd, int ops) throws IOException {
register(wfd, fd, ops);
}
void registerMultiple(int numfds) throws IOException {
registerMultiple(wfd, pollArrayAddress, numfds);
}
int poll(long timeout) throws IOException {
updateRegistrations();
updated = poll0(pollArrayAddress, NUM_POLLFDS, timeout, wfd);
for (int i=0; i<updated; i++) {
if (getDescriptor(i) == incomingInterruptFD) {
interruptedIndex = i;
interrupted = true;
break;
}
}
return updated;
return poll0(pollArrayAddress, NUM_POLLFDS, timeout, wfd);
}
void updateRegistrations() throws IOException {
synchronized (updateLock) {
// Populate pollfd array with updated masks
int j = 0;
int index = 0;
while (j < updateCount) {
int fd = updateDescriptors[j];
short events = getUpdateEvents(fd);
boolean wasRegistered = registered.get(fd);
// events = 0 => POLLREMOVE or do-nothing
if (events != IGNORE) {
if (events == 0) {
if (wasRegistered) {
events = POLLREMOVE;
registered.clear(fd);
} else {
events = IGNORE;
}
} else {
if (!wasRegistered) {
registered.set(fd);
}
}
}
// populate pollfd array with updated event
if (events != IGNORE) {
// insert POLLREMOVE if changing events
if (wasRegistered && events != POLLREMOVE) {
putPollFD(pollArray, index, fd, POLLREMOVE);
index++;
}
putPollFD(pollArray, index, fd, events);
index++;
if (index >= (NUM_POLLFDS-1)) {
registerMultiple(wfd, pollArray.address(), index);
index = 0;
}
// events for this fd now up to date
setUpdateEvents(fd, IGNORE);
}
j++;
}
// write any remaining updates
if (index > 0)
registerMultiple(wfd, pollArray.address(), index);
updateCount = 0;
}
int getDescriptor(int i) {
int offset = SIZE_POLLFD * i + FD_OFFSET;
return pollArray.getInt(offset);
}
private void putPollFD(AllocatedNativeObject array, int index, int fd,
short event)
{
short getEventOps(int i) {
int offset = SIZE_POLLFD * i + EVENT_OFFSET;
return pollArray.getShort(offset);
}
short getReventOps(int i) {
int offset = SIZE_POLLFD * i + REVENT_OFFSET;
return pollArray.getShort(offset);
}
/**
* Updates the pollfd structure at the given index
*/
void putPollFD(int index, int fd, short event) {
int structIndex = SIZE_POLLFD * index;
array.putInt(structIndex + FD_OFFSET, fd);
array.putShort(structIndex + EVENT_OFFSET, event);
array.putShort(structIndex + REVENT_OFFSET, (short)0);
}
boolean interrupted = false;
public void interrupt() {
try {
IOUtil.write1(outgoingInterruptFD, (byte)0);
} catch (IOException ioe) {
throw new InternalError(ioe);
}
}
public int interruptedIndex() {
return interruptedIndex;
}
boolean interrupted() {
return interrupted;
}
void clearInterrupted() {
interrupted = false;
pollArray.putInt(structIndex + FD_OFFSET, fd);
pollArray.putShort(structIndex + EVENT_OFFSET, event);
pollArray.putShort(structIndex + REVENT_OFFSET, (short)0);
}
private native int init() throws IOException;

264
src/java.base/solaris/classes/sun/nio/ch/DevPollSelectorImpl.java
View file

@ -26,67 +26,71 @@
package sun.nio.ch;
import java.io.IOException;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.util.*;
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.SelectorProvider;
import java.util.ArrayDeque;
import java.util.BitSet;
import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import static sun.nio.ch.DevPollArrayWrapper.NUM_POLLFDS;
import static sun.nio.ch.DevPollArrayWrapper.POLLREMOVE;
/**
* An implementation of Selector for Solaris.
* Solaris /dev/poll based Selector implementation
*/
class DevPollSelectorImpl
extends SelectorImpl
{
// File descriptors used for interrupt
// provides access to /dev/poll driver
private final DevPollArrayWrapper pollWrapper;
// file descriptors used for interrupt
private final int fd0;
private final int fd1;
// The poll object
private final DevPollArrayWrapper pollWrapper;
// maps file descriptor to selection key, synchronize on selector
private final Map<Integer, SelectionKeyImpl> fdToKey = new HashMap<>();
// Maps from file descriptors to keys
private final Map<Integer, SelectionKeyImpl> fdToKey;
// file descriptors registered with /dev/poll, synchronize on selector
private final BitSet registered = new BitSet();
// True if this Selector has been closed
private boolean closed;
// pending new registrations/updates, queued by implRegister and putEventOps
private final Object updateLock = new Object();
private final Deque<SelectionKeyImpl> newKeys = new ArrayDeque<>();
private final Deque<SelectionKeyImpl> updateKeys = new ArrayDeque<>();
private final Deque<Integer> updateOps = new ArrayDeque<>();
// Lock for close/cleanup
private final Object closeLock = new Object();
// Lock for interrupt triggering and clearing
// interrupt triggering and clearing
private final Object interruptLock = new Object();
private boolean interruptTriggered;
/**
* Package private constructor called by factory method in
* the abstract superclass Selector.
*/
DevPollSelectorImpl(SelectorProvider sp) throws IOException {
super(sp);
long pipeFds = IOUtil.makePipe(false);
fd0 = (int) (pipeFds >>> 32);
fd1 = (int) pipeFds;
this.pollWrapper = new DevPollArrayWrapper();
try {
pollWrapper = new DevPollArrayWrapper();
pollWrapper.initInterrupt(fd0, fd1);
fdToKey = new HashMap<>();
} catch (Throwable t) {
try {
FileDispatcherImpl.closeIntFD(fd0);
} catch (IOException ioe0) {
t.addSuppressed(ioe0);
}
try {
FileDispatcherImpl.closeIntFD(fd1);
} catch (IOException ioe1) {
t.addSuppressed(ioe1);
}
throw t;
long fds = IOUtil.makePipe(false);
this.fd0 = (int) (fds >>> 32);
this.fd1 = (int) fds;
} catch (IOException ioe) {
pollWrapper.close();
throw ioe;
}
// register one end of the socket pair for wakeups
pollWrapper.register(fd0, Net.POLLIN);
}
private void ensureOpen() {
if (closed)
if (!isOpen())
throw new ClosedSelectorException();
}
@ -94,62 +98,123 @@ class DevPollSelectorImpl
protected int doSelect(long timeout)
throws IOException
{
ensureOpen();
assert Thread.holdsLock(this);
int numEntries;
processUpdateQueue();
processDeregisterQueue();
try {
begin();
pollWrapper.poll(timeout);
numEntries = pollWrapper.poll(timeout);
} finally {
end();
}
processDeregisterQueue();
int numKeysUpdated = updateSelectedKeys();
if (pollWrapper.interrupted()) {
// Clear the wakeup pipe
pollWrapper.putReventOps(pollWrapper.interruptedIndex(), 0);
synchronized (interruptLock) {
pollWrapper.clearInterrupted();
IOUtil.drain(fd0);
interruptTriggered = false;
}
}
return numKeysUpdated;
return updateSelectedKeys(numEntries);
}
/**
* Update the keys whose fd's have been selected by the devpoll
* driver. Add the ready keys to the ready queue.
* Process new registrations and changes to the interest ops.
*/
private int updateSelectedKeys() {
int entries = pollWrapper.updated;
int numKeysUpdated = 0;
for (int i=0; i<entries; i++) {
int nextFD = pollWrapper.getDescriptor(i);
SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
// ski is null in the case of an interrupt
if (ski != null) {
int rOps = pollWrapper.getReventOps(i);
if (selectedKeys.contains(ski)) {
if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
numKeysUpdated++;
private void processUpdateQueue() throws IOException {
assert Thread.holdsLock(this);
synchronized (updateLock) {
SelectionKeyImpl ski;
// new registrations
while ((ski = newKeys.pollFirst()) != null) {
if (ski.isValid()) {
SelChImpl ch = ski.channel;
int fd = ch.getFDVal();
SelectionKeyImpl previous = fdToKey.put(fd, ski);
assert previous == null;
assert registered.get(fd) == false;
}
}
// Translate the queued updates to changes to the set of monitored
// file descriptors. The changes are written to the /dev/poll driver
// in bulk.
assert updateKeys.size() == updateOps.size();
int index = 0;
while ((ski = updateKeys.pollFirst()) != null) {
int ops = updateOps.pollFirst();
int fd = ski.channel.getFDVal();
if (ski.isValid() && fdToKey.containsKey(fd)) {
if (registered.get(fd)) {
if (ops == 0) {
// remove file descriptor
pollWrapper.putPollFD(index++, fd, POLLREMOVE);
registered.clear(fd);
} else {
// change events
pollWrapper.putPollFD(index++, fd, POLLREMOVE);
pollWrapper.putPollFD(index++, fd, (short)ops);
}
} else if (ops != 0) {
// add file descriptor
pollWrapper.putPollFD(index++, fd, (short)ops);
registered.set(fd);
}
} else {
ski.channel.translateAndSetReadyOps(rOps, ski);
if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
selectedKeys.add(ski);
numKeysUpdated++;
if (index > (NUM_POLLFDS-2)) {
pollWrapper.registerMultiple(index);
index = 0;
}
}
}
// write any remaining changes
if (index > 0)
pollWrapper.registerMultiple(index);
}
}
/**
* Update the keys whose fd's have been selected by the /dev/poll.
* Add the ready keys to the ready queue.
*/
private int updateSelectedKeys(int numEntries) throws IOException {
assert Thread.holdsLock(this);
assert Thread.holdsLock(nioSelectedKeys());
boolean interrupted = false;
int numKeysUpdated = 0;
for (int i=0; i<numEntries; i++) {
int fd = pollWrapper.getDescriptor(i);
if (fd == fd0) {
interrupted = true;
} else {
SelectionKeyImpl ski = fdToKey.get(fd);
if (ski != null) {
int rOps = pollWrapper.getReventOps(i);
if (selectedKeys.contains(ski)) {
if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
numKeysUpdated++;
}
} else {
ski.channel.translateAndSetReadyOps(rOps, ski);
if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
selectedKeys.add(ski);
numKeysUpdated++;
}
}
}
}
}
if (interrupted) {
clearInterrupt();
}
return numKeysUpdated;
}
@Override
protected void implClose() throws IOException {
if (closed)
return;
closed = true;
assert !isOpen();
assert Thread.holdsLock(this);
assert Thread.holdsLock(nioKeys());
// prevent further wakeup
synchronized (interruptLock) {
@ -174,42 +239,67 @@ class DevPollSelectorImpl
@Override
protected void implRegister(SelectionKeyImpl ski) {
int fd = IOUtil.fdVal(ski.channel.getFD());
fdToKey.put(Integer.valueOf(fd), ski);
assert Thread.holdsLock(nioKeys());
ensureOpen();
synchronized (updateLock) {
newKeys.addLast(ski);
}
keys.add(ski);
}
@Override
protected void implDereg(SelectionKeyImpl ski) throws IOException {
int i = ski.getIndex();
assert (i >= 0);
assert !ski.isValid();
assert Thread.holdsLock(this);
assert Thread.holdsLock(nioKeys());
assert Thread.holdsLock(nioSelectedKeys());
int fd = ski.channel.getFDVal();
fdToKey.remove(Integer.valueOf(fd));
pollWrapper.release(fd);
ski.setIndex(-1);
keys.remove(ski);
fdToKey.remove(fd);
if (registered.get(fd)) {
pollWrapper.register(fd, POLLREMOVE);
registered.clear(fd);
}
selectedKeys.remove(ski);
deregister((AbstractSelectionKey)ski);
keys.remove(ski);
// remove from channel's key set
deregister(ski);
SelectableChannel selch = ski.channel();
if (!selch.isOpen() && !selch.isRegistered())
((SelChImpl)selch).kill();
((SelChImpl) selch).kill();
}
@Override
public void putEventOps(SelectionKeyImpl sk, int ops) {
public void putEventOps(SelectionKeyImpl ski, int ops) {
ensureOpen();
int fd = IOUtil.fdVal(sk.channel.getFD());
pollWrapper.setInterest(fd, ops);
synchronized (updateLock) {
updateOps.addLast(ops); // ops first in case adding the key fails
updateKeys.addLast(ski);
}
}
@Override
public Selector wakeup() {
synchronized (interruptLock) {
if (!interruptTriggered) {
pollWrapper.interrupt();
try {
IOUtil.write1(fd1, (byte)0);
} catch (IOException ioe) {
throw new InternalError(ioe);
}
interruptTriggered = true;
}
}
return this;
}
private void clearInterrupt() throws IOException {
synchronized (interruptLock) {
IOUtil.drain(fd0);
interruptTriggered = false;
}
}
}

300
src/java.base/solaris/classes/sun/nio/ch/EventPortSelectorImpl.java
View file

@ -26,11 +26,31 @@
package sun.nio.ch;
import java.io.IOException;
import java.nio.channels.*;
import java.nio.channels.spi.*;
import java.util.Map;
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.SelectorProvider;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import jdk.internal.misc.Unsafe;
import static sun.nio.ch.SolarisEventPort.PORT_SOURCE_FD;
import static sun.nio.ch.SolarisEventPort.PORT_SOURCE_USER;
import static sun.nio.ch.SolarisEventPort.SIZEOF_PORT_EVENT;
import static sun.nio.ch.SolarisEventPort.OFFSETOF_EVENTS;
import static sun.nio.ch.SolarisEventPort.OFFSETOF_SOURCE;
import static sun.nio.ch.SolarisEventPort.OFFSETOF_OBJECT;
import static sun.nio.ch.SolarisEventPort.port_create;
import static sun.nio.ch.SolarisEventPort.port_close;
import static sun.nio.ch.SolarisEventPort.port_associate;
import static sun.nio.ch.SolarisEventPort.port_dissociate;
import static sun.nio.ch.SolarisEventPort.port_getn;
import static sun.nio.ch.SolarisEventPort.port_send;
/**
* Selector implementation based on the Solaris event port mechanism.
@ -39,89 +59,206 @@ import java.util.Iterator;
class EventPortSelectorImpl
extends SelectorImpl
{
private final EventPortWrapper pollWrapper;
// maximum number of events to retrive in one call to port_getn
static final int MAX_EVENTS = Math.min(IOUtil.fdLimit()-1, 1024);
// Maps from file descriptors to keys
private final Map<Integer, SelectionKeyImpl> fdToKey;
// port file descriptor
private final int pfd;
// True if this Selector has been closed
private boolean closed;
// the poll array (populated by port_getn)
private final long pollArrayAddress;
private final AllocatedNativeObject pollArray;
// Lock for interrupt triggering and clearing
// a registration of a file descriptor with a selector
private static class RegEntry {
final SelectionKeyImpl ski;
int registeredOps;
int lastUpdate;
RegEntry(SelectionKeyImpl ski) {
this.ski = ski;
}
}
// maps a file descriptor to registration entry, synchronize on selector
private final Map<Integer, RegEntry> fdToRegEntry = new HashMap<>();
// the last update operation, incremented by processUpdateQueue
private int lastUpdate;
// pending new registrations/updates, queued by implRegister, putEventOps,
// and updateSelectedKeys
private final Object updateLock = new Object();
private final Deque<SelectionKeyImpl> newKeys = new ArrayDeque<>();
private final Deque<SelectionKeyImpl> updateKeys = new ArrayDeque<>();
private final Deque<Integer> updateOps = new ArrayDeque<>();
// interrupt triggering and clearing
private final Object interruptLock = new Object();
private boolean interruptTriggered;
/**
* Package private constructor called by factory method in
* the abstract superclass Selector.
*/
EventPortSelectorImpl(SelectorProvider sp) throws IOException {
super(sp);
pollWrapper = new EventPortWrapper();
fdToKey = new HashMap<>();
this.pfd = port_create();
int allocationSize = MAX_EVENTS * SIZEOF_PORT_EVENT;
this.pollArray = new AllocatedNativeObject(allocationSize, false);
this.pollArrayAddress = pollArray.address();
}
private void ensureOpen() {
if (closed)
if (!isOpen())
throw new ClosedSelectorException();
}
@Override
protected int doSelect(long timeout) throws IOException {
ensureOpen();
assert Thread.holdsLock(this);
int numEvents;
processUpdateQueue();
processDeregisterQueue();
int entries;
try {
begin();
entries = pollWrapper.poll(timeout);
long to = timeout;
boolean timedPoll = (to > 0);
do {
long startTime = timedPoll ? System.nanoTime() : 0;
numEvents = port_getn(pfd, pollArrayAddress, MAX_EVENTS, to);
if (numEvents == IOStatus.INTERRUPTED && timedPoll) {
// timed poll interrupted so need to adjust timeout
long adjust = System.nanoTime() - startTime;
to -= TimeUnit.MILLISECONDS.convert(adjust, TimeUnit.NANOSECONDS);
if (to <= 0) {
// timeout also expired so no retry
numEvents = 0;
}
}
} while (numEvents == IOStatus.INTERRUPTED);
assert IOStatus.check(numEvents);
} finally {
end();
}
processDeregisterQueue();
int numKeysUpdated = updateSelectedKeys(entries);
if (pollWrapper.interrupted()) {
synchronized (interruptLock) {
interruptTriggered = false;
}
}
return numKeysUpdated;
return processPortEvents(numEvents);
}
private int updateSelectedKeys(int entries) {
int numKeysUpdated = 0;
for (int i=0; i<entries; i++) {
int nextFD = pollWrapper.getDescriptor(i);
SelectionKeyImpl ski = fdToKey.get(Integer.valueOf(nextFD));
if (ski != null) {
int rOps = pollWrapper.getEventOps(i);
if (selectedKeys.contains(ski)) {
if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
numKeysUpdated++;
}
} else {
ski.channel.translateAndSetReadyOps(rOps, ski);
if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
selectedKeys.add(ski);
numKeysUpdated++;
}
/**
* Process new registrations and changes to the interest ops.
*/
private void processUpdateQueue() throws IOException {
assert Thread.holdsLock(this);
// bump lastUpdate to ensure that the interest ops are changed at most
// once per bulk update
lastUpdate++;
synchronized (updateLock) {
SelectionKeyImpl ski;
// new registrations
while ((ski = newKeys.pollFirst()) != null) {
if (ski.isValid()) {
SelChImpl ch = ski.channel;
int fd = ch.getFDVal();
RegEntry previous = fdToRegEntry.put(fd, new RegEntry(ski));
assert previous == null;
}
}
// changes to interest ops
assert updateKeys.size() == updateOps.size();
while ((ski = updateKeys.pollFirst()) != null) {
int ops = updateOps.pollFirst();
int fd = ski.channel.getFDVal();
RegEntry e = fdToRegEntry.get(fd);
if (ski.isValid() && (e != null) && (e.lastUpdate != lastUpdate)) {
assert e.ski == ski;
if ((ops != e.registeredOps)) {
if (ops == 0) {
port_dissociate(pfd, PORT_SOURCE_FD, fd);
} else {
port_associate(pfd, PORT_SOURCE_FD, fd, ops);
}
e.registeredOps = ops;
}
e.lastUpdate = lastUpdate;
}
}
}
}
/**
* Process the port events. This method updates the keys of file descriptors
* that were polled. It also re-queues the key so that the file descriptor
* is re-associated at the next select operation.
*
* @return the number of selection keys updated.
*/
private int processPortEvents(int numEvents) throws IOException {
assert Thread.holdsLock(this);
assert Thread.holdsLock(nioSelectedKeys());
int numKeysUpdated = 0;
boolean interrupted = false;
synchronized (updateLock) {
for (int i = 0; i < numEvents; i++) {
short source = getSource(i);
if (source == PORT_SOURCE_FD) {
int fd = getDescriptor(i);
RegEntry e = fdToRegEntry.get(fd);
if (e != null) {
SelectionKeyImpl ski = e.ski;
int rOps = getEventOps(i);
if (selectedKeys.contains(ski)) {
if (ski.channel.translateAndSetReadyOps(rOps, ski)) {
numKeysUpdated++;
}
} else {
ski.channel.translateAndSetReadyOps(rOps, ski);
if ((ski.nioReadyOps() & ski.nioInterestOps()) != 0) {
selectedKeys.add(ski);
numKeysUpdated++;
}
}
// queue selection key so that it is re-associated at
// next select. Push to end of deque so that changes to
// the interest ops are processed first
updateKeys.addLast(ski);
updateOps.addLast(e.registeredOps);
e.registeredOps = 0;
}
} else if (source == PORT_SOURCE_USER) {
interrupted = true;
} else {
assert false;
}
}
}
if (interrupted) {
clearInterrupt();
}
return numKeysUpdated;
}
@Override
protected void implClose() throws IOException {
if (closed)
return;
closed = true;
assert !isOpen();
assert Thread.holdsLock(this);
assert Thread.holdsLock(nioKeys());
// prevent further wakeup
synchronized (interruptLock) {
interruptTriggered = true;
}
pollWrapper.close();
port_close(pfd);
pollArray.free();
// Deregister channels
Iterator<SelectionKey> i = keys.iterator();
@ -137,42 +274,83 @@ class EventPortSelectorImpl
@Override
protected void implRegister(SelectionKeyImpl ski) {
int fd = IOUtil.fdVal(ski.channel.getFD());
fdToKey.put(Integer.valueOf(fd), ski);
assert Thread.holdsLock(nioKeys());
ensureOpen();
synchronized (updateLock) {
newKeys.addLast(ski);
}
keys.add(ski);
}
@Override
protected void implDereg(SelectionKeyImpl ski) throws IOException {
int i = ski.getIndex();
assert (i >= 0);
assert !ski.isValid();
assert Thread.holdsLock(this);
assert Thread.holdsLock(nioKeys());
assert Thread.holdsLock(nioSelectedKeys());
int fd = ski.channel.getFDVal();
fdToKey.remove(Integer.valueOf(fd));
pollWrapper.release(fd);
ski.setIndex(-1);
keys.remove(ski);
RegEntry e = fdToRegEntry.remove(fd);
if (e != null && e.registeredOps != 0) {
port_dissociate(pfd, PORT_SOURCE_FD, fd);
}
selectedKeys.remove(ski);
deregister((AbstractSelectionKey)ski);
keys.remove(ski);
// remove from channel's key set
deregister(ski);
SelectableChannel selch = ski.channel();
if (!selch.isOpen() && !selch.isRegistered())
((SelChImpl)selch).kill();
((SelChImpl) selch).kill();
}
@Override
public void putEventOps(SelectionKeyImpl sk, int ops) {
public void putEventOps(SelectionKeyImpl ski, int ops) {
ensureOpen();
int fd = sk.channel.getFDVal();
pollWrapper.setInterest(fd, ops);
synchronized (updateLock) {
// push to front of deque so that it processed before other
// updates for the same key.
updateOps.addFirst(ops);
updateKeys.addFirst(ski);
}
}
@Override
public Selector wakeup() {
synchronized (interruptLock) {
if (!interruptTriggered) {
pollWrapper.interrupt();
try {
port_send(pfd, 0);
} catch (IOException ioe) {
throw new InternalError(ioe);
}
interruptTriggered = true;
}
}
return this;
}
private void clearInterrupt() throws IOException {
synchronized (interruptLock) {
interruptTriggered = false;
}
}
private short getSource(int i) {
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_SOURCE;
return pollArray.getShort(offset);
}
private int getEventOps(int i) {
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_EVENTS;
return pollArray.getInt(offset);
}
private int getDescriptor(int i) {
//assert Unsafe.getUnsafe().addressSize() == 8;
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_OBJECT;
return (int) pollArray.getLong(offset);
}
}

276
src/java.base/solaris/classes/sun/nio/ch/EventPortWrapper.java
View file

@ -1,276 +0,0 @@
/*
* Copyright (c) 2012, 2018, 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.io.IOException;
import java.security.AccessController;
import java.util.BitSet;
import java.util.HashMap;
import java.util.Map;
import jdk.internal.misc.Unsafe;
import sun.security.action.GetIntegerAction;
import static sun.nio.ch.SolarisEventPort.*;
/**
* Manages a Solaris event port and manipulates a native array of pollfd structs
* on Solaris.
*/
class EventPortWrapper {
private static final Unsafe unsafe = Unsafe.getUnsafe();
private static final int addressSize = unsafe.addressSize();
// Maximum number of open file descriptors
static final int OPEN_MAX = IOUtil.fdLimit();
// Maximum number of events to retrive in one call to port_getn
static final int POLL_MAX = Math.min(OPEN_MAX-1, 1024);
// initial size of the array to hold pending updates
private final int INITIAL_PENDING_UPDATE_SIZE = 256;
// maximum size of updateArray
private static final int MAX_UPDATE_ARRAY_SIZE = AccessController.doPrivileged(
new GetIntegerAction("sun.nio.ch.maxUpdateArraySize", Math.min(OPEN_MAX, 64*1024)));
// special update status to indicate that it should be ignored
private static final byte IGNORE = -1;
// port file descriptor
private final int pfd;
// the poll array (populated by port_getn)
private final long pollArrayAddress;
private final AllocatedNativeObject pollArray;
// required when accessing the update* fields
private final Object updateLock = new Object();
// the number of pending updates
private int updateCount;
// queue of file descriptors with updates pending
private int[] updateDescriptors = new int[INITIAL_PENDING_UPDATE_SIZE];
// events for file descriptors with registration changes pending, indexed
// by file descriptor and stored as bytes for efficiency reasons. For
// file descriptors higher than MAX_UPDATE_ARRAY_SIZE (unlimited case at
// least then the update is stored in a map.
private final byte[] eventsLow = new byte[MAX_UPDATE_ARRAY_SIZE];
private Map<Integer,Byte> eventsHigh;
// Used by release and updateRegistrations to track whether a file
// descriptor is registered with /dev/poll.
private final BitSet registered = new BitSet();
// bit set to indicate if a file descriptor has been visited when
// processing updates (used to avoid duplicates calls to port_associate)
private BitSet visited = new BitSet();
EventPortWrapper() throws IOException {
int allocationSize = POLL_MAX * SIZEOF_PORT_EVENT;
pollArray = new AllocatedNativeObject(allocationSize, true);
pollArrayAddress = pollArray.address();
this.pfd = port_create();
if (OPEN_MAX > MAX_UPDATE_ARRAY_SIZE)
eventsHigh = new HashMap<>();
}
void close() throws IOException {
port_close(pfd);
pollArray.free();
}
private short getSource(int i) {
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_SOURCE;
return pollArray.getShort(offset);
}
int getEventOps(int i) {
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_EVENTS;
return pollArray.getInt(offset);
}
int getDescriptor(int i) {
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_OBJECT;
if (addressSize == 4) {
return pollArray.getInt(offset);
} else {
return (int) pollArray.getLong(offset);
}
}
private void setDescriptor(int i, int fd) {
int offset = SIZEOF_PORT_EVENT * i + OFFSETOF_OBJECT;
if (addressSize == 4) {
pollArray.putInt(offset, fd);
} else {
pollArray.putLong(offset, fd);
}
}
private void setUpdate(int fd, byte events) {
if (fd < MAX_UPDATE_ARRAY_SIZE) {
eventsLow[fd] = events;
} else {
eventsHigh.put(Integer.valueOf(fd), Byte.valueOf(events));
}
}
private byte getUpdate(int fd) {
if (fd < MAX_UPDATE_ARRAY_SIZE) {
return eventsLow[fd];
} else {
Byte result = eventsHigh.get(Integer.valueOf(fd));
// result should never be null
return result.byteValue();
}
}
int poll(long timeout) throws IOException {
// update registrations prior to poll
synchronized (updateLock) {
// process newest updates first
int i = updateCount - 1;
while (i >= 0) {
int fd = updateDescriptors[i];
if (!visited.get(fd)) {
short ev = getUpdate(fd);
if (ev != IGNORE) {
if (ev == 0) {
if (registered.get(fd)) {
port_dissociate(pfd, PORT_SOURCE_FD, (long)fd);
registered.clear(fd);
}
} else {
if (port_associate(pfd, PORT_SOURCE_FD, (long)fd, ev)) {
registered.set(fd);
}
}
}
visited.set(fd);
}
i--;
}
updateCount = 0;
}
// poll for events
int numEntries;
long to = timeout;
boolean timedPoll = (to > 0);
do {
long startTime = timedPoll ? System.currentTimeMillis() : 0;
numEntries = port_getn(pfd, pollArrayAddress, POLL_MAX, timeout);
if (numEntries == IOStatus.INTERRUPTED && timedPoll) {
// timed poll interrupted so need to adjust timeout
to -= System.currentTimeMillis() - startTime;
if (to <= 0) {
// timeout also expired so no retry
numEntries = 0;
}
}
} while (numEntries == IOStatus.INTERRUPTED);
// after polling we need to queue all polled file descriptors as they
// are candidates to register for the next poll.
synchronized (updateLock) {
for (int i=0; i<numEntries; i++) {
if (getSource(i) == PORT_SOURCE_USER) {
interrupted = true;
setDescriptor(i, -1);
} else {
// the default is to re-associate for the next poll
int fd = getDescriptor(i);
registered.clear(fd);
setInterest(fd);
}
}
}
return numEntries;
}
private void setInterest(int fd) {
assert Thread.holdsLock(updateLock);
// record the file descriptor and events, expanding the
// respective arrays first if necessary.
int oldCapacity = updateDescriptors.length;
if (updateCount >= oldCapacity) {
int newCapacity = oldCapacity + INITIAL_PENDING_UPDATE_SIZE;
int[] newDescriptors = new int[newCapacity];
System.arraycopy(updateDescriptors, 0, newDescriptors, 0, oldCapacity);
updateDescriptors = newDescriptors;
}
updateDescriptors[updateCount++] = fd;
visited.clear(fd);
}
void setInterest(int fd, int mask) {
synchronized (updateLock) {
setInterest(fd);
setUpdate(fd, (byte)mask);
assert getUpdate(fd) == mask;
}
}
void release(int fd) {
synchronized (updateLock) {
if (registered.get(fd)) {
try {
port_dissociate(pfd, PORT_SOURCE_FD, (long)fd);
} catch (IOException ioe) {
throw new InternalError(ioe);
}
registered.clear(fd);
}
setUpdate(fd, IGNORE);
}
}
// -- wakeup support --
private boolean interrupted;
public void interrupt() {
try {
port_send(pfd, 0);
} catch (IOException ioe) {
throw new InternalError(ioe);
}
}
boolean interrupted() {
return interrupted;
}
void clearInterrupted() {
interrupted = false;
}
}