openhft / chronicle-threads Goto Github PK
View Code? Open in Web Editor NEWHome Page: http://chronicle.software
License: Other
Home Page: http://chronicle.software
License: Other
As part of this, deprecate net.openhft.chronicle.core.threads.EventLoop
method
void addHandler(boolean dontAttemptToRunImmediatelyInCurrentThread, @NotNull EventHandler handler)
as it does not make sense to run an EventHandler in the current thread, and especially not if the event loop has not been started, or been closed.
We don't need two PauserMode
classes
In time-critical applications, the extra imposed check for new handles might be unacceptable so it should be possible to configure the accept modulo counter.
The EventLoop used to only remove a handler after an InvalidEventHandlerException
was thrown.
Now it will remove the handler on any exception, logging a warning.
InvalidEventHandlerException
will remove the handler silently as it did before.
The VanillaEventLoop::runLoop
method basically computes a reduction of all the handlers' status and stores the result in the local variable busy
. New handlers are only admitted when the reduced busy is false
. In other words, as long as there is at least one handler that is busy, no new handlers are accepted.
This means that if there are long term operations (such as bootstrapping a queue subscriber), new handlers are effectively prevented from attaching in a reasonable time.
There might be other implementations of EventLoop
that have the same issue. Concurrency issues must probably be considered before just admitting new handlers.
It would be useful to add a lifecycle method EventHandle::loopStarting
that allows thread-specific properties to be initialized prior to any invocation of EventHandler::action
. The method is invoked by the event loop thread. This would provide at least three benefits:
A) Some properties must not be concurrent (e.g. ThreadLoacal, volatile or AtomicX) as the loopStarting()
is guaranteed to be invoked by the event loop thread.
B) The time-critical EventHandler::action
, that is potentially called a gazillion times, could now be ridded from code that is only executed once.
C) Would provide a more logical and distinct separation of concerns.
Example from the wild (TcpHandler
):
private volatile Thread actionThread;
...
@Override
public boolean action() throws InvalidEventHandlerException {
...
// Unnecessary volatile read each invocation
if (actionThread == null)
actionThread = Thread.currentThread();
Test Case:
`
Thread t;
try (final EventLoop eventGroup = new EventGroup(false)) {
eventGroup.start();
t = new Thread(eventGroup::awaitTermination);
t.start();
eventGroup.addHandler(new EventHandler() {
@OverRide
public boolean action() throws InvalidEventHandlerException, InterruptedException {
System.out.println("time="+ System.currentTimeMillis());
return false;
}
public HandlerPriority priority() {
return HandlerPriority.MONITOR;
}
public void eventLoop(EventLoop eventLoop){
System.err.println("time="+ System.currentTimeMillis());
}
});
}
`
It can be found that the action method is called multiple times in a loop. Is this appropriate?
Don't understand what the purpose or role of the action method is? For example, what is its role in this project?
See also #23
The interface EventLoop
is not documented with JavaDocs.
Some threads that are executing LockSupport.parkNanos
take a while to respond to net.openhft.chronicle.threads.Threads#shutdown(java.util.concurrent.ExecutorService)
Have added a call to unpark them to speed up their shutdown.
In VanillaEventLoop
we read status variables, iterate over a number of Lists and do some logic while in the inner loop whilst doing numerous volatile reads. This opens up for potential improvements:
A) Reduce the number of. volatile reads (ideally less than 1 volatile read on average per loop)
B) Pre-compile the order in which handlers are invoked
It should be noted that adding a handler is likely a relatively infrequent operation compared to actually invoking handlers. A and B are orthogonal solutions which could or could not be implemented in any particular solution.
We could have ArrayLists
and arrays instead of CopyOnWriteList
and boolean close
instead of AtomicBoolean
provided that only the loop itself updates said variables.
So, I was thinking about implementing an IsolatedEventLoop
that "flattens" the logic so it just contains raw invocations of the handlers. This is done by unrolling logics and loops and producing an array (or perhaps even single aggregate Runnable) that depends on the currently added handlers (instances, priorities, etc).
Let's say we are adding handler h1, h2, h3, h4 to a IsolatedEventLoop
. Let's further assume that h3 has HandlerPriority.HIGH
and the others have HandlerPriority.MEDIUM
. Upon adding h4, the CompilingEventLoop
unrolls the operations:
if (highHandlers.isEmpty()) {
busy = runMediumLoopOnly();
} else {
busy = runHighAndMediumTasks();
}
private boolean runHighAndMediumTasks() {
boolean busy = false;
for (int i = 0; i < 4; i++) {
loopStartMS = Time.currentTimeMillis();
busy |= runAllHighHandlers();
busy |= runOneQuarterMediumHandler(i);
}
return busy;
}
@HotMethod
private boolean runAllHighHandlers() {
boolean busy = false;
for (int i = 0; i < highHandlers.size(); i++) {
final EventHandler handler = highHandlers.get(i);
try {
boolean action = handler.action();
busy |= action;
} catch (InvalidEventHandlerException e) {
removeHandler(handler, highHandlers);
} catch (Throwable e) {
Jvm.warn().on(getClass(), e);
}
}
return busy;
}
@HotMethod
private boolean runOneQuarterMediumHandler(int i) {
boolean busy = false;
final EventHandler[] mediumHandlersArray = this.mediumHandlersArray;
for (int j = i; j < mediumHandlersArray.length; j += 4) {
final EventHandler handler = mediumHandlersArray[j];
try {
busy |= handler.action();
} catch (InvalidEventHandlerException e) {
removeHandler(handler, mediumHandlers);
this.mediumHandlersArray = mediumHandlers.toArray(NO_EVENT_HANDLERS);
} catch (Throwable e) {
Jvm.warn().on(getClass(), e);
}
}
return busy;
}
to basically:
boolean busy = false;
busy |= h3.action();
busy |= h1.action();
busy |= h3.action();
busy |= h2.action();
busy |= h3.action();
busy |= h4.action();
Additional provisions have to be added for handling failing handlers that throw InvalidEventHandlerException
but that can relatively easily be handled.
The proposed IsolatedEventLoop
could potentially execute faster than a VanillaEventLoop
if new handlers are never or seldom added.
/P
E.g. missing final
and declaring volatile AtomicReference
The hash of the threads was often causing a number of consumer to run on the same thread, killing performance, This change also latest to https://github.com/ChronicleEnterprise/Chronicle-Datagrid/issues/143
Increasing the busy wait count by 4x reduced the number of times it yield/slept by a factor of 500 in one test.
Maybe introduced as part of a refactor, but Timer handlers are run together with Daemon handlers (which is never if you are always busy) due to this switch case. Then consequently, the call to runTimerHandlers
is empty in the hot method. Can submit a PR if you like.
2016-11-25T00:34:49,749 [main/event-loop-monitor] [WARN |threads.MonitorEventLoop] -
java.lang.ArithmeticException: / by zero
at net.openhft.chronicle.threads.EventGroup$LoopBlockMonitor.action(EventGroup.java:226) ~[chronicle-threads-1.7.3-SNAPSHOT.jar:?]
at net.openhft.chronicle.threads.MonitorEventLoop.runHandlers(MonitorEventLoop.java:120) ~[chronicle-threads-1.7.3-SNAPSHOT.jar:?]
at net.openhft.chronicle.threads.MonitorEventLoop.run(MonitorEventLoop.java:99) ~[chronicle-threads-1.7.3-SNAPSHOT.jar:?]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_101]
at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_101]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_101]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_101]
at java.lang.Thread.run(Thread.java:745) [?:1.8.0_101]
Looks like this line (226):
long blockingInterval = blockingTimeMS / (monitoryIntervalMs / 2);
Define 300 tasks to be added to EventGroup. The task type is CONCURRENT. After executing close, the task is discarded directly. Warning message: "Handler in newHandlerQueue was not accepted before close net.openhft.chronicle.threads"!
Cause Analysis:
When the task is added to VanillaEventLoop and EventGroup.close() is executed, the task in newHandlerQueue is discarded directly. This is not logical!
Tasks added before close should be executed!
Isn't it waiting for the task in newHandlerQueue to complete and shutting down VanillaEventLoop?
If it is the current logic, I think its use scenario is almost non-existent. If the scenario is limited, where does the high performance performance not be realized?
The "threads" library is in need of documentation and code cleanups:
Also, test coverage should be improved, especially as most other modules rely heavily on the threads module.
It would be nice to provide some kind of "top" command where information on various properties associated with EventLoops, pausers, priorities etc. could be easily viewed and understood.
This is something that could be very useful when running almost any library that is using Threads.
TID. TYPE %CPU Pauser
16422 VanillaEventLoop 100.0 BusyPauser
List of event handlers
17552 VanillaEventLoop 1.3% LongPauser
List of event handlers
The table above is just an outline and may not be possible or even desirable as exemplified.
Was generating lots of NoSuchFieldException and also ArrayList
Hey! I was just wondering if there was a performance comparison between this and other threadpool implementations, e.g. those found in java.util.concurrent.*
. I'm just trying to gauge what the advantages and disadvantages of this implementation are.
Thanks!
It is possible to create a subclass of InvalidEventHandlerException
that only exists as a singleton (e.g. SingletonInvalidEventHandlerException
) and that is statically created. EventHandlers
that only use the InvalidEventHandlerException
as a means of removing itself from the EventLoop
could use the static Exception for pure removal with no underlying exception to signal. This will improve loop performance and reduce jittering.
Exceptional performance...
Upon leaving a VanillaEventLoop
, the method EventHandler::loopFinished
is invoked as specified in the API.
However, there is nothing in the API contract even remotely suggesting that Closeable::close
should be invoked if an EventHandle
incidentally happens to implement Closeable
.
This might lead to highly undesirable side-effects including attempting to release native memory a plurality of times.
This could be remedied in at least two ways:
A) Add extends Closeable
to the interface EventHandler
and add appropriate JavaDoc defining if and when the method is invoked.
B) Remove any magic invocation of Closeable::close()
from any and all implementations of the EventLoop
interface.
Both A and B will have compatibility issues.
A client reported a StreamCorrupted error and believes this is related to this issue https://stackoverflow.com/questions/51398967/how-would-comparison-cause-numerical-overflow-when-using-system-nanotime.
This happens when pauser.pause(timeout, timeunit) would throw TimeoutException BEFORE the desired timeout (15 seconds by default), and TableStoreWriterLock would forceUnlock very quickly, causing concurrent writers to corrupt the queue file.
Following code in BusyTimedPauser:
@Override
public void pause(long timeout, TimeUnit timeUnit) throws TimeoutException {
if (time == Long.MAX_VALUE)
time = System.nanoTime();
if (time + timeUnit.toNanos(timeout) < System.nanoTime())
throw new TimeoutException();
}
Should be changed to something like:
@Override
public void pause(long timeout, TimeUnit timeUnit) throws TimeoutException {
if (time == Long.MAX_VALUE)
time = System.nanoTime();
if (time + timeUnit.toNanos(timeout) - System.nanoTime() < 0)
throw new TimeoutException();
}
This explains why it happens randomly and is hard to reproduce.
A) The API does not specify which guarantees (if any) are made to the order in which EventHandler::action
is invoked across added handlers.
EventHandler
objects added to a VanillaEventHandler
are handled inconsistently with respect to the order in which they were added.
add | action |
---|---|
1 | 1 |
1,2 | 2,1 |
1,2,3 | 3,2,1 |
1,2,3,4 | 4,3,2,1 |
1,2,3,4,5 | 1,2,3,4,5 |
1,.., N, N>4 | 1, ..., N |
Customer use case: event handler was set up as BLOCKING as it performs I/O. When it does not do anything, it is fast, so we need a pause to stop busy spinning thread
source code:
private void addNewHandler(@NotNull EventHandler handler) {
HandlerPriority t1 = handler.priority();
switch (t1 == null ? HandlerPriority.MEDIUM : t1) {
case HIGH:
if (!highHandlers.contains(handler))
highHandlers.add(handler);
break;
case REPLICATION:
case CONCURRENT:
case MEDIUM:
if (!mediumHandlers.contains(handler)) {
mediumHandlers.add(handler);
mediumHandlersArray = mediumHandlers.toArray(NO_EVENT_HANDLERS);
}
break;
case TIMER:
if (!timerHandlers.contains(handler))
timerHandlers.add(handler);
break;
case DAEMON:
if (!daemonHandlers.contains(handler))
daemonHandlers.add(handler);
break;
default:
throw new IllegalArgumentException("Cannot add a " + handler.priority() + " task to a busy waiting thread");
}
handler.eventLoop(parent);
}
After adding a new handler, the following code:
handler.eventLoop(parent)
Don't know what purpose this code is based on? Can my dear friend inform this?
eventLoop is the default empty method in EventHandler, chronicle-thread does not override the eventLoop method: how does the use of handler.eventLoop(parent) be implemented?
Can my dear friend help answer it?
I really like your code on github, @peter-lawrey , and I'm always looking at you, hoping to get your point!
[main] DEBUG net.openhft.chronicle.threads.Threads -
java.util.ConcurrentModificationException
at java.util.HashMap$HashIterator.nextNode(HashMap.java:1445)
at java.util.HashMap$KeyIterator.next(HashMap.java:1469)
at java.util.AbstractCollection.toArray(AbstractCollection.java:141)
at java.util.ArrayList.<init>(ArrayList.java:178)
at net.openhft.chronicle.threads.Threads.forEachThread(Threads.java:187)
at net.openhft.chronicle.threads.Threads.interrupt(Threads.java:164)
at net.openhft.chronicle.threads.Threads.shutdown(Threads.java:132)
at net.openhft.chronicle.threads.BlockingEventLoop.close(BlockingEventLoop.java:146)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:42)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:38)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:30)
at net.openhft.chronicle.threads.EventGroup.close(EventGroup.java:353)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:42)
at software.chronicle.enterprise.map.cluster.MapClusterContext.close(MapClusterContext.java:102)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:42)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:38)
at net.openhft.chronicle.core.io.Closeable.closeQuietly(Closeable.java:30)
at software.chronicle.enterprise.map.ReplicatedMap.close(ReplicatedMap.java:234)
at software.chronicle.fix.router.integration.IntegrationRouterTest.test(IntegrationRouterTest.java:178)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:498)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)
at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)
at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)
at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)
at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)
at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)
at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)
at org.junit.runners.ParentRunner.run(ParentRunner.java:363)
at org.junit.runner.JUnitCore.run(JUnitCore.java:137)
at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)
at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)
at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)
at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)
For backward compatibility, treat HIGH as MEDIUM.
A very minor issue: says "Event Group" but should say "MediumEventLoop".
2020-04-21T15:04:25.355 main/replication-acceptor-2 AcceptorEventHandler localhost80901, port=55169java.lang.IllegalStateException: Event Group has been interrupted
at net.openhft.chronicle.threads.MediumEventLoop.checkInterrupted(MediumEventLoop.java:176)
at net.openhft.chronicle.threads.MediumEventLoop.addHandler(MediumEventLoop.java:156)
at net.openhft.chronicle.network.AcceptorEventHandler.action(AcceptorEventHandler.java:97)
at net.openhft.chronicle.threads.BlockingEventLoop$Runner.run(BlockingEventLoop.java:172)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.