#2875

pending completion

Build # #2875

Build Type

Pull #563

github-actions

Committed by

web-flow

Commit Message

Merge 430795bb9 into 19b20fefe

Pull Request Pull Request #563: Bump pgpainless

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

16372 of 41845 relevant lines covered (39.13%)

0.39 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

37.85

/smack-core/src/main/java/org/jivesoftware/smack/SmackReactor.java

/**
 *
 * Copyright 2018-2020 Florian Schmaus
 *
 * 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.jivesoftware.smack;

import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * The SmackReactor for non-blocking I/O.
 * <p>
 * Highlights include:
 * <ul>
 * <li>Multiple reactor threads</li>
 * <li>Scheduled actions</li>
 * </ul>
 *
 * <pre>
 *
 *           ) ) )
 *        ( ( (
 *      ) ) )
 *   (~~~~~~~~~)
 *    | Smack |
 *    |Reactor|
 *    I      _._
 *    I    /'   `\
 *    I   |       |
 *    f   |   |~~~~~~~~~~~~~~|
 *  .'    |   | #   #   #  # |
 * '______|___|___________###|
 * </pre>
 */
public class SmackReactor {

    private static final Logger LOGGER = Logger.getLogger(SmackReactor.class.getName());

    private static final int DEFAULT_REACTOR_THREAD_COUNT = 2;

    private static final int PENDING_SET_INTEREST_OPS_MAX_BATCH_SIZE = 1024;

    private static SmackReactor INSTANCE;

    static synchronized SmackReactor getInstance() {
        if (INSTANCE == null) {
            INSTANCE = new SmackReactor("DefaultReactor");
        }
        return INSTANCE;
    }

    private final Selector selector;
    private final String reactorName;

    private final List<Reactor> reactorThreads = Collections.synchronizedList(new ArrayList<>());

    private final DelayQueue<ScheduledAction> scheduledActions = new DelayQueue<>();

    private final Lock registrationLock = new ReentrantLock();

    /**
     * The semaphore protecting the handling of the actions. Note that it is
     * initialized with -1, which basically means that one thread will always do I/O using
     * select().
     */
    private final Semaphore actionsSemaphore = new Semaphore(-1, false);

    private final Queue<SelectionKey> pendingSelectionKeys = new ConcurrentLinkedQueue<>();

    private final Queue<SetInterestOps> pendingSetInterestOps = new ConcurrentLinkedQueue<>();

    SmackReactor(String reactorName) {
        this.reactorName = reactorName;

        try {
            selector = Selector.open();
        }
        catch (IOException e) {
            throw new IllegalStateException(e);
        }

        setReactorThreadCount(DEFAULT_REACTOR_THREAD_COUNT);
    }

    public SelectionKey registerWithSelector(SelectableChannel channel, int ops, ChannelSelectedCallback callback)
            throws ClosedChannelException {
        SelectionKeyAttachment selectionKeyAttachment = new SelectionKeyAttachment(callback);

        registrationLock.lock();
        try {
            selector.wakeup();
            return channel.register(selector, ops, selectionKeyAttachment);
        } finally {
            registrationLock.unlock();
        }
    }

    public void setInterestOps(SelectionKey selectionKey, int interestOps) {
        SetInterestOps setInterestOps = new SetInterestOps(selectionKey, interestOps);
        pendingSetInterestOps.add(setInterestOps);
        selector.wakeup();
    }

    private static final class SetInterestOps {
        private final SelectionKey selectionKey;
        private final int interestOps;

        private SetInterestOps(SelectionKey selectionKey, int interestOps) {
            this.selectionKey = selectionKey;
            this.interestOps = interestOps;
        }
    }

    ScheduledAction schedule(Runnable runnable, long delay, TimeUnit unit, ScheduledAction.Kind scheduledActionKind) {
        long releaseTimeEpoch = System.currentTimeMillis() + unit.toMillis(delay);
        Date releaseTimeDate = new Date(releaseTimeEpoch);
        ScheduledAction scheduledAction = new ScheduledAction(runnable, releaseTimeDate, this, scheduledActionKind);
        scheduledActions.add(scheduledAction);
        selector.wakeup();
        return scheduledAction;
    }

