<NME> NioXnioWorker.java
<BEF> /*
 * JBoss, Home of Professional Open Source.
 * Copyright 2012 Red Hat, Inc., and individual contributors
 * as indicated by the @author tags.
 *
 * Licensed 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.xnio.nio;

import static org.xnio.IoUtils.safeClose;
import static org.xnio.nio.Log.log;

import java.io.Closeable;
import java.io.IOException;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.StandardProtocolFamily;
import java.nio.channels.DatagramChannel;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
import java.util.concurrent.locks.LockSupport;

import org.wildfly.common.net.CidrAddressTable;
import org.xnio.Bits;
import org.xnio.ChannelListener;
import org.xnio.ChannelListeners;
import org.xnio.ManagementRegistration;
import org.xnio.ClosedWorkerException;
import org.xnio.IoUtils;
import org.xnio.Option;
import org.xnio.OptionMap;
import org.xnio.Options;
import org.xnio.StreamConnection;
import org.xnio.XnioWorker;
import org.xnio.channels.AcceptingChannel;
import org.xnio.channels.MulticastMessageChannel;
import org.xnio.management.XnioServerMXBean;
import org.xnio.management.XnioWorkerMXBean;

/**
 * @author <a href="mailto:david.lloyd@redhat.com">David M. Lloyd</a>
 */
final class NioXnioWorker extends XnioWorker {

    private static final int CLOSE_REQ = (1 << 31);
    private static final int CLOSE_COMP = (1 << 30);
    private final long workerStackSize;

    private volatile int state = 1;

    private final WorkerThread[] workerThreads;
    private final WorkerThread acceptThread;
    private final NioWorkerMetrics metrics;

    @SuppressWarnings("unused")
    private volatile Thread shutdownWaiter;

    private static final AtomicReferenceFieldUpdater<NioXnioWorker, Thread> shutdownWaiterUpdater = AtomicReferenceFieldUpdater.newUpdater(NioXnioWorker.class, Thread.class, "shutdownWaiter");

    private static final AtomicIntegerFieldUpdater<NioXnioWorker> stateUpdater = AtomicIntegerFieldUpdater.newUpdater(NioXnioWorker.class, "state");

    @SuppressWarnings("deprecation")
    NioXnioWorker(final Builder builder) {
        super(builder);
        final NioXnio xnio = (NioXnio) builder.getXnio();
        final int threadCount = builder.getWorkerIoThreads();
        this.workerStackSize = builder.getWorkerStackSize();
        final String workerName = getName();
        WorkerThread[] workerThreads;
        workerThreads = new WorkerThread[threadCount];
        final ThreadGroup threadGroup = builder.getThreadGroup();
        final boolean markWorkerThreadAsDaemon = builder.isDaemon();
        boolean ok = false;
        try {
            for (int i = 0; i < threadCount; i++) {
                final Selector threadSelector;
                try {
                    threadSelector = xnio.mainSelectorCreator.open();
                } catch (IOException e) {
                    throw Log.log.unexpectedSelectorOpenProblem(e);
                }
                final WorkerThread workerThread = new WorkerThread(this, threadSelector, String.format("%s I/O-%d", workerName, Integer.valueOf(i + 1)), threadGroup, workerStackSize, i);
                // Mark as daemon if the Options.THREAD_DAEMON has been set
                if (markWorkerThreadAsDaemon) {
                    workerThread.setDaemon(true);
                }
                workerThreads[i] = workerThread;
            }
            final Selector threadSelector;
            try {
                threadSelector = xnio.mainSelectorCreator.open();
            } catch (IOException e) {
                throw Log.log.unexpectedSelectorOpenProblem(e);
            }
            acceptThread = new WorkerThread(this, threadSelector, String.format("%s Accept", workerName), threadGroup, workerStackSize, threadCount);
            if (markWorkerThreadAsDaemon) {
                acceptThread.setDaemon(true);
            }
            ok = true;
        } finally {
            if (! ok) {
                for (WorkerThread worker : workerThreads) {
                    if (worker != null) safeClose(worker.getSelector());
                }
            }
        }
        this.workerThreads = workerThreads;
        this.metrics = new NioWorkerMetrics(workerName);
        metrics.register();
    }

    void start() {
        for (WorkerThread worker : workerThreads) {
            openResourceUnconditionally();
            worker.start();
        }
        openResourceUnconditionally();
        acceptThread.start();
    }

    protected CidrAddressTable<InetSocketAddress> getBindAddressTable() {
        return super.getBindAddressTable();
    }

    protected WorkerThread chooseThread() {
        return getIoThread(ThreadLocalRandom.current().nextInt());
    }

    public WorkerThread getIoThread(final int hashCode) {

    private static final WorkerThread[] NO_WORKERS = new WorkerThread[0];

    WorkerThread choose() {
        final WorkerThread[] write = writeWorkers;
        final WorkerThread[] read = readWorkers;
        final int writeLength = write.length;
        final int readLength = read.length;
        if (writeLength == 0) {
            return choose(false);
        }
        if (readLength == 0) {
            return choose(true);
        }
        final Random random = IoUtils.getThreadLocalRandom();
        final int idx = random.nextInt(writeLength + readLength);
        return idx >= readLength ? write[idx - readLength] : read[idx];
    }

    WorkerThread chooseOptional(final boolean write) {
        final WorkerThread[] orig = write ? writeWorkers : readWorkers;
        final int length = orig.length;
            return workerThreads[0];
        }
        return workerThreads[Math.abs(hashCode % length)];
    }

    public int getIoThreadCount() {
        return workerThreads.length;
    }

    WorkerThread[] getAll() {
        return workerThreads;
    }

