Требуется, чтобы какая-то система выдавала мне прерывания по объему, чтобы прерывание потока было отключено, когда Thread.sleep (..) не выполняется. Я бы обычно использовал защищенные блоки, которые работали бы нормально, но есть несколько причин, по которым я не могу в этом случае, главным образом потому, что конфликтные ситуации мешают всей работе должным образом. Симуляция просто зависает через несколько секунд. Прерывания с заданной областью заменили бы любой механизм блокировки и предотвратили бы «утечку» прерывания за пределы метода, на котором тестируется прерывание. В этом случае это будет Thread.sleep(...). Тем не менее, мне интересно, если это требует чего-то, что принципиально невозможно в Java. Возможно, лучше всего использовать механизм защищенной блокировки.
Я создал собственную реализацию селектора для использования в симуляторе сети (селектор взаимодействует с обратными вызовами симулятора и тому подобным для оптимизации), и в настоящее время я использую CountDownLatch в данный момент. Это хорошо в одних ситуациях, в то время как в других это создает еще хуже блокировку, при этом не замораживая всю симуляцию как таковую (однако, она настолько медленная, что не работает). Зависит от сетевой среды, используемой в отношении того, какие части кода получают больше, чем другие. Я также думаю, что в «хороших» симуляциях проблема может возникать, редко, вызывая ожидание 13 мс и так, пока внутренний CountDownLatch получен или какая-то магия происходит внутри.
Если вы можете решить эту проблему без каких-либо вышеупомянутых предложений, это было бы замечательно :) В любом случае, мой код ниже:
/**
*
*/
package kokunet;
import java.io.IOException;
import java.nio.channels.ClosedSelectorException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import kokuks.IConnectionSocket;
import kokuks.KKSAddress;
import kokuks.KKSSocket;
import kokuks.KKSSocketListener;
public class KSelector extends SelectorImpl {
// True if this Selector has been closed
private volatile boolean closed = false;
// Lock for close and cleanup
final class CloseLock {}
private final Object closeLock = new CloseLock();
private volatile boolean selecting = false;
private volatile boolean wakeup = false;
class SocketListener implements KKSSocketListener {
protected volatile CountDownLatch latch = null;
/**
*
*/
public SocketListener() {
newLatch();
}
protected synchronized CountDownLatch newLatch() {
return this.latch = new CountDownLatch(1);
}
protected synchronized void refreshReady(KKSSocket socket) {
if (!selecting) return;
synchronized (socketToChannel) {
SelChImpl ch = socketToChannel.get(socket);
if (ch == null) {
System.out.println("ks sendCB: channel not found for socket: " + socket);
return;
}
synchronized (channelToKey) {
SelectionKeyImpl sk = channelToKey.get(ch);
if (sk != null) {
if (handleSelect(sk)) {
latch.countDown();
}
}
}
}
}
@Override
public void connectionSucceeded(KKSSocket socket) {
refreshReady(socket);
}
@Override
public void connectionFailed(KKSSocket socket) {
refreshReady(socket);
}
@Override
public void dataSent(KKSSocket socket, long bytesSent) {
refreshReady(socket);
}
@Override
public void sendCB(KKSSocket socket, long bytesAvailable) {
refreshReady(socket);
}
@Override
public void onRecv(KKSSocket socket) {
refreshReady(socket);
}
@Override
public void newConnectionCreated(KKSSocket socket, KKSSocket newSocket, KKSAddress remoteaddress) {
refreshReady(socket);
}
@Override
public void normalClose(KKSSocket socket) {
wakeup();
}
@Override
public void errorClose(KKSSocket socket) {
wakeup();
}
}
protected final Map<KKSSocket, SelChImpl> socketToChannel = new HashMap<KKSSocket, SelChImpl>();
protected final Map<SelChImpl, SelectionKeyImpl> channelToKey = new HashMap<SelChImpl, SelectionKeyImpl>();
protected final SocketListener currListener = new SocketListener();
SelChImpl getChannelForSocket(KKSSocket s) {
synchronized (socketToChannel) {
return socketToChannel.get(s);
}
}
SelectionKeyImpl getSelKeyForChannel(KKSSocket s) {
synchronized (channelToKey) {
return channelToKey.get(s);
}
}
protected boolean markRead(SelectionKeyImpl impl) {
synchronized (impl) {
if (!impl.isValid()) return false;
impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_READ);
return selectedKeys.add(impl);
}
}
protected boolean markWrite(SelectionKeyImpl impl) {
synchronized (impl) {
if (!impl.isValid()) return false;
impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_WRITE);
return selectedKeys.add(impl);
}
}
protected boolean markAccept(SelectionKeyImpl impl) {
synchronized (impl) {
if (!impl.isValid()) return false;
impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_ACCEPT);
return selectedKeys.add(impl);
}
}
protected boolean markConnect(SelectionKeyImpl impl) {
synchronized (impl) {
if (!impl.isValid()) return false;
impl.nioReadyOps(impl.readyOps() | SelectionKeyImpl.OP_CONNECT);
return selectedKeys.add(impl);
}
}
/**
* @param provider
*/
protected KSelector(SelectorProvider provider) {
super(provider);
}
/* (non-Javadoc)
* @see kokunet.SelectorImpl#implClose()
*/
@Override
protected void implClose() throws IOException {
provider().getApp().printMessage("implClose: closed: " + closed);
synchronized (closeLock) {
if (closed) return;
closed = true;
for (SelectionKey sk : keys) {
provider().getApp().printMessage("dereg1");
deregister((AbstractSelectionKey)sk);
provider().getApp().printMessage("dereg2");
SelectableChannel selch = sk.channel();
if (!selch.isOpen() && !selch.isRegistered())
((SelChImpl)selch).kill();
}
implCloseInterrupt();
}
}
protected void implCloseInterrupt() {
wakeup();
}
private boolean handleSelect(SelectionKey k) {
synchronized (k) {
boolean notify = false;
if (!k.isValid()) {
k.cancel();
((SelectionKeyImpl)k).channel.socket().removeListener(currListener);
return false;
}
SelectionKeyImpl ski = (SelectionKeyImpl)k;
if ((ski.interestOps() & SelectionKeyImpl.OP_READ) != 0) {
if (ski.channel.socket().getRxAvailable() > 0) {
notify |= markRead(ski);
}
}
if ((ski.interestOps() & SelectionKeyImpl.OP_WRITE) != 0) {
if (ski.channel.socket().getTxAvailable() > 0) {
notify |= markWrite(ski);
}
}
if ((ski.interestOps() & SelectionKeyImpl.OP_CONNECT) != 0) {
if (!ski.channel.socket().isConnectionless()) {
IConnectionSocket cs = (IConnectionSocket)ski.channel.socket();
if (!ski.channel.socket().isAccepting() && !cs.isConnecting() && !cs.isConnected()) {
notify |= markConnect(ski);
}
}
}
if ((ski.interestOps() & SelectionKeyImpl.OP_ACCEPT) != 0) {
//provider().getApp().printMessage("accept check: ski: " + ski + ", connectionless: " + ski.channel.socket().isConnectionless() + ", listening: " + ski.channel.socket().isListening() + ", hasPendingConn: " + (ski.channel.socket().isConnectionless() ? "nope!" : ((IConnectionSocket)ski.channel.socket()).hasPendingConnections()));
if (!ski.channel.socket().isConnectionless() && ski.channel.socket().isListening()) {
IConnectionSocket cs = (IConnectionSocket)ski.channel.socket();
if (cs.hasPendingConnections()) {
notify |= markAccept(ski);
}
}
}
return notify;
}
}
private boolean handleSelect() {
boolean notify = false;
// get initial status
for (SelectionKey k : keys) {
notify |= handleSelect(k);
}
return notify;
}
/* (non-Javadoc)
* @see kokunet.SelectorImpl#doSelect(long)
*/
@Override
protected int doSelect(long timeout) throws IOException {
processDeregisterQueue();
long timestartedms = System.currentTimeMillis();
synchronized (selectedKeys) {
wakeup = false;
selecting = true;
try {
handleSelect();
if (!selectedKeys.isEmpty() || timeout == 0) {
return selectedKeys.size();
}
//TODO: useless op if we have keys available
for (SelectionKey key : keys) {
((SelectionKeyImpl)key).channel.socket().addListener(currListener);
}
try {
while (!wakeup && isOpen() && selectedKeys.isEmpty()) {
CountDownLatch latch = null;
synchronized (currListener) {
if (wakeup || !isOpen() || !selectedKeys.isEmpty()) {
break;
}
latch = currListener.newLatch();
}
try {
if (timeout > 0) {
long currtimems = System.currentTimeMillis();
long remainingMS = (timestartedms + timeout) - currtimems;
if (remainingMS > 0) {
latch.await(remainingMS, TimeUnit.MILLISECONDS);
} else {
break;
}
} else {
latch.await();
}
} catch (InterruptedException e) {
wakeup();
}
}
return selectedKeys.size();
} finally {
for (SelectionKey key : keys) {
((SelectionKeyImpl)key).channel.socket().removeListener(currListener);
}
processDeregisterQueue();
}
} finally {
selecting = false;
wakeup = false;
}
}
}
/* (non-Javadoc)
* @see kokunet.SelectorImpl#implRegister(kokunet.SelectionKeyImpl)
*/
@Override
protected void implRegister(SelectionKeyImpl ski) {
synchronized (closeLock) {
if (closed) throw new ClosedSelectorException();
synchronized (channelToKey) {
synchronized (socketToChannel) {
keys.add(ski);
socketToChannel.put(ski.channel.socket(), ski.channel);
channelToKey.put(ski.channel, ski);
}
}
}
}
/* (non-Javadoc)
* @see kokunet.SelectorImpl#implDereg(kokunet.SelectionKeyImpl)
*/
@Override
protected void implDereg(SelectionKeyImpl ski) throws IOException {
synchronized (channelToKey) {
synchronized (socketToChannel) {
keys.remove(ski);
socketToChannel.remove(ski.channel.socket());
channelToKey.remove(ski.channel);
SelectableChannel selch = ski.channel();
if (!selch.isOpen() && !selch.isRegistered())
((SelChImpl)selch).kill();
}
}
}
/* (non-Javadoc)
* @see kokunet.SelectorImpl#wakeup()
*/
@Override
public Selector wakeup() {
synchronized (selectedKeys) {
wakeup = true;
selectedKeys.notifyAll();
}
return this;
}
}
Извините, что не разместил SCEE в этом случае, но в этом случае это немного сложно. Любые предложения будут полезны.
Приветствия,
Chris