    /**
     * Cancels the scheduled action.
     *
     * @param scheduledAction the scheduled action to cancel.
     * @return <code>true</code> if the scheduled action was still pending and got removed, <code>false</code> otherwise.
     */
    boolean cancel(ScheduledAction scheduledAction) {
        return scheduledActions.remove(scheduledAction);
    }

    private class Reactor extends Thread {

        private volatile long shutdownRequestTimestamp = -1;

        @Override
        public void run() {
            try {
                reactorLoop();
            } finally {
                if (shutdownRequestTimestamp > 0) {
                    long shutDownDelay = System.currentTimeMillis() - shutdownRequestTimestamp;
                    LOGGER.info(this + " shut down after " + shutDownDelay + "ms");
                } else {
                    boolean contained = reactorThreads.remove(this);
                    assert contained;
                }
            }
        }

        private void reactorLoop() {
            // Loop until reactor shutdown was requested.
            while (shutdownRequestTimestamp < 0) {
                handleScheduledActionsOrPerformSelect();

                handlePendingSelectionKeys();
            }
        }

        @SuppressWarnings("LockNotBeforeTry")
        private void handleScheduledActionsOrPerformSelect() {
            ScheduledAction dueScheduledAction = null;

            boolean permitToHandleScheduledActions = actionsSemaphore.tryAcquire();
            if (permitToHandleScheduledActions) {
                try {
                    dueScheduledAction = scheduledActions.poll();
                } finally {
                    actionsSemaphore.release();
                }
            }

            if (dueScheduledAction != null) {
                dueScheduledAction.run();
                return;
            }

            int newSelectedKeysCount = 0;
            List<SelectionKey> selectedKeys;
            synchronized (selector) {
                ScheduledAction nextScheduledAction = scheduledActions.peek();

                long selectWait;
                if (nextScheduledAction == null) {
                    // There is no next scheduled action, wait indefinitely in select() or until another thread invokes
                    // selector.wakeup().
                    selectWait = 0;
                } else {
                    selectWait = nextScheduledAction.getTimeToDueMillis();
                    if (selectWait <= 0) {
                        // A scheduled action was just released and became ready to execute.
                        return;
                    }
                }

                // Before we call select, we handle the pending the interest Ops. This will not block since no other
                // thread is currently in select() at this time.
                // Note: This was put deliberately before the registration lock. It may cause more synchronization but
                // allows for more parallelism.
                // Hopefully that assumption is right.
                int myHandledPendingSetInterestOps = 0;
                for (SetInterestOps setInterestOps; (setInterestOps = pendingSetInterestOps.poll()) != null;) {
                    setInterestOpsCancelledKeySafe(setInterestOps.selectionKey, setInterestOps.interestOps);

                    if (myHandledPendingSetInterestOps++ >= PENDING_SET_INTEREST_OPS_MAX_BATCH_SIZE) {
                        // This thread has handled enough "set pending interest ops" requests. Wakeup another one to
                        // handle the remaining (if any).
                        selector.wakeup();
                        break;
                    }
                }

                // Ensure that a wakeup() in registerWithSelector() gives the corresponding
                // register() in the same method the chance to actually register the channel. In
                // other words: This construct ensures that there is never another select()
                // between a corresponding wakeup() and register() calls.
                // See also https://stackoverflow.com/a/1112809/194894
                registrationLock.lock();
                registrationLock.unlock();

                try {
                    newSelectedKeysCount = selector.select(selectWait);
                } catch (IOException e) {
                    LOGGER.log(Level.SEVERE, "IOException while using select()", e);
                    return;
                }

                if (newSelectedKeysCount == 0) {
                    return;
                }

                // Copy the selected-key set over to selectedKeys, remove the keys from the
                // selected key set and loose interest of the key OPs for the time being.
                // Note that we perform this operation in two steps in order to maximize the
                // timespan setRacing() is set.
                Set<SelectionKey> selectedKeySet = selector.selectedKeys();
                for (SelectionKey selectionKey : selectedKeySet) {
                    SelectionKeyAttachment selectionKeyAttachment = (SelectionKeyAttachment) selectionKey.attachment();
                    selectionKeyAttachment.setRacing();
                }
                for (SelectionKey selectionKey : selectedKeySet) {
                    setInterestOpsCancelledKeySafe(selectionKey, 0);
                }

                selectedKeys = new ArrayList<>(selectedKeySet.size());
                selectedKeys.addAll(selectedKeySet);
                selectedKeySet.clear();
            }

            int selectedKeysCount = selectedKeys.size();
            int currentReactorThreadCount = reactorThreads.size();
            int myKeyCount;
            if (selectedKeysCount > currentReactorThreadCount) {
                myKeyCount = selectedKeysCount / currentReactorThreadCount;
            } else {
                myKeyCount = selectedKeysCount;
            }

            final Level reactorSelectStatsLogLevel = Level.FINE;
            if (LOGGER.isLoggable(reactorSelectStatsLogLevel)) {
                LOGGER.log(reactorSelectStatsLogLevel,
                                "New selected key count: " + newSelectedKeysCount
                                + ". Total selected key count " + selectedKeysCount
                                + ". My key count: " + myKeyCount
                                + ". Current reactor thread count: " + currentReactorThreadCount);
            }

            Collection<SelectionKey> mySelectedKeys = new ArrayList<>(myKeyCount);
            Iterator<SelectionKey> it = selectedKeys.iterator();
            for (int i = 0; i < myKeyCount; i++) {
                SelectionKey selectionKey = it.next();
                mySelectedKeys.add(selectionKey);
            }
            while (it.hasNext()) {
                // Drain to pendingSelectionKeys.
                SelectionKey selectionKey = it.next();
                pendingSelectionKeys.add(selectionKey);
            }

            if (selectedKeysCount - myKeyCount > 0) {
                // There where pending selection keys: Wakeup another reactor thread to handle them.
                selector.wakeup();
            }

            handleSelectedKeys(mySelectedKeys);
        }