    protected AcceptingChannel<StreamConnection> createTcpConnectionServer(final InetSocketAddress bindAddress, final ChannelListener<? super AcceptingChannel<StreamConnection>> acceptListener, final OptionMap optionMap) throws IOException {
        checkShutdown();
        boolean ok = false;
        final ServerSocketChannel channel = ServerSocketChannel.open();
        try {
            if (optionMap.contains(Options.RECEIVE_BUFFER)) channel.socket().setReceiveBufferSize(optionMap.get(Options.RECEIVE_BUFFER, -1));
            channel.socket().setReuseAddress(optionMap.get(Options.REUSE_ADDRESSES, true));
            channel.configureBlocking(false);
            if (optionMap.contains(Options.BACKLOG)) {
                channel.socket().bind(bindAddress, optionMap.get(Options.BACKLOG, 128));
            } else {
                channel.socket().bind(bindAddress);
            }
            if (false) {
                final NioTcpServer server = new NioTcpServer(this, channel, optionMap, false);
                server.setAcceptListener(acceptListener);
                ok = true;
                return server;
            } else {
                final QueuedNioTcpServer2 server = new QueuedNioTcpServer2(new NioTcpServer(this, channel, optionMap, true));
                server.setAcceptListener(acceptListener);
                ok = true;
                return server;
            }
        } finally {
            if (! ok) {
                IoUtils.safeClose(channel);
            }
        }
    }


    /** {@inheritDoc} */
    public MulticastMessageChannel createUdpServer(final InetSocketAddress bindAddress, final ChannelListener<? super MulticastMessageChannel> bindListener, final OptionMap optionMap) throws IOException {
        checkShutdown();
        final DatagramChannel channel;
        if (bindAddress != null) {
            InetAddress address = bindAddress.getAddress();
            if (address instanceof Inet6Address) {
                channel = DatagramChannel.open(StandardProtocolFamily.INET6);
            } else {
                channel = DatagramChannel.open(StandardProtocolFamily.INET);
            }
        } else {
            channel = DatagramChannel.open();
        }
        channel.configureBlocking(false);
        if (optionMap.contains(Options.BROADCAST)) channel.socket().setBroadcast(optionMap.get(Options.BROADCAST, false));
        if (optionMap.contains(Options.IP_TRAFFIC_CLASS)) channel.socket().setTrafficClass(optionMap.get(Options.IP_TRAFFIC_CLASS, -1));
        if (optionMap.contains(Options.RECEIVE_BUFFER)) channel.socket().setReceiveBufferSize(optionMap.get(Options.RECEIVE_BUFFER, -1));
        channel.socket().setReuseAddress(optionMap.get(Options.REUSE_ADDRESSES, true));
        if (optionMap.contains(Options.SEND_BUFFER)) channel.socket().setSendBufferSize(optionMap.get(Options.SEND_BUFFER, -1));
        channel.socket().bind(bindAddress);
        final NioUdpChannel udpChannel = new NioUdpChannel(this, channel);
        ChannelListeners.invokeChannelListener(udpChannel, bindListener);
        return udpChannel;
    }

    public boolean isShutdown() {
        return (state & CLOSE_REQ) != 0;
    }

    public boolean isTerminated() {
        return (state & CLOSE_COMP) != 0;
    }