        private void handlePendingSelectionKeys() {
            final int pendingSelectionKeysSize = pendingSelectionKeys.size();
            if (pendingSelectionKeysSize == 0) {
                return;
            }

            int currentReactorThreadCount = reactorThreads.size();
            int myKeyCount = pendingSelectionKeysSize / currentReactorThreadCount;
            Collection<SelectionKey> selectedKeys = new ArrayList<>(myKeyCount);
            for (int i = 0; i < myKeyCount; i++) {
                SelectionKey selectionKey = pendingSelectionKeys.poll();
                if (selectionKey == null) {
                    // We lost a race and can abort here since the pendingSelectionKeys queue is empty.
                    break;
                }
                selectedKeys.add(selectionKey);
            }

            if (!pendingSelectionKeys.isEmpty()) {
                // There are more pending selection keys, wakeup a thread blocked in select() to handle them.
                selector.wakeup();
            }

            handleSelectedKeys(selectedKeys);
        }

        private void setInterestOpsCancelledKeySafe(SelectionKey selectionKey, int interestOps) {
            try {
                selectionKey.interestOps(interestOps);
            }
            catch (CancelledKeyException e) {
                final Level keyCancelledLogLevel = Level.FINER;
                if (LOGGER.isLoggable(keyCancelledLogLevel)) {
                    LOGGER.log(keyCancelledLogLevel, "Key '" + selectionKey + "' has been cancelled", e);
                }
            }
        }

        void requestShutdown() {
            shutdownRequestTimestamp = System.currentTimeMillis();
        }
    }

    private static void handleSelectedKeys(Collection<SelectionKey> selectedKeys) {
        for (SelectionKey selectionKey : selectedKeys) {
            SelectableChannel channel = selectionKey.channel();
            SelectionKeyAttachment selectionKeyAttachment = (SelectionKeyAttachment) selectionKey.attachment();
            ChannelSelectedCallback channelSelectedCallback = selectionKeyAttachment.channelSelectedCallback;
            channelSelectedCallback.onChannelSelected(channel, selectionKey);
        }
    }

    public interface ChannelSelectedCallback {
        void onChannelSelected(SelectableChannel channel, SelectionKey selectionKey);
    }

    public void setReactorThreadCount(int reactorThreadCount) {
        if (reactorThreadCount < 2) {
            throw new IllegalArgumentException("Must have at least two reactor threads, but you requested " + reactorThreadCount);
        }

        synchronized (reactorThreads) {
            int deltaThreads = reactorThreadCount - reactorThreads.size();
            if (deltaThreads > 0) {
                // Start new reactor thread. Note that we start the threads before we increase the permits of the
                // actionsSemaphore.
                for (int i = 0; i < deltaThreads; i++) {
                    Reactor reactor = new Reactor();
                    reactor.setDaemon(true);
                    reactor.setName("Smack " + reactorName + " Thread #" + i);
                    reactorThreads.add(reactor);
                    reactor.start();
                }

                actionsSemaphore.release(deltaThreads);
            } else {
                // Stop existing reactor threads. First we change the sign of deltaThreads, then we decrease the permits
                // of the actionsSemaphore *before* we signal the selected reactor threads that they should shut down.
                deltaThreads -= deltaThreads;

                for (int i = deltaThreads - 1; i > 0; i--) {
                    // Note that this could potentially block forever, starving on the unfair semaphore.
                    actionsSemaphore.acquireUninterruptibly();
                }

                for (int i = deltaThreads - 1; i > 0; i--) {
                    Reactor reactor = reactorThreads.remove(i);
                    reactor.requestShutdown();
                }

                selector.wakeup();
            }
        }
    }

    public static final class SelectionKeyAttachment {
        private final ChannelSelectedCallback channelSelectedCallback;
        private final AtomicBoolean reactorThreadRacing = new AtomicBoolean();

        private SelectionKeyAttachment(ChannelSelectedCallback channelSelectedCallback) {
            this.channelSelectedCallback = channelSelectedCallback;
        }

        private void setRacing() {
            // We use lazySet here since it is sufficient if the value does not become visible immediately.
            reactorThreadRacing.lazySet(true);
        }

        public void resetReactorThreadRacing() {
            reactorThreadRacing.set(false);
        }

        public boolean isReactorThreadRacing() {
            return reactorThreadRacing.get();
        }

    }
}