    /**
     * Open a resource unconditionally (i.e. accepting a connection on an open server).
     */
    void openResourceUnconditionally() {
        int oldState = stateUpdater.getAndIncrement(this);
        if (log.isTraceEnabled()) {
            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));
        }
    }

    void checkShutdown() throws ClosedWorkerException {
        if (isShutdown())
            throw log.workerShutDown(this);
    }

    void closeResource() {
        int oldState = stateUpdater.decrementAndGet(this);
        if (log.isTraceEnabled()) {
            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState + 1), Integer.valueOf(oldState));
        }
        while (oldState == CLOSE_REQ) {
            if (stateUpdater.compareAndSet(this, CLOSE_REQ, CLOSE_REQ | CLOSE_COMP)) {
                log.tracef("CAS %s %08x -> %08x (close complete)", this, Integer.valueOf(CLOSE_REQ), Integer.valueOf(CLOSE_REQ | CLOSE_COMP));
                safeUnpark(shutdownWaiterUpdater.getAndSet(this, null));
                final Runnable task = getTerminationTask();
                if (task != null) try {
                    task.run();
                } catch (Throwable ignored) {}
                return;
            }
            oldState = state;
        }
    }

    public void shutdown() {
        int oldState;
        oldState = state;
        while ((oldState & CLOSE_REQ) == 0) {
            // need to do the close ourselves...
            if (! stateUpdater.compareAndSet(this, oldState, oldState | CLOSE_REQ)) {
                // changed in the meantime
                oldState = state;
                continue;
            }
            log.tracef("Initiating shutdown of %s", this);
            for (WorkerThread worker : workerThreads) {
                worker.shutdown();
            }
            acceptThread.shutdown();
            shutDownTaskPool();
            return;
        }
        log.tracef("Idempotent shutdown of %s", this);
        return;
    }

    public List<Runnable> shutdownNow() {
        shutdown();
        return shutDownTaskPoolNow();
    }

    public boolean awaitTermination(final long timeout, final TimeUnit unit) throws InterruptedException {
        int oldState = state;
        if (Bits.allAreSet(oldState, CLOSE_COMP)) {
            return true;
        }
        long then = System.nanoTime();
        long duration = unit.toNanos(timeout);
        final Thread myThread = Thread.currentThread();
        while (Bits.allAreClear(oldState = state, CLOSE_COMP)) {
            final Thread oldThread = shutdownWaiterUpdater.getAndSet(this, myThread);
            try {
                if (Bits.allAreSet(oldState = state, CLOSE_COMP)) {
                    break;
                }
                LockSupport.parkNanos(this, duration);
                if (Thread.interrupted()) {
                    throw new InterruptedException();
                }
                long now = System.nanoTime();
                duration -= now - then;
                if (duration < 0L) {
                    oldState = state;
                    break;
                }
            } finally {
                safeUnpark(oldThread);
            }
        }
        return Bits.allAreSet(oldState, CLOSE_COMP);
    }

    public void awaitTermination() throws InterruptedException {
        int oldState = state;
        if (Bits.allAreSet(oldState, CLOSE_COMP)) {
            return;
        }
        final Thread myThread = Thread.currentThread();
        while (Bits.allAreClear(state, CLOSE_COMP)) {
            final Thread oldThread = shutdownWaiterUpdater.getAndSet(this, myThread);
            try {
                if (Bits.allAreSet(state, CLOSE_COMP)) {
                    break;
                }
                LockSupport.park(this);
                if (Thread.interrupted()) {
                    throw new InterruptedException();
                }
            } finally {
                safeUnpark(oldThread);
            }
        }
    }

    private static void safeUnpark(final Thread waiter) {
        if (waiter != null) LockSupport.unpark(waiter);
    }

    protected void taskPoolTerminated() {
        safeClose(metrics);
        closeResource();
    }

    @Override
    public <T> T getOption(Option<T> option) throws IOException {
        if (option.equals(Options.WORKER_IO_THREADS)) {
            return option.cast(workerThreads.length);
        } else if (option.equals(Options.STACK_SIZE)) {
            return option.cast(workerStackSize);
        } else {
            return super.getOption(option);
        }
    }

    public NioXnio getXnio() {
        return (NioXnio) super.getXnio();
    }

    WorkerThread getAcceptThread() {
        return acceptThread;
    }

    @Override
    public XnioWorkerMXBean getMXBean() {
        return metrics;
    }

    @Override
    protected ManagementRegistration registerServerMXBean(XnioServerMXBean serverMXBean) {
        return metrics.registerServerMXBean(serverMXBean);
    }

    private class NioWorkerMetrics implements XnioWorkerMXBean,Closeable {
        private final String workerName;
        private final CopyOnWriteArrayList<XnioServerMXBean> serverMetrics = new CopyOnWriteArrayList<>();
        private Closeable mbeanHandle;

        private NioWorkerMetrics(String workerName) {
            this.workerName = workerName;
        }

        public String getProviderName() {
            return "nio";
        }

        public String getName() {
            return workerName;
        }

        public boolean isShutdownRequested() {
            return isShutdown();
        }

        public int getCoreWorkerPoolSize() {
            return NioXnioWorker.this.getCoreWorkerPoolSize();
        }

        public int getMaxWorkerPoolSize() {
            return NioXnioWorker.this.getMaxWorkerPoolSize();
        }

        public int getWorkerPoolSize() {
            return NioXnioWorker.this.getWorkerPoolSize();
        }

        public int getBusyWorkerThreadCount() {
            return NioXnioWorker.this.getBusyWorkerThreadCount();
        }

        public int getIoThreadCount() {
            return NioXnioWorker.this.getIoThreadCount();
        }

        public int getWorkerQueueSize() {
            return NioXnioWorker.this.getWorkerQueueSize();
        }

        private ManagementRegistration registerServerMXBean(XnioServerMXBean serverMXBean){
            serverMetrics.addIfAbsent(serverMXBean);
            final Closeable handle = NioXnio.register(serverMXBean);
            return () -> {
                serverMetrics.remove(serverMXBean);
                safeClose(handle);
            };
        }

        public Set<XnioServerMXBean> getServerMXBeans() {
            return new LinkedHashSet<>(serverMetrics);
        }
        private void register(){
            this.mbeanHandle = NioXnio.register(this);
        }

        @Override
        public void close() throws IOException {
            safeClose(mbeanHandle);
            serverMetrics.clear();
        }
    }
}
        }
    }

    /**
     * Open a resource.  Must be matched with a corresponding {@code closeResource()}.
     *
     * @throws ClosedWorkerException if the worker is closed
     */
    void openResource() throws ClosedWorkerException {
        int oldState;
        do {
            oldState = state;
            if ((oldState & CLOSE_REQ) != 0) {
                throw new ClosedWorkerException("Worker is shutting down");
            }
        } while (! stateUpdater.compareAndSet(this, oldState, oldState + 1));
        if (log.isTraceEnabled()) {
            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));
        }
    }

    void closeResource() {
        int oldState = stateUpdater.decrementAndGet(this);

<MSG> Unreachable code cleanup.

<DFF> @@ -150,22 +150,6 @@ final class NioXnioWorker extends XnioWorker {
 
     private static final WorkerThread[] NO_WORKERS = new WorkerThread[0];
 
-    WorkerThread choose() {
-        final WorkerThread[] write = writeWorkers;
-        final WorkerThread[] read = readWorkers;
-        final int writeLength = write.length;
-        final int readLength = read.length;
-        if (writeLength == 0) {
-            return choose(false);
-        }
-        if (readLength == 0) {
-            return choose(true);
-        }
-        final Random random = IoUtils.getThreadLocalRandom();
-        final int idx = random.nextInt(writeLength + readLength);
-        return idx >= readLength ? write[idx - readLength] : read[idx];
-    }
-
     WorkerThread chooseOptional(final boolean write) {
         final WorkerThread[] orig = write ? writeWorkers : readWorkers;
         final int length = orig.length;
@@ -568,23 +552,6 @@ final class NioXnioWorker extends XnioWorker {
         }
     }
 
-    /**
-     * Open a resource.  Must be matched with a corresponding {@code closeResource()}.
-     *
-     * @throws ClosedWorkerException if the worker is closed
-     */
-    void openResource() throws ClosedWorkerException {
-        int oldState;
-        do {
-            oldState = state;
-            if ((oldState & CLOSE_REQ) != 0) {
-                throw new ClosedWorkerException("Worker is shutting down");
-            }
-        } while (! stateUpdater.compareAndSet(this, oldState, oldState + 1));
-        if (log.isTraceEnabled()) {
-            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));
-        }
-    }
 
     void closeResource() {
         int oldState = stateUpdater.decrementAndGet(this);

{'commit': 'diff --git a/nio-impl/src/main/java/org/xnio/nio/NioHandle.java b/nio-impl/src/main/java/org/xnio/nio/NioHandle.java\nindex db6fa2df..f9247bfb 100644\n--- a/nio-impl/src/main/java/org/xnio/nio/NioHandle.java\n+++ b/nio-impl/src/main/java/org/xnio/nio/NioHandle.java\n@@ -43,10 +43,6 @@ final class NioHandle<C extends Channel> implements Runnable {\n         this.channel = channel;\n     }\n \n-    SelectionKey getSelectionKey() {\n-        return selectionKey;\n-    }\n-\n     WorkerThread getWorkerThread() {\n         return workerThread;\n     }\ndiff --git a/nio-impl/src/main/java/org/xnio/nio/NioPipeChannel.java b/nio-impl/src/main/java/org/xnio/nio/NioPipeChannel.java\nindex d44c62f6..545cea8c 100644\n--- a/nio-impl/src/main/java/org/xnio/nio/NioPipeChannel.java\n+++ b/nio-impl/src/main/java/org/xnio/nio/NioPipeChannel.java\n@@ -110,10 +110,13 @@ final class NioPipeChannel extends AbstractNioStreamChannel<NioPipeChannel> {\n                 sourceChannel.close();\n             } finally {\n                 cancelReadKey();\n-            } else if (old == 0x02) try {\n-                sinkChannel.close();\n-            } finally {\n-                cancelWriteKey();\n+            } else {\n+                assert old == 0x02;\n+                try {\n+                    sinkChannel.close();\n+                } finally {\n+                    cancelWriteKey();\n+                }\n             }\n         } finally {\n             invokeCloseHandler();\ndiff --git a/nio-impl/src/main/java/org/xnio/nio/NioTcpServer.java b/nio-impl/src/main/java/org/xnio/nio/NioTcpServer.java\nindex afb1016d..9e9e79b5 100644\n--- a/nio-impl/src/main/java/org/xnio/nio/NioTcpServer.java\n+++ b/nio-impl/src/main/java/org/xnio/nio/NioTcpServer.java\n@@ -341,7 +341,7 @@ final class NioTcpServer extends AbstractNioChannel<NioTcpServer> implements Acc\n                 newVal |= CONN_SUSPENDING | CONN_FULL;\n             }\n         } while (! connectionStatusUpdater.compareAndSet(this, oldVal, newVal));\n-        boolean wasSuspended = anyAreSet(oldVal, CONN_FULL) || allAreClear(oldVal, CONN_RESUMED);\n+        boolean wasSuspended = allAreClear(oldVal, CONN_RESUMED);\n         boolean doSuspend = ! wasSuspended && allAreClear(oldVal, CONN_SUSPENDING) && allAreSet(newVal, CONN_FULL | CONN_SUSPENDING);\n         final SocketChannel accepted;\n         try {\n@@ -382,10 +382,9 @@ final class NioTcpServer extends AbstractNioChannel<NioTcpServer> implements Acc\n         }\n         if (doSuspend) {\n             // handle suspend\n-            if (allAreSet(oldVal, CONN_RESUMED)) {\n-                // we were previously resumed, so stop calling accept handlers for now\n-                doResume(0);\n-            }\n+            assert allAreSet(oldVal, CONN_RESUMED);\n+            // we were previously resumed, so stop calling accept handlers for now\n+            doResume(0);\n             // now attempt to synchronize the connection state with the new suspend state\n             synchronizeConnectionState(oldVal, doSuspend);\n         }\ndiff --git a/nio-impl/src/main/java/org/xnio/nio/NioUdpChannel.java b/nio-impl/src/main/java/org/xnio/nio/NioUdpChannel.java\nindex 5fc20db0..b7c72b05 100644\n--- a/nio-impl/src/main/java/org/xnio/nio/NioUdpChannel.java\n+++ b/nio-impl/src/main/java/org/xnio/nio/NioUdpChannel.java\n@@ -255,18 +255,14 @@ class NioUdpChannel extends AbstractNioChannel<NioUdpChannel> implements Multica\n     public void wakeupReads() {\n         resumeReads();\n         final NioHandle<NioUdpChannel> readHandle = this.readHandle;\n-        if (readHandle == null) {\n-            throw new IllegalArgumentException("No thread configured");\n-        }\n+        assert readHandle != null;\n         readHandle.execute();\n     }\n \n     public void wakeupWrites() {\n         resumeWrites();\n         final NioHandle<NioUdpChannel> writeHandle = this.writeHandle;\n-        if (writeHandle == null) {\n-            throw new IllegalArgumentException("No thread configured");\n-        }\n+        assert writeHandle != null;\n         writeHandle.execute();\n     }\n \ndiff --git a/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java b/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java\nindex 15e8eaf4..ea4b245c 100644\n--- a/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java\n+++ b/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java\n@@ -150,22 +150,6 @@ final class NioXnioWorker extends XnioWorker {\n \n     private static final WorkerThread[] NO_WORKERS = new WorkerThread[0];\n \n-    WorkerThread choose() {\n-        final WorkerThread[] write = writeWorkers;\n-        final WorkerThread[] read = readWorkers;\n-        final int writeLength = write.length;\n-        final int readLength = read.length;\n-        if (writeLength == 0) {\n-            return choose(false);\n-        }\n-        if (readLength == 0) {\n-            return choose(true);\n-        }\n-        final Random random = IoUtils.getThreadLocalRandom();\n-        final int idx = random.nextInt(writeLength + readLength);\n-        return idx >= readLength ? write[idx - readLength] : read[idx];\n-    }\n-\n     WorkerThread chooseOptional(final boolean write) {\n         final WorkerThread[] orig = write ? writeWorkers : readWorkers;\n         final int length = orig.length;\n@@ -568,23 +552,6 @@ final class NioXnioWorker extends XnioWorker {\n         }\n     }\n \n-    /**\n-     * Open a resource.  Must be matched with a corresponding {@code closeResource()}.\n-     *\n-     * @throws ClosedWorkerException if the worker is closed\n-     */\n-    void openResource() throws ClosedWorkerException {\n-        int oldState;\n-        do {\n-            oldState = state;\n-            if ((oldState & CLOSE_REQ) != 0) {\n-                throw new ClosedWorkerException("Worker is shutting down");\n-            }\n-        } while (! stateUpdater.compareAndSet(this, oldState, oldState + 1));\n-        if (log.isTraceEnabled()) {\n-            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));\n-        }\n-    }\n \n     void closeResource() {\n         int oldState = stateUpdater.decrementAndGet(this);\n', 'commit_subject': 'Unreachable code cleanup.', 'repo_name': 'xnio/xnio', 'committer': {'name': 'Flavia Rainone', 'email': 'b92a79ee58a09a40619fea32c08e5143af6f49fa@jboss.com', 'time_sec': '1344607164', 'tz_offset': '-180', 'date': {'seconds': '1344607164'}}, 'commit_tree': '3f59a8d6d6b0f0369b5fdfd5c2432ca834999e84', 'commit_message': 'Unreachable code cleanup.\n', 'language_name': 'Java', 'license': 'apache-2.0', 'hunks': '@@ -150,22 +150,6 @@ final class NioXnioWorker extends XnioWorker {\n \n     private static final WorkerThread[] NO_WORKERS = new WorkerThread[0];\n \n-    WorkerThread choose() {\n-        final WorkerThread[] write = writeWorkers;\n-        final WorkerThread[] read = readWorkers;\n-        final int writeLength = write.length;\n-        final int readLength = read.length;\n-        if (writeLength == 0) {\n-            return choose(false);\n-        }\n-        if (readLength == 0) {\n-            return choose(true);\n-        }\n-        final Random random = IoUtils.getThreadLocalRandom();\n-        final int idx = random.nextInt(writeLength + readLength);\n-        return idx >= readLength ? write[idx - readLength] : read[idx];\n-    }\n-\n     WorkerThread chooseOptional(final boolean write) {\n         final WorkerThread[] orig = write ? writeWorkers : readWorkers;\n         final int length = orig.length;\n@@ -568,23 +552,6 @@ final class NioXnioWorker extends XnioWorker {\n         }\n     }\n \n-    /**\n-     * Open a resource.  Must be matched with a corresponding {@code closeResource()}.\n-     *\n-     * @throws ClosedWorkerException if the worker is closed\n-     */\n-    void openResource() throws ClosedWorkerException {\n-        int oldState;\n-        do {\n-            oldState = state;\n-            if ((oldState & CLOSE_REQ) != 0) {\n-                throw new ClosedWorkerException("Worker is shutting down");\n-            }\n-        } while (! stateUpdater.compareAndSet(this, oldState, oldState + 1));\n-        if (log.isTraceEnabled()) {\n-            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));\n-        }\n-    }\n \n     void closeResource() {\n         int oldState = stateUpdater.decrementAndGet(this);\n', 'addition_count': 0, 'deletion_count': 33, 'src_file': 'a/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java', 'tgt_file': 'b/nio-impl/src/main/java/org/xnio/nio/NioXnioWorker.java', 'file_extension': '.java', 'before_file': '/*\n * JBoss, Home of Professional Open Source.\n * Copyright 2012 Red Hat, Inc., and individual contributors\n * as indicated by the @author tags.\n *\n * Licensed under the Apache License, Version 2.0 (the "License");\n * you may not use this file except in compliance with the License.\n * You may obtain a copy of the License at\n *\n *     http://www.apache.org/licenses/LICENSE-2.0\n *\n * Unless required by applicable law or agreed to in writing, software\n * distributed under the License is distributed on an "AS IS" BASIS,\n * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n * See the License for the specific language governing permissions and\n * limitations under the License.\n */\n\npackage org.xnio.nio;\n\nimport static org.xnio.IoUtils.safeClose;\nimport static org.xnio.nio.Log.log;\n\nimport java.io.Closeable;\nimport java.io.IOException;\nimport java.net.Inet6Address;\nimport java.net.InetAddress;\nimport java.net.InetSocketAddress;\nimport java.net.StandardProtocolFamily;\nimport java.nio.channels.DatagramChannel;\nimport java.nio.channels.Selector;\nimport java.nio.channels.ServerSocketChannel;\nimport java.util.LinkedHashSet;\nimport java.util.List;\nimport java.util.Set;\nimport java.util.concurrent.CopyOnWriteArrayList;\nimport java.util.concurrent.ThreadLocalRandom;\nimport java.util.concurrent.TimeUnit;\nimport java.util.concurrent.atomic.AtomicIntegerFieldUpdater;\nimport java.util.concurrent.atomic.AtomicReferenceFieldUpdater;\nimport java.util.concurrent.locks.LockSupport;\n\nimport org.wildfly.common.net.CidrAddressTable;\nimport org.xnio.Bits;\nimport org.xnio.ChannelListener;\nimport org.xnio.ChannelListeners;\nimport org.xnio.ManagementRegistration;\nimport org.xnio.ClosedWorkerException;\nimport org.xnio.IoUtils;\nimport org.xnio.Option;\nimport org.xnio.OptionMap;\nimport org.xnio.Options;\nimport org.xnio.StreamConnection;\nimport org.xnio.XnioWorker;\nimport org.xnio.channels.AcceptingChannel;\nimport org.xnio.channels.MulticastMessageChannel;\nimport org.xnio.management.XnioServerMXBean;\nimport org.xnio.management.XnioWorkerMXBean;\n\n/**\n * @author <a href="mailto:david.lloyd@redhat.com">David M. Lloyd</a>\n */\nfinal class NioXnioWorker extends XnioWorker {\n\n    private static final int CLOSE_REQ = (1 << 31);\n    private static final int CLOSE_COMP = (1 << 30);\n    private final long workerStackSize;\n\n    private volatile int state = 1;\n\n    private final WorkerThread[] workerThreads;\n    private final WorkerThread acceptThread;\n    private final NioWorkerMetrics metrics;\n\n    @SuppressWarnings("unused")\n    private volatile Thread shutdownWaiter;\n\n    private static final AtomicReferenceFieldUpdater<NioXnioWorker, Thread> shutdownWaiterUpdater = AtomicReferenceFieldUpdater.newUpdater(NioXnioWorker.class, Thread.class, "shutdownWaiter");\n\n    private static final AtomicIntegerFieldUpdater<NioXnioWorker> stateUpdater = AtomicIntegerFieldUpdater.newUpdater(NioXnioWorker.class, "state");\n\n    @SuppressWarnings("deprecation")\n    NioXnioWorker(final Builder builder) {\n        super(builder);\n        final NioXnio xnio = (NioXnio) builder.getXnio();\n        final int threadCount = builder.getWorkerIoThreads();\n        this.workerStackSize = builder.getWorkerStackSize();\n        final String workerName = getName();\n        WorkerThread[] workerThreads;\n        workerThreads = new WorkerThread[threadCount];\n        final ThreadGroup threadGroup = builder.getThreadGroup();\n        final boolean markWorkerThreadAsDaemon = builder.isDaemon();\n        boolean ok = false;\n        try {\n            for (int i = 0; i < threadCount; i++) {\n                final Selector threadSelector;\n                try {\n                    threadSelector = xnio.mainSelectorCreator.open();\n                } catch (IOException e) {\n                    throw Log.log.unexpectedSelectorOpenProblem(e);\n                }\n                final WorkerThread workerThread = new WorkerThread(this, threadSelector, String.format("%s I/O-%d", workerName, Integer.valueOf(i + 1)), threadGroup, workerStackSize, i);\n                // Mark as daemon if the Options.THREAD_DAEMON has been set\n                if (markWorkerThreadAsDaemon) {\n                    workerThread.setDaemon(true);\n                }\n                workerThreads[i] = workerThread;\n            }\n            final Selector threadSelector;\n            try {\n                threadSelector = xnio.mainSelectorCreator.open();\n            } catch (IOException e) {\n                throw Log.log.unexpectedSelectorOpenProblem(e);\n            }\n            acceptThread = new WorkerThread(this, threadSelector, String.format("%s Accept", workerName), threadGroup, workerStackSize, threadCount);\n            if (markWorkerThreadAsDaemon) {\n                acceptThread.setDaemon(true);\n            }\n            ok = true;\n        } finally {\n            if (! ok) {\n                for (WorkerThread worker : workerThreads) {\n                    if (worker != null) safeClose(worker.getSelector());\n                }\n            }\n        }\n        this.workerThreads = workerThreads;\n        this.metrics = new NioWorkerMetrics(workerName);\n        metrics.register();\n    }\n\n    void start() {\n        for (WorkerThread worker : workerThreads) {\n            openResourceUnconditionally();\n            worker.start();\n        }\n        openResourceUnconditionally();\n        acceptThread.start();\n    }\n\n    protected CidrAddressTable<InetSocketAddress> getBindAddressTable() {\n        return super.getBindAddressTable();\n    }\n\n    protected WorkerThread chooseThread() {\n        return getIoThread(ThreadLocalRandom.current().nextInt());\n    }\n\n    public WorkerThread getIoThread(final int hashCode) {\n\n    private static final WorkerThread[] NO_WORKERS = new WorkerThread[0];\n\n    WorkerThread choose() {\n        final WorkerThread[] write = writeWorkers;\n        final WorkerThread[] read = readWorkers;\n        final int writeLength = write.length;\n        final int readLength = read.length;\n        if (writeLength == 0) {\n            return choose(false);\n        }\n        if (readLength == 0) {\n            return choose(true);\n        }\n        final Random random = IoUtils.getThreadLocalRandom();\n        final int idx = random.nextInt(writeLength + readLength);\n        return idx >= readLength ? write[idx - readLength] : read[idx];\n    }\n\n    WorkerThread chooseOptional(final boolean write) {\n        final WorkerThread[] orig = write ? writeWorkers : readWorkers;\n        final int length = orig.length;\n            return workerThreads[0];\n        }\n        return workerThreads[Math.abs(hashCode % length)];\n    }\n\n    public int getIoThreadCount() {\n        return workerThreads.length;\n    }\n\n    WorkerThread[] getAll() {\n        return workerThreads;\n    }\n\n    protected AcceptingChannel<StreamConnection> createTcpConnectionServer(final InetSocketAddress bindAddress, final ChannelListener<? super AcceptingChannel<StreamConnection>> acceptListener, final OptionMap optionMap) throws IOException {\n        checkShutdown();\n        boolean ok = false;\n        final ServerSocketChannel channel = ServerSocketChannel.open();\n        try {\n            if (optionMap.contains(Options.RECEIVE_BUFFER)) channel.socket().setReceiveBufferSize(optionMap.get(Options.RECEIVE_BUFFER, -1));\n            channel.socket().setReuseAddress(optionMap.get(Options.REUSE_ADDRESSES, true));\n            channel.configureBlocking(false);\n            if (optionMap.contains(Options.BACKLOG)) {\n                channel.socket().bind(bindAddress, optionMap.get(Options.BACKLOG, 128));\n            } else {\n                channel.socket().bind(bindAddress);\n            }\n            if (false) {\n                final NioTcpServer server = new NioTcpServer(this, channel, optionMap, false);\n                server.setAcceptListener(acceptListener);\n                ok = true;\n                return server;\n            } else {\n                final QueuedNioTcpServer2 server = new QueuedNioTcpServer2(new NioTcpServer(this, channel, optionMap, true));\n                server.setAcceptListener(acceptListener);\n                ok = true;\n                return server;\n            }\n        } finally {\n            if (! ok) {\n                IoUtils.safeClose(channel);\n            }\n        }\n    }\n\n\n    /** {@inheritDoc} */\n    public MulticastMessageChannel createUdpServer(final InetSocketAddress bindAddress, final ChannelListener<? super MulticastMessageChannel> bindListener, final OptionMap optionMap) throws IOException {\n        checkShutdown();\n        final DatagramChannel channel;\n        if (bindAddress != null) {\n            InetAddress address = bindAddress.getAddress();\n            if (address instanceof Inet6Address) {\n                channel = DatagramChannel.open(StandardProtocolFamily.INET6);\n            } else {\n                channel = DatagramChannel.open(StandardProtocolFamily.INET);\n            }\n        } else {\n            channel = DatagramChannel.open();\n        }\n        channel.configureBlocking(false);\n        if (optionMap.contains(Options.BROADCAST)) channel.socket().setBroadcast(optionMap.get(Options.BROADCAST, false));\n        if (optionMap.contains(Options.IP_TRAFFIC_CLASS)) channel.socket().setTrafficClass(optionMap.get(Options.IP_TRAFFIC_CLASS, -1));\n        if (optionMap.contains(Options.RECEIVE_BUFFER)) channel.socket().setReceiveBufferSize(optionMap.get(Options.RECEIVE_BUFFER, -1));\n        channel.socket().setReuseAddress(optionMap.get(Options.REUSE_ADDRESSES, true));\n        if (optionMap.contains(Options.SEND_BUFFER)) channel.socket().setSendBufferSize(optionMap.get(Options.SEND_BUFFER, -1));\n        channel.socket().bind(bindAddress);\n        final NioUdpChannel udpChannel = new NioUdpChannel(this, channel);\n        ChannelListeners.invokeChannelListener(udpChannel, bindListener);\n        return udpChannel;\n    }\n\n    public boolean isShutdown() {\n        return (state & CLOSE_REQ) != 0;\n    }\n\n    public boolean isTerminated() {\n        return (state & CLOSE_COMP) != 0;\n    }\n\n    /**\n     * Open a resource unconditionally (i.e. accepting a connection on an open server).\n     */\n    void openResourceUnconditionally() {\n        int oldState = stateUpdater.getAndIncrement(this);\n        if (log.isTraceEnabled()) {\n            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));\n        }\n    }\n\n    void checkShutdown() throws ClosedWorkerException {\n        if (isShutdown())\n            throw log.workerShutDown(this);\n    }\n\n    void closeResource() {\n        int oldState = stateUpdater.decrementAndGet(this);\n        if (log.isTraceEnabled()) {\n            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState + 1), Integer.valueOf(oldState));\n        }\n        while (oldState == CLOSE_REQ) {\n            if (stateUpdater.compareAndSet(this, CLOSE_REQ, CLOSE_REQ | CLOSE_COMP)) {\n                log.tracef("CAS %s %08x -> %08x (close complete)", this, Integer.valueOf(CLOSE_REQ), Integer.valueOf(CLOSE_REQ | CLOSE_COMP));\n                safeUnpark(shutdownWaiterUpdater.getAndSet(this, null));\n                final Runnable task = getTerminationTask();\n                if (task != null) try {\n                    task.run();\n                } catch (Throwable ignored) {}\n                return;\n            }\n            oldState = state;\n        }\n    }\n\n    public void shutdown() {\n        int oldState;\n        oldState = state;\n        while ((oldState & CLOSE_REQ) == 0) {\n            // need to do the close ourselves...\n            if (! stateUpdater.compareAndSet(this, oldState, oldState | CLOSE_REQ)) {\n                // changed in the meantime\n                oldState = state;\n                continue;\n            }\n            log.tracef("Initiating shutdown of %s", this);\n            for (WorkerThread worker : workerThreads) {\n                worker.shutdown();\n            }\n            acceptThread.shutdown();\n            shutDownTaskPool();\n            return;\n        }\n        log.tracef("Idempotent shutdown of %s", this);\n        return;\n    }\n\n    public List<Runnable> shutdownNow() {\n        shutdown();\n        return shutDownTaskPoolNow();\n    }\n\n    public boolean awaitTermination(final long timeout, final TimeUnit unit) throws InterruptedException {\n        int oldState = state;\n        if (Bits.allAreSet(oldState, CLOSE_COMP)) {\n            return true;\n        }\n        long then = System.nanoTime();\n        long duration = unit.toNanos(timeout);\n        final Thread myThread = Thread.currentThread();\n        while (Bits.allAreClear(oldState = state, CLOSE_COMP)) {\n            final Thread oldThread = shutdownWaiterUpdater.getAndSet(this, myThread);\n            try {\n                if (Bits.allAreSet(oldState = state, CLOSE_COMP)) {\n                    break;\n                }\n                LockSupport.parkNanos(this, duration);\n                if (Thread.interrupted()) {\n                    throw new InterruptedException();\n                }\n                long now = System.nanoTime();\n                duration -= now - then;\n                if (duration < 0L) {\n                    oldState = state;\n                    break;\n                }\n            } finally {\n                safeUnpark(oldThread);\n            }\n        }\n        return Bits.allAreSet(oldState, CLOSE_COMP);\n    }\n\n    public void awaitTermination() throws InterruptedException {\n        int oldState = state;\n        if (Bits.allAreSet(oldState, CLOSE_COMP)) {\n            return;\n        }\n        final Thread myThread = Thread.currentThread();\n        while (Bits.allAreClear(state, CLOSE_COMP)) {\n            final Thread oldThread = shutdownWaiterUpdater.getAndSet(this, myThread);\n            try {\n                if (Bits.allAreSet(state, CLOSE_COMP)) {\n                    break;\n                }\n                LockSupport.park(this);\n                if (Thread.interrupted()) {\n                    throw new InterruptedException();\n                }\n            } finally {\n                safeUnpark(oldThread);\n            }\n        }\n    }\n\n    private static void safeUnpark(final Thread waiter) {\n        if (waiter != null) LockSupport.unpark(waiter);\n    }\n\n    protected void taskPoolTerminated() {\n        safeClose(metrics);\n        closeResource();\n    }\n\n    @Override\n    public <T> T getOption(Option<T> option) throws IOException {\n        if (option.equals(Options.WORKER_IO_THREADS)) {\n            return option.cast(workerThreads.length);\n        } else if (option.equals(Options.STACK_SIZE)) {\n            return option.cast(workerStackSize);\n        } else {\n            return super.getOption(option);\n        }\n    }\n\n    public NioXnio getXnio() {\n        return (NioXnio) super.getXnio();\n    }\n\n    WorkerThread getAcceptThread() {\n        return acceptThread;\n    }\n\n    @Override\n    public XnioWorkerMXBean getMXBean() {\n        return metrics;\n    }\n\n    @Override\n    protected ManagementRegistration registerServerMXBean(XnioServerMXBean serverMXBean) {\n        return metrics.registerServerMXBean(serverMXBean);\n    }\n\n    private class NioWorkerMetrics implements XnioWorkerMXBean,Closeable {\n        private final String workerName;\n        private final CopyOnWriteArrayList<XnioServerMXBean> serverMetrics = new CopyOnWriteArrayList<>();\n        private Closeable mbeanHandle;\n\n        private NioWorkerMetrics(String workerName) {\n            this.workerName = workerName;\n        }\n\n        public String getProviderName() {\n            return "nio";\n        }\n\n        public String getName() {\n            return workerName;\n        }\n\n        public boolean isShutdownRequested() {\n            return isShutdown();\n        }\n\n        public int getCoreWorkerPoolSize() {\n            return NioXnioWorker.this.getCoreWorkerPoolSize();\n        }\n\n        public int getMaxWorkerPoolSize() {\n            return NioXnioWorker.this.getMaxWorkerPoolSize();\n        }\n\n        public int getWorkerPoolSize() {\n            return NioXnioWorker.this.getWorkerPoolSize();\n        }\n\n        public int getBusyWorkerThreadCount() {\n            return NioXnioWorker.this.getBusyWorkerThreadCount();\n        }\n\n        public int getIoThreadCount() {\n            return NioXnioWorker.this.getIoThreadCount();\n        }\n\n        public int getWorkerQueueSize() {\n            return NioXnioWorker.this.getWorkerQueueSize();\n        }\n\n        private ManagementRegistration registerServerMXBean(XnioServerMXBean serverMXBean){\n            serverMetrics.addIfAbsent(serverMXBean);\n            final Closeable handle = NioXnio.register(serverMXBean);\n            return () -> {\n                serverMetrics.remove(serverMXBean);\n                safeClose(handle);\n            };\n        }\n\n        public Set<XnioServerMXBean> getServerMXBeans() {\n            return new LinkedHashSet<>(serverMetrics);\n        }\n        private void register(){\n            this.mbeanHandle = NioXnio.register(this);\n        }\n\n        @Override\n        public void close() throws IOException {\n            safeClose(mbeanHandle);\n            serverMetrics.clear();\n        }\n    }\n}\n        }\n    }\n\n    /**\n     * Open a resource.  Must be matched with a corresponding {@code closeResource()}.\n     *\n     * @throws ClosedWorkerException if the worker is closed\n     */\n    void openResource() throws ClosedWorkerException {\n        int oldState;\n        do {\n            oldState = state;\n            if ((oldState & CLOSE_REQ) != 0) {\n                throw new ClosedWorkerException("Worker is shutting down");\n            }\n        } while (! stateUpdater.compareAndSet(this, oldState, oldState + 1));\n        if (log.isTraceEnabled()) {\n            log.tracef("CAS %s %08x -> %08x", this, Integer.valueOf(oldState), Integer.valueOf(oldState + 1));\n        }\n    }\n\n    void closeResource() {\n        int oldState = stateUpdater.decrementAndGet(this);\n', 'message': 'Unreachable code cleanup.\n', 'author_name': 'Flavia Rainone', 'lang': 'java'}