1	/**
2	*
3	* Copyright 2018-2020 Florian Schmaus
4	*
5	* Licensed under the Apache License, Version 2.0 (the "License");
6	* you may not use this file except in compliance with the License.
7	* You may obtain a copy of the License at
8	*
9	* http://www.apache.org/licenses/LICENSE-2.0
10	*
11	* Unless required by applicable law or agreed to in writing, software
12	* distributed under the License is distributed on an "AS IS" BASIS,
13	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	* See the License for the specific language governing permissions and
15	* limitations under the License.
16	*/
17	package org.jivesoftware.smack;
18
19	import java.io.IOException;
20	import java.nio.channels.CancelledKeyException;
21	import java.nio.channels.ClosedChannelException;
22	import java.nio.channels.SelectableChannel;
23	import java.nio.channels.SelectionKey;
24	import java.nio.channels.Selector;
25	import java.util.ArrayList;
26	import java.util.Collection;
27	import java.util.Collections;
28	import java.util.Date;
29	import java.util.Iterator;
30	import java.util.List;
31	import java.util.Queue;
32	import java.util.Set;
33	import java.util.concurrent.ConcurrentLinkedQueue;
34	import java.util.concurrent.DelayQueue;
35	import java.util.concurrent.Semaphore;
36	import java.util.concurrent.TimeUnit;
37	import java.util.concurrent.atomic.AtomicBoolean;
38	import java.util.concurrent.locks.Lock;
39	import java.util.concurrent.locks.ReentrantLock;
40	import java.util.logging.Level;
41	import java.util.logging.Logger;
42
43	/**
44	* The SmackReactor for non-blocking I/O.
45	* <p>
46	* Highlights include:
47	* <ul>
48	* <li>Multiple reactor threads</li>
49	* <li>Scheduled actions</li>
50	* </ul>
51	*
52	* <pre>
53	*
54	* ) ) )
55	* ( ( (
56	* ) ) )
57	* (~~~~~~~~~)
58	* \| Smack \|
59	* \|Reactor\|
60	* I _._
61	* I /' `\
62	* I \| \|
63	* f \| \|~~~~~~~~~~~~~~\|
64	* .' \| \| # # # # \|
65	* '______\|___\|___________###\|
66	* </pre>
67	*/
68	public class SmackReactor {
69
70	private static final Logger LOGGER = Logger.getLogger(SmackReactor.class.getName());	1✔
71
72	private static final int DEFAULT_REACTOR_THREAD_COUNT = 2;
73
74	private static final int PENDING_SET_INTEREST_OPS_MAX_BATCH_SIZE = 1024;
75
76	private static SmackReactor INSTANCE;
77
78	static synchronized SmackReactor getInstance() {
79	if (INSTANCE == null) {	1✔
80	INSTANCE = new SmackReactor("DefaultReactor");	1✔
81	}
82	return INSTANCE;	1✔
83	}
84
85	private final Selector selector;
86	private final String reactorName;
87
88	private final List<Reactor> reactorThreads = Collections.synchronizedList(new ArrayList<>());	1✔
89
90	private final DelayQueue<ScheduledAction> scheduledActions = new DelayQueue<>();	1✔
91
92	private final Lock registrationLock = new ReentrantLock();	1✔
93
94	/**
95	* The semaphore protecting the handling of the actions. Note that it is
96	* initialized with -1, which basically means that one thread will always do I/O using
97	* select().
98	*/
99	private final Semaphore actionsSemaphore = new Semaphore(-1, false);	1✔
100
101	private final Queue<SelectionKey> pendingSelectionKeys = new ConcurrentLinkedQueue<>();	1✔
102
103	private final Queue<SetInterestOps> pendingSetInterestOps = new ConcurrentLinkedQueue<>();	1✔
104
105	SmackReactor(String reactorName) {	1✔
106	this.reactorName = reactorName;	1✔
107
108	try {
109	selector = Selector.open();	1✔
110	}
111	catch (IOException e) {	×
112	throw new IllegalStateException(e);	×
113	}	1✔
114
115	setReactorThreadCount(DEFAULT_REACTOR_THREAD_COUNT);	1✔
116	}	1✔
117
118	public SelectionKey registerWithSelector(SelectableChannel channel, int ops, ChannelSelectedCallback callback)
119	throws ClosedChannelException {
120	SelectionKeyAttachment selectionKeyAttachment = new SelectionKeyAttachment(callback);	×
121
122	registrationLock.lock();	×
123	try {
124	selector.wakeup();	×
125	return channel.register(selector, ops, selectionKeyAttachment);	×
126	} finally {
127	registrationLock.unlock();	×
128	}
129	}
130
131	public void setInterestOps(SelectionKey selectionKey, int interestOps) {
132	SetInterestOps setInterestOps = new SetInterestOps(selectionKey, interestOps);	×
133	pendingSetInterestOps.add(setInterestOps);	×
134	selector.wakeup();	×
135	}	×
136
137	private static final class SetInterestOps {
138	private final SelectionKey selectionKey;
139	private final int interestOps;
140
141	private SetInterestOps(SelectionKey selectionKey, int interestOps) {	×
142	this.selectionKey = selectionKey;	×
143	this.interestOps = interestOps;	×
144	}	×
145	}
146
147	ScheduledAction schedule(Runnable runnable, long delay, TimeUnit unit, ScheduledAction.Kind scheduledActionKind) {
148	long releaseTimeEpoch = System.currentTimeMillis() + unit.toMillis(delay);	1✔
149	Date releaseTimeDate = new Date(releaseTimeEpoch);	1✔
150	ScheduledAction scheduledAction = new ScheduledAction(runnable, releaseTimeDate, this, scheduledActionKind);	1✔
151	scheduledActions.add(scheduledAction);	1✔
152	selector.wakeup();	1✔
153	return scheduledAction;	1✔
154	}
155
156	/**
157	* Cancels the scheduled action.
158	*
159	* @param scheduledAction the scheduled action to cancel.
160	* @return <code>true</code> if the scheduled action was still pending and got removed, <code>false</code> otherwise.
161	*/
162	boolean cancel(ScheduledAction scheduledAction) {
163	return scheduledActions.remove(scheduledAction);	1✔
164	}
165
166	private class Reactor extends Thread {	1✔
167
168	private volatile long shutdownRequestTimestamp = -1;	1✔
169
170	@Override
171	public void run() {
172	try {
173	reactorLoop();	×
174	} finally {
175	if (shutdownRequestTimestamp > 0) {	×
176	long shutDownDelay = System.currentTimeMillis() - shutdownRequestTimestamp;	×
177	LOGGER.info(this + " shut down after " + shutDownDelay + "ms");	×
178	} else {	×
179	boolean contained = reactorThreads.remove(this);	×
180	assert contained;	×
181	}
182	}
183	}	×
184
185	private void reactorLoop() {
186	// Loop until reactor shutdown was requested.
187	while (shutdownRequestTimestamp < 0) {	1✔
188	handleScheduledActionsOrPerformSelect();	1✔
189
190	handlePendingSelectionKeys();	1✔
191	}
192	}	×
193
194	@SuppressWarnings("LockNotBeforeTry")
195	private void handleScheduledActionsOrPerformSelect() {
196	ScheduledAction dueScheduledAction = null;	1✔
197
198	boolean permitToHandleScheduledActions = actionsSemaphore.tryAcquire();	1✔
199	if (permitToHandleScheduledActions) {	1✔
200	try {
201	dueScheduledAction = scheduledActions.poll();	1✔
202	} finally {
203	actionsSemaphore.release();	1✔
204	}
205	}
206
207	if (dueScheduledAction != null) {	1✔
208	dueScheduledAction.run();	1✔
209	return;	1✔
210	}
211
212	int newSelectedKeysCount = 0;	1✔
213	List<SelectionKey> selectedKeys;
214	synchronized (selector) {	1✔
215	ScheduledAction nextScheduledAction = scheduledActions.peek();	1✔
216
217	long selectWait;
218	if (nextScheduledAction == null) {	1✔
219	// There is no next scheduled action, wait indefinitely in select() or until another thread invokes
220	// selector.wakeup().
221	selectWait = 0;	1✔
222	} else {
223	selectWait = nextScheduledAction.getTimeToDueMillis();	1✔
224	if (selectWait <= 0) {	1✔
225	// A scheduled action was just released and became ready to execute.
226	return;	×
227	}
228	}
229
230	// Before we call select, we handle the pending the interest Ops. This will not block since no other
231	// thread is currently in select() at this time.
232	// Note: This was put deliberately before the registration lock. It may cause more synchronization but
233	// allows for more parallelism.
234	// Hopefully that assumption is right.
235	int myHandledPendingSetInterestOps = 0;	1✔
236	for (SetInterestOps setInterestOps; (setInterestOps = pendingSetInterestOps.poll()) != null;) {	1✔
237	setInterestOpsCancelledKeySafe(setInterestOps.selectionKey, setInterestOps.interestOps);	×
238
239	if (myHandledPendingSetInterestOps++ >= PENDING_SET_INTEREST_OPS_MAX_BATCH_SIZE) {	×
240	// This thread has handled enough "set pending interest ops" requests. Wakeup another one to
241	// handle the remaining (if any).
242	selector.wakeup();	×
243	break;	×
244	}
245	}
246
247	// Ensure that a wakeup() in registerWithSelector() gives the corresponding
248	// register() in the same method the chance to actually register the channel. In
249	// other words: This construct ensures that there is never another select()
250	// between a corresponding wakeup() and register() calls.
251	// See also https://stackoverflow.com/a/1112809/194894
252	registrationLock.lock();	1✔
253	registrationLock.unlock();	1✔
254
255	try {
256	newSelectedKeysCount = selector.select(selectWait);	1✔
257	} catch (IOException e) {	×
258	LOGGER.log(Level.SEVERE, "IOException while using select()", e);	×
259	return;	×
260	}	1✔
261
262	if (newSelectedKeysCount == 0) {	1✔
263	return;	1✔
264	}
265
266	// Copy the selected-key set over to selectedKeys, remove the keys from the
267	// selected key set and loose interest of the key OPs for the time being.
268	// Note that we perform this operation in two steps in order to maximize the
269	// timespan setRacing() is set.
270	Set<SelectionKey> selectedKeySet = selector.selectedKeys();	×
271	for (SelectionKey selectionKey : selectedKeySet) {	×
272	SelectionKeyAttachment selectionKeyAttachment = (SelectionKeyAttachment) selectionKey.attachment();	×
273	selectionKeyAttachment.setRacing();	×
274	}	×
275	for (SelectionKey selectionKey : selectedKeySet) {	×
276	setInterestOpsCancelledKeySafe(selectionKey, 0);	×
277	}	×
278
279	selectedKeys = new ArrayList<>(selectedKeySet.size());	×
280	selectedKeys.addAll(selectedKeySet);	×
281	selectedKeySet.clear();	×
282	}	×
283
284	int selectedKeysCount = selectedKeys.size();	×
285	int currentReactorThreadCount = reactorThreads.size();	×
286	int myKeyCount;
287	if (selectedKeysCount > currentReactorThreadCount) {	×
288	myKeyCount = selectedKeysCount / currentReactorThreadCount;	×
289	} else {
290	myKeyCount = selectedKeysCount;	×
291	}
292
293	final Level reactorSelectStatsLogLevel = Level.FINE;	×
294	if (LOGGER.isLoggable(reactorSelectStatsLogLevel)) {	×
295	LOGGER.log(reactorSelectStatsLogLevel,	×
296	"New selected key count: " + newSelectedKeysCount
297	+ ". Total selected key count " + selectedKeysCount
298	+ ". My key count: " + myKeyCount
299	+ ". Current reactor thread count: " + currentReactorThreadCount);
300	}
301
302	Collection<SelectionKey> mySelectedKeys = new ArrayList<>(myKeyCount);	×
303	Iterator<SelectionKey> it = selectedKeys.iterator();	×
304	for (int i = 0; i < myKeyCount; i++) {	×
305	SelectionKey selectionKey = it.next();	×
306	mySelectedKeys.add(selectionKey);	×
307	}
308	while (it.hasNext()) {	×
309	// Drain to pendingSelectionKeys.
310	SelectionKey selectionKey = it.next();	×
311	pendingSelectionKeys.add(selectionKey);	×
312	}	×
313
314	if (selectedKeysCount - myKeyCount > 0) {	×
315	// There where pending selection keys: Wakeup another reactor thread to handle them.
316	selector.wakeup();	×
317	}
318
319	handleSelectedKeys(mySelectedKeys);	×
320	}	×
321
322	private void handlePendingSelectionKeys() {
323	final int pendingSelectionKeysSize = pendingSelectionKeys.size();	1✔
324	if (pendingSelectionKeysSize == 0) {	1✔
325	return;	1✔
326	}
327
328	int currentReactorThreadCount = reactorThreads.size();	×
329	int myKeyCount = pendingSelectionKeysSize / currentReactorThreadCount;	×
330	Collection<SelectionKey> selectedKeys = new ArrayList<>(myKeyCount);	×
331	for (int i = 0; i < myKeyCount; i++) {	×
332	SelectionKey selectionKey = pendingSelectionKeys.poll();	×
333	if (selectionKey == null) {	×
334	// We lost a race and can abort here since the pendingSelectionKeys queue is empty.
335	break;	×
336	}
337	selectedKeys.add(selectionKey);	×
338	}
339
340	if (!pendingSelectionKeys.isEmpty()) {	×
341	// There are more pending selection keys, wakeup a thread blocked in select() to handle them.
342	selector.wakeup();	×
343	}
344
345	handleSelectedKeys(selectedKeys);	×
346	}	×
347
348	private void setInterestOpsCancelledKeySafe(SelectionKey selectionKey, int interestOps) {
349	try {
350	selectionKey.interestOps(interestOps);	×
351	}
352	catch (CancelledKeyException e) {	×
353	final Level keyCancelledLogLevel = Level.FINER;	×
354	if (LOGGER.isLoggable(keyCancelledLogLevel)) {	×
355	LOGGER.log(keyCancelledLogLevel, "Key '" + selectionKey + "' has been cancelled", e);	×
356	}
357	}	×
358	}	×
359
360	void requestShutdown() {
361	shutdownRequestTimestamp = System.currentTimeMillis();	×
362	}	×
363	}
364
365	private static void handleSelectedKeys(Collection<SelectionKey> selectedKeys) {
366	for (SelectionKey selectionKey : selectedKeys) {	×
367	SelectableChannel channel = selectionKey.channel();	×
368	SelectionKeyAttachment selectionKeyAttachment = (SelectionKeyAttachment) selectionKey.attachment();	×
369	ChannelSelectedCallback channelSelectedCallback = selectionKeyAttachment.channelSelectedCallback;	×
370	channelSelectedCallback.onChannelSelected(channel, selectionKey);	×
371	}	×
372	}	×
373
374	public interface ChannelSelectedCallback {
375	void onChannelSelected(SelectableChannel channel, SelectionKey selectionKey);
376	}
377
378	public void setReactorThreadCount(int reactorThreadCount) {
379	if (reactorThreadCount < 2) {	1✔
380	throw new IllegalArgumentException("Must have at least two reactor threads, but you requested " + reactorThreadCount);	×
381	}
382
383	synchronized (reactorThreads) {	1✔
384	int deltaThreads = reactorThreadCount - reactorThreads.size();	1✔
385	if (deltaThreads > 0) {	1✔
386	// Start new reactor thread. Note that we start the threads before we increase the permits of the
387	// actionsSemaphore.
388	for (int i = 0; i < deltaThreads; i++) {	1✔
389	Reactor reactor = new Reactor();	1✔
390	reactor.setDaemon(true);	1✔
391	reactor.setName("Smack " + reactorName + " Thread #" + i);	1✔
392	reactorThreads.add(reactor);	1✔
393	reactor.start();	1✔
394	}
395
396	actionsSemaphore.release(deltaThreads);	1✔
397	} else {
398	// Stop existing reactor threads. First we change the sign of deltaThreads, then we decrease the permits
399	// of the actionsSemaphore before we signal the selected reactor threads that they should shut down.
400	deltaThreads -= deltaThreads;	×
401
402	for (int i = deltaThreads - 1; i > 0; i--) {	×
403	// Note that this could potentially block forever, starving on the unfair semaphore.
404	actionsSemaphore.acquireUninterruptibly();	×
405	}
406
407	for (int i = deltaThreads - 1; i > 0; i--) {	×
408	Reactor reactor = reactorThreads.remove(i);	×
409	reactor.requestShutdown();	×
410	}
411
412	selector.wakeup();	×
413	}
414	}	1✔
415	}	1✔
416
417	public static final class SelectionKeyAttachment {
418	private final ChannelSelectedCallback channelSelectedCallback;
419	private final AtomicBoolean reactorThreadRacing = new AtomicBoolean();	×
420
421	private SelectionKeyAttachment(ChannelSelectedCallback channelSelectedCallback) {	×
422	this.channelSelectedCallback = channelSelectedCallback;	×
423	}	×
424
425	private void setRacing() {
426	// We use lazySet here since it is sufficient if the value does not become visible immediately.
427	reactorThreadRacing.lazySet(true);	×
428	}	×
429
430	public void resetReactorThreadRacing() {
431	reactorThreadRacing.set(false);	×
432	}	×
433
434	public boolean isReactorThreadRacing() {
435	return reactorThreadRacing.get();	×
436	}
437
438	}
439	}

igniterealtime / Smack / #2875

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous