WIP - got queue to no longer deadlock

This commit is contained in:
nathan 2015-04-19 14:25:12 +02:00
parent 2e6ba71eed
commit 42230706c7
20 changed files with 594 additions and 1509 deletions

View File

@ -1,10 +0,0 @@
usePlugin('java')
group="org.mbassy"
version="1.1.2"
dependencies {
addMavenRepo()
testCompile "junit:junit:4.10@jar"
}

View File

@ -1,742 +0,0 @@
#
# A fatal error has been detected by the Java Runtime Environment:
#
# SIGSEGV (0xb) at pc=0x00007f2e07f877a4, pid=26955, tid=139835696731904
#
# JRE version: Java(TM) SE Runtime Environment (7.0_55-b13) (build 1.7.0_55-b13)
# Java VM: Java HotSpot(TM) 64-Bit Server VM (24.55-b03 mixed mode linux-amd64 compressed oops)
# Problematic frame:
# V [libjvm.so+0x9697a4] Unsafe_GetObject+0xc4
#
# Failed to write core dump. Core dumps have been disabled. To enable core dumping, try "ulimit -c unlimited" before starting Java again
#
# If you would like to submit a bug report, please visit:
# http://bugreport.sun.com/bugreport/crash.jsp
#
--------------- T H R E A D ---------------
Current thread (0x00007f2e0000a000): JavaThread "main" [_thread_in_vm, id=26960, stack(0x00007f2e08f7d000,0x00007f2e0907e000)]
siginfo:si_signo=SIGSEGV: si_errno=0, si_code=1 (SEGV_MAPERR), si_addr=0x0000000000000010
Registers:
RAX=0x0000000000000000, RBX=0x00007f2e0000a000, RCX=0x0000000000000010, RDX=0x00007f2e08455a50
RSP=0x00007f2e0907c650, RBP=0x00007f2e0907c6a0, RSI=0x00007f2e0907c750, RDI=0x00007f2e0000a1e8
R8 =0x00000000ffffffff, R9 =0x0000000000005300, R10=0x00007f2dfd012cac, R11=0x00007f2e07ec5e40
R12=0x0000000000000000, R13=0x0000000000000010, R14=0x00007f2e0000a1e8, R15=0x00007f2e0844be58
RIP=0x00007f2e07f877a4, EFLAGS=0x0000000000010202, CSGSFS=0x0000000000000033, ERR=0x0000000000000004
TRAPNO=0x000000000000000e
Top of Stack: (sp=0x00007f2e0907c650)
0x00007f2e0907c650: 00007f2e0000a000 00007f2dfd01da03
0x00007f2e0907c660: 00007f2e0000a000 00007f2d00000000
0x00007f2e0907c670: 00007f2e0907c798 0000000708ce72b8
0x00007f2e0907c680: 0000000000000000 0000000708ce72b8
0x00007f2e0907c690: 00007f2e0907c750 00007f2e0000a000
0x00007f2e0907c6a0: 00007f2e0907c718 00007f2dfd012cd8
0x00007f2e0907c6b0: 0000000709241b7f 00007f2e000000b6
0x00007f2e0907c6c0: 00007f2e0000add8 00007f2e00198b90
0x00007f2e0907c6d0: 00007f2e0000a000 00007f2e0907c6d8
0x00007f2e0907c6e0: 0000000000000000 00007f2e0907c750
0x00007f2e0907c6f0: 0000000708cecc78 0000000000000000
0x00007f2e0907c700: 0000000708ce72b8 0000000000000000
0x00007f2e0907c710: 00007f2e0907c738 00007f2e0907c798
0x00007f2e0907c720: 00007f2dfd006233 0000000000000000
0x00007f2e0907c730: 00007f2dfd00ecdb 0000000000000010
0x00007f2e0907c740: 00007f2dfd01d01f 0000000000000000
0x00007f2e0907c750: 000000070de009a0 00007f2e0907c758
0x00007f2e0907c760: 0000000709241b7f 00007f2e0907c7a8
0x00007f2e0907c770: 0000000709248f30 0000000000000000
0x00007f2e0907c780: 0000000709241b98 00007f2e0907c738
0x00007f2e0907c790: 00007f2e0907c7a8 00007f2e0907c7f0
0x00007f2e0907c7a0: 00007f2dfd006233 0000000000000000
0x00007f2e0907c7b0: 00007f2e0907c7b0 000000070924200d
0x00007f2e0907c7c0: 00007f2e0907c888 0000000709248f30
0x00007f2e0907c7d0: 0000000000000000 00000007092420e8
0x00007f2e0907c7e0: 00007f2e0907c7a8 00007f2e0907c888
0x00007f2e0907c7f0: 00007f2e0907c8d0 00007f2dfd006233
0x00007f2e0907c800: 0000000000000000 0000000000000000
0x00007f2e0907c810: 0000000000000090 0000000000000000
0x00007f2e0907c820: 0000000000000000 0000000000000000
0x00007f2e0907c830: 0000000000000001 0000000000000000
0x00007f2e0907c840: 0000000000000001 0000000000000000
Instructions: (pc=0x00007f2e07f877a4)
0x00007f2e07f87784: 83 60 02 00 00 06 00 00 00 48 89 5d b0 74 04 49
0x00007f2e07f87794: 8b 04 24 4c 8b 3d ca ef 48 00 41 80 3f 00 74 7c
0x00007f2e07f877a4: 42 8b 14 28 31 f6 85 d2 0f 85 ae 00 00 00 4c 89
0x00007f2e07f877b4: f7 e8 c6 3d d0 ff 48 89 c3 48 8b 05 ac c9 48 00
Register to memory mapping:
RAX=0x0000000000000000 is an unknown value
RBX=0x00007f2e0000a000 is a thread
RCX=0x0000000000000010 is an unknown value
RDX=0x00007f2e08455a50: <offset 0xe37a50> in /home/user/dorkbox/eclipse/jre/jre/lib/amd64/server/libjvm.so at 0x00007f2e0761e000
RSP=0x00007f2e0907c650 is pointing into the stack for thread: 0x00007f2e0000a000
RBP=0x00007f2e0907c6a0 is pointing into the stack for thread: 0x00007f2e0000a000
RSI=0x00007f2e0907c750 is pointing into the stack for thread: 0x00007f2e0000a000
RDI=0x00007f2e0000a1e8 is an unknown value
R8 =0x00000000ffffffff is an unknown value
R9 =0x0000000000005300 is an unknown value
R10=0x00007f2dfd012cac is at code_begin+620 in an Interpreter codelet
method entry point (kind = native) [0x00007f2dfd012a40, 0x00007f2dfd0132a0] 2144 bytes
R11=0x00007f2e07ec5e40: <offset 0x8a7e40> in /home/user/dorkbox/eclipse/jre/jre/lib/amd64/server/libjvm.so at 0x00007f2e0761e000
R12=0x0000000000000000 is an unknown value
R13=0x0000000000000010 is an unknown value
R14=0x00007f2e0000a1e8 is an unknown value
R15=0x00007f2e0844be58: <offset 0xe2de58> in /home/user/dorkbox/eclipse/jre/jre/lib/amd64/server/libjvm.so at 0x00007f2e0761e000
Stack: [0x00007f2e08f7d000,0x00007f2e0907e000], sp=0x00007f2e0907c650, free space=1021k
Native frames: (J=compiled Java code, j=interpreted, Vv=VM code, C=native code)
V [libjvm.so+0x9697a4] Unsafe_GetObject+0xc4
j sun.misc.Unsafe.getObject(Ljava/lang/Object;J)Ljava/lang/Object;+0
j dorkbox.util.messagebus.common.simpleq.MpmcExchangerQueue.getMessage1(Ljava/lang/Object;)Ljava/lang/Object;+7
j dorkbox.util.messagebus.common.simpleq.MpmcExchangerQueue.take()Ljava/lang/Object;+85
j dorkbox.util.messagebus.MpmcQueueAltPerfTest.performanceRun(ILdorkbox/util/messagebus/common/simpleq/MpmcExchangerQueue;)J+32
j dorkbox.util.messagebus.MpmcQueueAltPerfTest.main([Ljava/lang/String;)V+86
v ~StubRoutines::call_stub
V [libjvm.so+0x5f84b5] JavaCalls::call_helper(JavaValue*, methodHandle*, JavaCallArguments*, Thread*)+0x365
V [libjvm.so+0x5f6f18] JavaCalls::call(JavaValue*, methodHandle, JavaCallArguments*, Thread*)+0x28
V [libjvm.so+0x630109] jni_invoke_static(JNIEnv_*, JavaValue*, _jobject*, JNICallType, _jmethodID*, JNI_ArgumentPusher*, Thread*)+0x219
V [libjvm.so+0x639192] jni_CallStaticVoidMethod+0x162
C [libjli.so+0x36d9] JavaMain+0x7e9
Java frames: (J=compiled Java code, j=interpreted, Vv=VM code)
j sun.misc.Unsafe.getObject(Ljava/lang/Object;J)Ljava/lang/Object;+0
j dorkbox.util.messagebus.common.simpleq.MpmcExchangerQueue.getMessage1(Ljava/lang/Object;)Ljava/lang/Object;+7
j dorkbox.util.messagebus.common.simpleq.MpmcExchangerQueue.take()Ljava/lang/Object;+85
j dorkbox.util.messagebus.MpmcQueueAltPerfTest.performanceRun(ILdorkbox/util/messagebus/common/simpleq/MpmcExchangerQueue;)J+32
j dorkbox.util.messagebus.MpmcQueueAltPerfTest.main([Ljava/lang/String;)V+86
v ~StubRoutines::call_stub
--------------- P R O C E S S ---------------
Java Threads: ( => current thread )
0x00007f2e0019c800 JavaThread "Thread-1" [_thread_in_Java, id=27002, stack(0x00007f2de1f06000,0x00007f2de2007000)]
0x00007f2dc4001000 JavaThread "Attach Listener" daemon [_thread_blocked, id=27000, stack(0x00007f2de1d03000,0x00007f2de1e04000)]
0x00007f2e00120800 JavaThread "process reaper" daemon [_thread_blocked, id=26979, stack(0x00007f2de24cb000,0x00007f2de2504000)]
0x00007f2e0009f800 JavaThread "Service Thread" daemon [_thread_blocked, id=26975, stack(0x00007f2de280d000,0x00007f2de290e000)]
0x00007f2e0009d000 JavaThread "C2 CompilerThread1" daemon [_thread_blocked, id=26974, stack(0x00007f2de290e000,0x00007f2de2a0f000)]
0x00007f2e0009a000 JavaThread "C2 CompilerThread0" daemon [_thread_blocked, id=26973, stack(0x00007f2de2a0f000,0x00007f2de2b10000)]
0x00007f2e00090000 JavaThread "Signal Dispatcher" daemon [_thread_blocked, id=26972, stack(0x00007f2de2b10000,0x00007f2de2c11000)]
0x00007f2e00079000 JavaThread "Finalizer" daemon [_thread_blocked, id=26971, stack(0x00007f2de32f3000,0x00007f2de33f4000)]
0x00007f2e00075000 JavaThread "Reference Handler" daemon [_thread_blocked, id=26970, stack(0x00007f2de33f4000,0x00007f2de34f5000)]
=>0x00007f2e0000a000 JavaThread "main" [_thread_in_vm, id=26960, stack(0x00007f2e08f7d000,0x00007f2e0907e000)]
Other Threads:
0x00007f2e00072800 VMThread [stack: 0x00007f2de34f5000,0x00007f2de35f6000] [id=26969]
0x00007f2e000b2000 WatcherThread [stack: 0x00007f2de270c000,0x00007f2de280d000] [id=26976]
VM state:not at safepoint (normal execution)
VM Mutex/Monitor currently owned by a thread: None
Heap
PSYoungGen total 72704K, used 2498K [0x00000007af480000, 0x00000007b4580000, 0x0000000800000000)
eden space 62464K, 4% used [0x00000007af480000,0x00000007af6f0ab0,0x00000007b3180000)
from space 10240K, 0% used [0x00000007b3180000,0x00000007b3180000,0x00000007b3b80000)
to space 10240K, 0% used [0x00000007b3b80000,0x00000007b3b80000,0x00000007b4580000)
ParOldGen total 164864K, used 7104K [0x000000070de00000, 0x0000000717f00000, 0x00000007af480000)
object space 164864K, 4% used [0x000000070de00000,0x000000070e4f01e8,0x0000000717f00000)
PSPermGen total 21504K, used 6446K [0x0000000708c00000, 0x000000070a100000, 0x000000070de00000)
object space 21504K, 29% used [0x0000000708c00000,0x000000070924b910,0x000000070a100000)
Card table byte_map: [0x00007f2e05ac5000,0x00007f2e06280000] byte_map_base: 0x00007f2e0227f000
Polling page: 0x00007f2e090a1000
Code Cache [0x00007f2dfd000000, 0x00007f2dfd270000, 0x00007f2e00000000)
total_blobs=292 nmethods=29 adapters=217 free_code_cache=48669Kb largest_free_block=49810368
Compilation events (10 events):
Event: 0.477 Thread 0x00007f2e0009a000 24 java.io.UnixFileSystem::normalize (75 bytes)
Event: 0.480 Thread 0x00007f2e0009a000 nmethod 24 0x00007f2dfd076a50 code [0x00007f2dfd076bc0, 0x00007f2dfd076e18]
Event: 0.485 Thread 0x00007f2e0009d000 25 dorkbox.util.messagebus.common.simpleq.jctools.ConcurrentSequencedCircularArrayQueue::calcSequenceOffset (9 bytes)
Event: 0.485 Thread 0x00007f2e0009a000 26 dorkbox.util.messagebus.common.simpleq.jctools.ConcurrentSequencedCircularArrayQueue::calcSequenceOffset (12 bytes)
Event: 0.485 Thread 0x00007f2e0009a000 nmethod 26 0x00007f2dfd07ae50 code [0x00007f2dfd07af80, 0x00007f2dfd07afd8]
Event: 0.485 Thread 0x00007f2e0009a000 27 dorkbox.util.messagebus.common.simpleq.jctools.ConcurrentSequencedCircularArrayQueue::soSequence (11 bytes)
Event: 0.485 Thread 0x00007f2e0009d000 nmethod 25 0x00007f2dfd076490 code [0x00007f2dfd0765c0, 0x00007f2dfd076638]
Event: 0.485 Thread 0x00007f2e0009a000 nmethod 27 0x00007f2dfd076210 code [0x00007f2dfd076340, 0x00007f2dfd076398]
Event: 0.486 Thread 0x00007f2e0009d000 29 % dorkbox.util.messagebus.common.simpleq.jctools.ConcurrentSequencedCircularArrayQueue::<init> @ 35 (61 bytes)
Event: 0.487 Thread 0x00007f2e0009d000 nmethod 29% 0x00007f2dfd06f450 code [0x00007f2dfd06f5a0, 0x00007f2dfd06f698]
GC Heap History (4 events):
Event: 0.488 GC heap before
{Heap before GC invocations=1 (full 0):
PSYoungGen total 72704K, used 14896K [0x00000007af480000, 0x00000007b4580000, 0x0000000800000000)
eden space 62464K, 23% used [0x00000007af480000,0x00000007b030c030,0x00000007b3180000)
from space 10240K, 0% used [0x00000007b3b80000,0x00000007b3b80000,0x00000007b4580000)
to space 10240K, 0% used [0x00000007b3180000,0x00000007b3180000,0x00000007b3b80000)
ParOldGen total 164864K, used 0K [0x000000070de00000, 0x0000000717f00000, 0x00000007af480000)
object space 164864K, 0% used [0x000000070de00000,0x000000070de00000,0x0000000717f00000)
PSPermGen total 21504K, used 6443K [0x0000000708c00000, 0x000000070a100000, 0x000000070de00000)
object space 21504K, 29% used [0x0000000708c00000,0x000000070924ae18,0x000000070a100000)
Event: 0.492 GC heap after
Heap after GC invocations=1 (full 0):
PSYoungGen total 72704K, used 7264K [0x00000007af480000, 0x00000007b4580000, 0x0000000800000000)
eden space 62464K, 0% used [0x00000007af480000,0x00000007af480000,0x00000007b3180000)
from space 10240K, 70% used [0x00000007b3180000,0x00000007b3898340,0x00000007b3b80000)
to space 10240K, 0% used [0x00000007b3b80000,0x00000007b3b80000,0x00000007b4580000)
ParOldGen total 164864K, used 8K [0x000000070de00000, 0x0000000717f00000, 0x00000007af480000)
object space 164864K, 0% used [0x000000070de00000,0x000000070de02000,0x0000000717f00000)
PSPermGen total 21504K, used 6443K [0x0000000708c00000, 0x000000070a100000, 0x000000070de00000)
object space 21504K, 29% used [0x0000000708c00000,0x000000070924ae18,0x000000070a100000)
}
Event: 0.492 GC heap before
{Heap before GC invocations=2 (full 1):
PSYoungGen total 72704K, used 7264K [0x00000007af480000, 0x00000007b4580000, 0x0000000800000000)
eden space 62464K, 0% used [0x00000007af480000,0x00000007af480000,0x00000007b3180000)
from space 10240K, 70% used [0x00000007b3180000,0x00000007b3898340,0x00000007b3b80000)
to space 10240K, 0% used [0x00000007b3b80000,0x00000007b3b80000,0x00000007b4580000)
ParOldGen total 164864K, used 8K [0x000000070de00000, 0x0000000717f00000, 0x00000007af480000)
object space 164864K, 0% used [0x000000070de00000,0x000000070de02000,0x0000000717f00000)
PSPermGen total 21504K, used 6443K [0x0000000708c00000, 0x000000070a100000, 0x000000070de00000)
object space 21504K, 29% used [0x0000000708c00000,0x000000070924ae18,0x000000070a100000)
Event: 0.516 GC heap after
Heap after GC invocations=2 (full 1):
PSYoungGen total 72704K, used 0K [0x00000007af480000, 0x00000007b4580000, 0x0000000800000000)
eden space 62464K, 0% used [0x00000007af480000,0x00000007af480000,0x00000007b3180000)
from space 10240K, 0% used [0x00000007b3180000,0x00000007b3180000,0x00000007b3b80000)
to space 10240K, 0% used [0x00000007b3b80000,0x00000007b3b80000,0x00000007b4580000)
ParOldGen total 164864K, used 7104K [0x000000070de00000, 0x0000000717f00000, 0x00000007af480000)
object space 164864K, 4% used [0x000000070de00000,0x000000070e4f01e8,0x0000000717f00000)
PSPermGen total 21504K, used 6441K [0x0000000708c00000, 0x000000070a100000, 0x000000070de00000)
object space 21504K, 29% used [0x0000000708c00000,0x000000070924a658,0x000000070a100000)
}
Deoptimization events (0 events):
No events
Internal exceptions (10 events):
Event: 0.476 Thread 0x00007f2e0000a000 Threw 0x00000007af9cf800 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.476 Thread 0x00007f2e0000a000 Threw 0x00000007af9d2250 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.476 Thread 0x00007f2e0000a000 Threw 0x00000007af9d5920 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.477 Thread 0x00007f2e0000a000 Threw 0x00000007af9d86d8 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.477 Thread 0x00007f2e0000a000 Threw 0x00000007af9dc4e8 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.477 Thread 0x00007f2e0000a000 Threw 0x00000007af9e1b48 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.477 Thread 0x00007f2e0000a000 Threw 0x00000007af9e4c70 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.478 Thread 0x00007f2e0000a000 Threw 0x00000007af9e7e68 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.478 Thread 0x00007f2e0000a000 Threw 0x00000007af9f0298 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Event: 0.516 Thread 0x00007f2e0000a000 Threw 0x00000007af480538 at /HUDSON/workspace/7u-2-build-linux-amd64/jdk7u55/869/hotspot/src/share/vm/prims/jvm.cpp:1244
Events (10 events):
Event: 0.477 loading class 0x00007f2e0019e130 done
Event: 0.478 loading class 0x00007f2e00163280
Event: 0.478 loading class 0x00007f2e00163280 done
Event: 0.478 loading class 0x00007f2e0019d6d0
Event: 0.478 loading class 0x00007f2e0019d6d0 done
Event: 0.487 Executing VM operation: ParallelGCSystemGC
Event: 0.516 Executing VM operation: ParallelGCSystemGC done
Event: 0.516 loading class 0x00007f2e000b8fd0
Event: 0.516 loading class 0x00007f2e000b8fd0 done
Event: 0.516 Thread 0x00007f2e0019c800 Thread added: 0x00007f2e0019c800
Dynamic libraries:
00400000-00401000 r-xp 00000000 fc:02 16917903 /home/user/dorkbox/eclipse/jre/bin/java
00600000-00601000 rw-p 00000000 fc:02 16917903 /home/user/dorkbox/eclipse/jre/bin/java
0120b000-0122c000 rw-p 00000000 00:00 0 [heap]
708c00000-70a100000 rw-p 00000000 00:00 0
70a100000-70de00000 rw-p 00000000 00:00 0
70de00000-717f00000 rw-p 00000000 00:00 0
717f00000-7af480000 rw-p 00000000 00:00 0
7af480000-7b4580000 rw-p 00000000 00:00 0
7b4580000-800000000 rw-p 00000000 00:00 0
7f2da0000000-7f2da0021000 rw-p 00000000 00:00 0
7f2da0021000-7f2da4000000 ---p 00000000 00:00 0
7f2da8000000-7f2da8021000 rw-p 00000000 00:00 0
7f2da8021000-7f2dac000000 ---p 00000000 00:00 0
7f2dac000000-7f2dac033000 rw-p 00000000 00:00 0
7f2dac033000-7f2db0000000 ---p 00000000 00:00 0
7f2db0000000-7f2db0021000 rw-p 00000000 00:00 0
7f2db0021000-7f2db4000000 ---p 00000000 00:00 0
7f2db4000000-7f2db4021000 rw-p 00000000 00:00 0
7f2db4021000-7f2db8000000 ---p 00000000 00:00 0
7f2db8000000-7f2db8219000 rw-p 00000000 00:00 0
7f2db8219000-7f2dbc000000 ---p 00000000 00:00 0
7f2dbc000000-7f2dbc1db000 rw-p 00000000 00:00 0
7f2dbc1db000-7f2dc0000000 ---p 00000000 00:00 0
7f2dc0000000-7f2dc0021000 rw-p 00000000 00:00 0
7f2dc0021000-7f2dc4000000 ---p 00000000 00:00 0
7f2dc4000000-7f2dc4021000 rw-p 00000000 00:00 0
7f2dc4021000-7f2dc8000000 ---p 00000000 00:00 0
7f2dc8000000-7f2dc8021000 rw-p 00000000 00:00 0
7f2dc8021000-7f2dcc000000 ---p 00000000 00:00 0
7f2dcc460000-7f2dd4000000 rw-p 00000000 00:00 0
7f2dd4000000-7f2dd4021000 rw-p 00000000 00:00 0
7f2dd4021000-7f2dd8000000 ---p 00000000 00:00 0
7f2dd8000000-7f2dd8021000 rw-p 00000000 00:00 0
7f2dd8021000-7f2ddc000000 ---p 00000000 00:00 0
7f2ddc000000-7f2ddc021000 rw-p 00000000 00:00 0
7f2ddc021000-7f2de0000000 ---p 00000000 00:00 0
7f2de198e000-7f2de1af9000 rw-p 00000000 00:00 0
7f2de1af9000-7f2de1b03000 r-xp 00000000 fc:02 15997771 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libinstrument.so
7f2de1b03000-7f2de1d02000 ---p 0000a000 fc:02 15997771 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libinstrument.so
7f2de1d02000-7f2de1d03000 rw-p 00009000 fc:02 15997771 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libinstrument.so
7f2de1d03000-7f2de1d06000 ---p 00000000 00:00 0
7f2de1d06000-7f2de1e04000 rw-p 00000000 00:00 0 [stack:27000]
7f2de1e2c000-7f2de1f06000 rw-p 00000000 00:00 0
7f2de1f06000-7f2de1f09000 ---p 00000000 00:00 0
7f2de1f09000-7f2de2007000 rw-p 00000000 00:00 0 [stack:27002]
7f2de2007000-7f2de2072000 r--s 00e1f000 fc:02 15603316 /home/user/dorkbox/eclipse/jre/lib/tools.jar
7f2de2072000-7f2de2082000 r-xp 00000000 fc:02 15997753 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libnio.so
7f2de2082000-7f2de2282000 ---p 00010000 fc:02 15997753 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libnio.so
7f2de2282000-7f2de2283000 rw-p 00010000 fc:02 15997753 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libnio.so
7f2de2283000-7f2de22b4000 rw-p 00000000 00:00 0
7f2de22b4000-7f2de22ca000 r-xp 00000000 fc:02 15997793 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libnet.so
7f2de22ca000-7f2de24ca000 ---p 00016000 fc:02 15997793 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libnet.so
7f2de24ca000-7f2de24cb000 rw-p 00016000 fc:02 15997793 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libnet.so
7f2de24cb000-7f2de24ce000 ---p 00000000 00:00 0
7f2de24ce000-7f2de2504000 rw-p 00000000 00:00 0 [stack:26979]
7f2de2504000-7f2de250c000 r-xp 00000000 fc:02 15997764 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libmanagement.so
7f2de250c000-7f2de270b000 ---p 00008000 fc:02 15997764 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libmanagement.so
7f2de270b000-7f2de270c000 rw-p 00007000 fc:02 15997764 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libmanagement.so
7f2de270c000-7f2de270d000 ---p 00000000 00:00 0
7f2de270d000-7f2de280d000 rw-p 00000000 00:00 0 [stack:26976]
7f2de280d000-7f2de2810000 ---p 00000000 00:00 0
7f2de2810000-7f2de290e000 rw-p 00000000 00:00 0 [stack:26975]
7f2de290e000-7f2de2911000 ---p 00000000 00:00 0
7f2de2911000-7f2de2a0f000 rw-p 00000000 00:00 0 [stack:26974]
7f2de2a0f000-7f2de2a12000 ---p 00000000 00:00 0
7f2de2a12000-7f2de2b10000 rw-p 00000000 00:00 0 [stack:26973]
7f2de2b10000-7f2de2b13000 ---p 00000000 00:00 0
7f2de2b13000-7f2de2c11000 rw-p 00000000 00:00 0 [stack:26972]
7f2de2c11000-7f2de32f3000 r--p 00000000 fc:02 17046099 /usr/lib/locale/locale-archive
7f2de32f3000-7f2de32f6000 ---p 00000000 00:00 0
7f2de32f6000-7f2de33f4000 rw-p 00000000 00:00 0 [stack:26971]
7f2de33f4000-7f2de33f7000 ---p 00000000 00:00 0
7f2de33f7000-7f2de34f5000 rw-p 00000000 00:00 0 [stack:26970]
7f2de34f5000-7f2de34f6000 ---p 00000000 00:00 0
7f2de34f6000-7f2de4000000 rw-p 00000000 00:00 0 [stack:26969]
7f2de4000000-7f2de4021000 rw-p 00000000 00:00 0
7f2de4021000-7f2de8000000 ---p 00000000 00:00 0
7f2de8000000-7f2de8021000 rw-p 00000000 00:00 0
7f2de8021000-7f2dec000000 ---p 00000000 00:00 0
7f2dec000000-7f2dec021000 rw-p 00000000 00:00 0
7f2dec021000-7f2df0000000 ---p 00000000 00:00 0
7f2df0000000-7f2df0021000 rw-p 00000000 00:00 0
7f2df0021000-7f2df4000000 ---p 00000000 00:00 0
7f2df4000000-7f2df4021000 rw-p 00000000 00:00 0
7f2df4021000-7f2df8000000 ---p 00000000 00:00 0
7f2df8000000-7f2df8021000 rw-p 00000000 00:00 0
7f2df8021000-7f2dfc000000 ---p 00000000 00:00 0
7f2dfc005000-7f2dfc02b000 r--s 0020e000 fc:02 14943167 /home/user/dorkbox/resources/Dependencies/netty/netty-all-4.1.0.jar
7f2dfc02b000-7f2dfd000000 rw-p 00000000 00:00 0
7f2dfd000000-7f2dfd270000 rwxp 00000000 00:00 0
7f2dfd270000-7f2e001e1000 rw-p 00000000 00:00 0
7f2e001e1000-7f2e04000000 ---p 00000000 00:00 0
7f2e04019000-7f2e042c9000 rw-p 00000000 00:00 0
7f2e042c9000-7f2e04487000 r--s 039d0000 fc:02 15736512 /home/user/dorkbox/eclipse/jre/jre/lib/rt.jar
7f2e04487000-7f2e04d88000 rw-p 00000000 00:00 0
7f2e04d88000-7f2e04d89000 ---p 00000000 00:00 0
7f2e04d89000-7f2e04e89000 rw-p 00000000 00:00 0 [stack:26968]
7f2e04e89000-7f2e04e8a000 ---p 00000000 00:00 0
7f2e04e8a000-7f2e04f8a000 rw-p 00000000 00:00 0 [stack:26967]
7f2e04f8a000-7f2e04f8b000 ---p 00000000 00:00 0
7f2e04f8b000-7f2e0508b000 rw-p 00000000 00:00 0 [stack:26966]
7f2e0508b000-7f2e0508c000 ---p 00000000 00:00 0
7f2e0508c000-7f2e0518c000 rw-p 00000000 00:00 0 [stack:26965]
7f2e0518c000-7f2e0518d000 ---p 00000000 00:00 0
7f2e0518d000-7f2e0528d000 rw-p 00000000 00:00 0 [stack:26964]
7f2e0528d000-7f2e0528e000 ---p 00000000 00:00 0
7f2e0528e000-7f2e0538e000 rw-p 00000000 00:00 0 [stack:26963]
7f2e0538e000-7f2e0538f000 ---p 00000000 00:00 0
7f2e0538f000-7f2e0548f000 rw-p 00000000 00:00 0 [stack:26962]
7f2e0548f000-7f2e05490000 ---p 00000000 00:00 0
7f2e05490000-7f2e0559b000 rw-p 00000000 00:00 0 [stack:26961]
7f2e0559b000-7f2e055b9000 rw-p 00000000 00:00 0
7f2e055b9000-7f2e0560a000 rw-p 00000000 00:00 0
7f2e0560a000-7f2e05ac5000 rw-p 00000000 00:00 0
7f2e05ac5000-7f2e05ad0000 rw-p 00000000 00:00 0
7f2e05ad0000-7f2e05aee000 rw-p 00000000 00:00 0
7f2e05aee000-7f2e05b3f000 rw-p 00000000 00:00 0
7f2e05b3f000-7f2e05ff9000 rw-p 00000000 00:00 0
7f2e05ff9000-7f2e06022000 rw-p 00000000 00:00 0
7f2e06022000-7f2e0627f000 rw-p 00000000 00:00 0
7f2e0627f000-7f2e06280000 rw-p 00000000 00:00 0
7f2e06280000-7f2e0629a000 r-xp 00000000 fc:02 15997747 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libzip.so
7f2e0629a000-7f2e0649a000 ---p 0001a000 fc:02 15997747 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libzip.so
7f2e0649a000-7f2e0649b000 rw-p 0001a000 fc:02 15997747 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libzip.so
7f2e0649b000-7f2e064a6000 r-xp 00000000 fc:02 27003108 /lib/x86_64-linux-gnu/libnss_files-2.19.so
7f2e064a6000-7f2e066a5000 ---p 0000b000 fc:02 27003108 /lib/x86_64-linux-gnu/libnss_files-2.19.so
7f2e066a5000-7f2e066a6000 r--p 0000a000 fc:02 27003108 /lib/x86_64-linux-gnu/libnss_files-2.19.so
7f2e066a6000-7f2e066a7000 rw-p 0000b000 fc:02 27003108 /lib/x86_64-linux-gnu/libnss_files-2.19.so
7f2e066a7000-7f2e066b2000 r-xp 00000000 fc:02 27003173 /lib/x86_64-linux-gnu/libnss_nis-2.19.so
7f2e066b2000-7f2e068b1000 ---p 0000b000 fc:02 27003173 /lib/x86_64-linux-gnu/libnss_nis-2.19.so
7f2e068b1000-7f2e068b2000 r--p 0000a000 fc:02 27003173 /lib/x86_64-linux-gnu/libnss_nis-2.19.so
7f2e068b2000-7f2e068b3000 rw-p 0000b000 fc:02 27003173 /lib/x86_64-linux-gnu/libnss_nis-2.19.so
7f2e068b3000-7f2e068ca000 r-xp 00000000 fc:02 27003162 /lib/x86_64-linux-gnu/libnsl-2.19.so
7f2e068ca000-7f2e06ac9000 ---p 00017000 fc:02 27003162 /lib/x86_64-linux-gnu/libnsl-2.19.so
7f2e06ac9000-7f2e06aca000 r--p 00016000 fc:02 27003162 /lib/x86_64-linux-gnu/libnsl-2.19.so
7f2e06aca000-7f2e06acb000 rw-p 00017000 fc:02 27003162 /lib/x86_64-linux-gnu/libnsl-2.19.so
7f2e06acb000-7f2e06acd000 rw-p 00000000 00:00 0
7f2e06acd000-7f2e06ad6000 r-xp 00000000 fc:02 27003161 /lib/x86_64-linux-gnu/libnss_compat-2.19.so
7f2e06ad6000-7f2e06cd5000 ---p 00009000 fc:02 27003161 /lib/x86_64-linux-gnu/libnss_compat-2.19.so
7f2e06cd5000-7f2e06cd6000 r--p 00008000 fc:02 27003161 /lib/x86_64-linux-gnu/libnss_compat-2.19.so
7f2e06cd6000-7f2e06cd7000 rw-p 00009000 fc:02 27003161 /lib/x86_64-linux-gnu/libnss_compat-2.19.so
7f2e06cd7000-7f2e06d00000 r-xp 00000000 fc:02 15997776 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libjava.so
7f2e06d00000-7f2e06f00000 ---p 00029000 fc:02 15997776 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libjava.so
7f2e06f00000-7f2e06f02000 rw-p 00029000 fc:02 15997776 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libjava.so
7f2e06f02000-7f2e06f0f000 r-xp 00000000 fc:02 15997780 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libverify.so
7f2e06f0f000-7f2e0710e000 ---p 0000d000 fc:02 15997780 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libverify.so
7f2e0710e000-7f2e07110000 rw-p 0000c000 fc:02 15997780 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/libverify.so
7f2e07110000-7f2e07117000 r-xp 00000000 fc:02 27003170 /lib/x86_64-linux-gnu/librt-2.19.so
7f2e07117000-7f2e07316000 ---p 00007000 fc:02 27003170 /lib/x86_64-linux-gnu/librt-2.19.so
7f2e07316000-7f2e07317000 r--p 00006000 fc:02 27003170 /lib/x86_64-linux-gnu/librt-2.19.so
7f2e07317000-7f2e07318000 rw-p 00007000 fc:02 27003170 /lib/x86_64-linux-gnu/librt-2.19.so
7f2e07318000-7f2e0741d000 r-xp 00000000 fc:02 27003107 /lib/x86_64-linux-gnu/libm-2.19.so
7f2e0741d000-7f2e0761c000 ---p 00105000 fc:02 27003107 /lib/x86_64-linux-gnu/libm-2.19.so
7f2e0761c000-7f2e0761d000 r--p 00104000 fc:02 27003107 /lib/x86_64-linux-gnu/libm-2.19.so
7f2e0761d000-7f2e0761e000 rw-p 00105000 fc:02 27003107 /lib/x86_64-linux-gnu/libm-2.19.so
7f2e0761e000-7f2e08180000 r-xp 00000000 fc:02 15997803 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/server/libjvm.so
7f2e08180000-7f2e08380000 ---p 00b62000 fc:02 15997803 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/server/libjvm.so
7f2e08380000-7f2e08443000 rw-p 00b62000 fc:02 15997803 /home/user/dorkbox/eclipse/jre/jre/lib/amd64/server/libjvm.so
7f2e08443000-7f2e08484000 rw-p 00000000 00:00 0
7f2e08484000-7f2e0863f000 r-xp 00000000 fc:02 27003167 /lib/x86_64-linux-gnu/libc-2.19.so
7f2e0863f000-7f2e0883e000 ---p 001bb000 fc:02 27003167 /lib/x86_64-linux-gnu/libc-2.19.so
7f2e0883e000-7f2e08842000 r--p 001ba000 fc:02 27003167 /lib/x86_64-linux-gnu/libc-2.19.so
7f2e08842000-7f2e08844000 rw-p 001be000 fc:02 27003167 /lib/x86_64-linux-gnu/libc-2.19.so
7f2e08844000-7f2e08849000 rw-p 00000000 00:00 0
7f2e08849000-7f2e0884c000 r-xp 00000000 fc:02 27003118 /lib/x86_64-linux-gnu/libdl-2.19.so
7f2e0884c000-7f2e08a4b000 ---p 00003000 fc:02 27003118 /lib/x86_64-linux-gnu/libdl-2.19.so
7f2e08a4b000-7f2e08a4c000 r--p 00002000 fc:02 27003118 /lib/x86_64-linux-gnu/libdl-2.19.so
7f2e08a4c000-7f2e08a4d000 rw-p 00003000 fc:02 27003118 /lib/x86_64-linux-gnu/libdl-2.19.so
7f2e08a4d000-7f2e08a63000 r-xp 00000000 fc:02 15736033 /home/user/dorkbox/eclipse/jre/lib/amd64/jli/libjli.so
7f2e08a63000-7f2e08c63000 ---p 00016000 fc:02 15736033 /home/user/dorkbox/eclipse/jre/lib/amd64/jli/libjli.so
7f2e08c63000-7f2e08c64000 rw-p 00016000 fc:02 15736033 /home/user/dorkbox/eclipse/jre/lib/amd64/jli/libjli.so
7f2e08c64000-7f2e08c7d000 r-xp 00000000 fc:02 27003168 /lib/x86_64-linux-gnu/libpthread-2.19.so
7f2e08c7d000-7f2e08e7c000 ---p 00019000 fc:02 27003168 /lib/x86_64-linux-gnu/libpthread-2.19.so
7f2e08e7c000-7f2e08e7d000 r--p 00018000 fc:02 27003168 /lib/x86_64-linux-gnu/libpthread-2.19.so
7f2e08e7d000-7f2e08e7e000 rw-p 00019000 fc:02 27003168 /lib/x86_64-linux-gnu/libpthread-2.19.so
7f2e08e7e000-7f2e08e82000 rw-p 00000000 00:00 0
7f2e08e82000-7f2e08ea5000 r-xp 00000000 fc:02 27003164 /lib/x86_64-linux-gnu/ld-2.19.so
7f2e08ead000-7f2e08eb1000 r--s 0008a000 fc:02 15736510 /home/user/dorkbox/eclipse/jre/jre/lib/jsse.jar
7f2e08eb1000-7f2e08eb4000 r--s 00013000 fc:02 21495935 /home/user/dorkbox/resources/Dependencies/jol-core-0.3.1.jar
7f2e08eb4000-7f2e08ebd000 r--s 0005c000 fc:02 20712771 /home/user/dorkbox/resources/Dependencies/logging/logback-core-1.0.13.jar
7f2e08ebd000-7f2e08ec7000 rw-p 00000000 00:00 0
7f2e08ec7000-7f2e08f7d000 rw-p 00000000 00:00 0
7f2e08f7d000-7f2e08f80000 ---p 00000000 00:00 0
7f2e08f80000-7f2e09083000 rw-p 00000000 00:00 0 [stack:26960]
7f2e09083000-7f2e09084000 rw-p 00000000 00:00 0
7f2e09084000-7f2e09087000 r--s 00011000 fc:02 20060212 /home/user/dorkbox/resources/Dependencies/disruptor/disruptor-3.3.0.jar
7f2e09087000-7f2e09088000 r--s 00002000 fc:02 14945430 /home/user/dorkbox/resources/Dependencies/concurrent-locks/concurrent-locks-1.0.0.jar
7f2e09088000-7f2e09089000 r--s 00004000 fc:02 15728677 /home/user/dorkbox/resources/Dependencies/kryo/reflectasm.jar
7f2e09089000-7f2e0908a000 r--s 0000c000 fc:02 20845186 /home/user/dorkbox/resources/Dependencies/asm/asm-5.0.3.jar
7f2e0908a000-7f2e0908f000 r--s 0003a000 fc:02 20712769 /home/user/dorkbox/resources/Dependencies/logging/logback-classic-1.0.13.jar
7f2e0908f000-7f2e09091000 r--s 0000c000 fc:02 16918224 /home/user/dorkbox/eclipse/plugins/org.hamcrest.core_1.3.0.v201303031735.jar
7f2e09091000-7f2e09097000 r--s 00036000 fc:02 17700265 /home/user/dorkbox/eclipse/plugins/org.junit_4.11.0.v201303080030/junit.jar
7f2e09097000-7f2e09098000 r--s 00005000 fc:02 20712766 /home/user/dorkbox/resources/Dependencies/logging/slf4j-api-1.7.5.jar
7f2e09098000-7f2e090a0000 rw-s 00000000 fc:02 13633475 /tmp/hsperfdata_user/26955
7f2e090a0000-7f2e090a1000 rw-p 00000000 00:00 0
7f2e090a1000-7f2e090a2000 r--p 00000000 00:00 0
7f2e090a2000-7f2e090a4000 rw-p 00000000 00:00 0
7f2e090a4000-7f2e090a5000 r--p 00022000 fc:02 27003164 /lib/x86_64-linux-gnu/ld-2.19.so
7f2e090a5000-7f2e090a6000 rw-p 00023000 fc:02 27003164 /lib/x86_64-linux-gnu/ld-2.19.so
7f2e090a6000-7f2e090a7000 rw-p 00000000 00:00 0
7fff565c7000-7fff565e8000 rw-p 00000000 00:00 0 [stack]
7fff565e8000-7fff565ea000 r-xp 00000000 00:00 0 [vdso]
ffffffffff600000-ffffffffff601000 r-xp 00000000 00:00 0 [vsyscall]
VM Arguments:
jvm_args: -Dfile.encoding=UTF-8
java_command: dorkbox.util.messagebus.MpmcQueueAltPerfTest
Launcher Type: SUN_STANDARD
Environment Variables:
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games
SHELL=/bin/bash
DISPLAY=:0
Signal Handlers:
SIGSEGV: [libjvm.so+0x993980], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGBUS: [libjvm.so+0x993980], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGFPE: [libjvm.so+0x8106e0], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGPIPE: [libjvm.so+0x8106e0], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGXFSZ: [libjvm.so+0x8106e0], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGILL: [libjvm.so+0x8106e0], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGUSR1: SIG_DFL, sa_mask[0]=0x00000000, sa_flags=0x00000000
SIGUSR2: [libjvm.so+0x811bd0], sa_mask[0]=0x00000004, sa_flags=0x10000004
SIGHUP: [libjvm.so+0x812b40], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGINT: [libjvm.so+0x812b40], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGTERM: [libjvm.so+0x812b40], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
SIGQUIT: [libjvm.so+0x812b40], sa_mask[0]=0x7ffbfeff, sa_flags=0x10000004
--------------- S Y S T E M ---------------
OS:jessie/sid
uname:Linux 3.13.0-46-generic #79-Ubuntu SMP Tue Mar 10 20:06:50 UTC 2015 x86_64
libc:glibc 2.19 NPTL 2.19
rlimit: STACK 8192k, CORE 0k, NPROC 123695, NOFILE 4096, AS infinity
load average:1.52 1.02 1.16
/proc/meminfo:
MemTotal: 15863996 kB
MemFree: 1765612 kB
Buffers: 175376 kB
Cached: 1862528 kB
SwapCached: 369940 kB
Active: 8051080 kB
Inactive: 1975324 kB
Active(anon): 7632440 kB
Inactive(anon): 1566592 kB
Active(file): 418640 kB
Inactive(file): 408732 kB
Unevictable: 288 kB
Mlocked: 288 kB
SwapTotal: 33554428 kB
SwapFree: 31429596 kB
Dirty: 192 kB
Writeback: 4 kB
AnonPages: 7811544 kB
Mapped: 3827496 kB
Shmem: 1210532 kB
Slab: 339396 kB
SReclaimable: 234608 kB
SUnreclaim: 104788 kB
KernelStack: 10136 kB
PageTables: 136940 kB
NFS_Unstable: 0 kB
Bounce: 0 kB
WritebackTmp: 0 kB
CommitLimit: 41486424 kB
Committed_AS: 29935756 kB
VmallocTotal: 34359738367 kB
VmallocUsed: 389944 kB
VmallocChunk: 34359337300 kB
HardwareCorrupted: 0 kB
AnonHugePages: 2963456 kB
HugePages_Total: 0
HugePages_Free: 0
HugePages_Rsvd: 0
HugePages_Surp: 0
Hugepagesize: 2048 kB
DirectMap4k: 935372 kB
DirectMap2M: 15263744 kB
DirectMap1G: 0 kB
CPU:total 8 (4 cores per cpu, 2 threads per core) family 6 model 60 stepping 3, cmov, cx8, fxsr, mmx, sse, sse2, sse3, ssse3, sse4.1, sse4.2, popcnt, avx, avx2, aes, erms, ht, tsc, tscinvbit
/proc/cpuinfo:
processor : 0
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 2401.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 0
cpu cores : 4
apicid : 0
initial apicid : 0
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 1
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 800.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 1
cpu cores : 4
apicid : 2
initial apicid : 2
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 2
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 1300.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 2
cpu cores : 4
apicid : 4
initial apicid : 4
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 3
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 800.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 3
cpu cores : 4
apicid : 6
initial apicid : 6
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 4
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 1600.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 0
cpu cores : 4
apicid : 1
initial apicid : 1
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 5
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 800.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 1
cpu cores : 4
apicid : 3
initial apicid : 3
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 6
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 1100.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 2
cpu cores : 4
apicid : 5
initial apicid : 5
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
processor : 7
vendor_id : GenuineIntel
cpu family : 6
model : 60
model name : Intel(R) Core(TM) i7-4700HQ CPU @ 2.40GHz
stepping : 3
microcode : 0x17
cpu MHz : 1300.000
cache size : 6144 KB
physical id : 0
siblings : 8
core id : 3
cpu cores : 4
apicid : 7
initial apicid : 7
fpu : yes
fpu_exception : yes
cpuid level : 13
wp : yes
flags : fpu vme de pse tsc msr pae mce cx8 apic sep mtrr pge mca cmov pat pse36 clflush dts acpi mmx fxsr sse sse2 ss ht tm pbe syscall nx pdpe1gb rdtscp lm constant_tsc arch_perfmon pebs bts rep_good nopl xtopology nonstop_tsc aperfmperf eagerfpu pni pclmulqdq dtes64 monitor ds_cpl vmx est tm2 ssse3 fma cx16 xtpr pdcm pcid sse4_1 sse4_2 x2apic movbe popcnt tsc_deadline_timer aes xsave avx f16c rdrand lahf_lm abm ida arat epb xsaveopt pln pts dtherm tpr_shadow vnmi flexpriority ept vpid fsgsbase tsc_adjust bmi1 avx2 smep bmi2 erms invpcid
bogomips : 4788.47
clflush size : 64
cache_alignment : 64
address sizes : 39 bits physical, 48 bits virtual
power management:
Memory: 4k page, physical 15863996k(1765612k free), swap 33554428k(31429596k free)
vm_info: Java HotSpot(TM) 64-Bit Server VM (24.55-b03) for linux-amd64 JRE (1.7.0_55-b13), built on Mar 17 2014 19:43:58 by "java_re" with gcc 4.3.0 20080428 (Red Hat 4.3.0-8)
time: Sun Apr 12 23:24:36 2015
elapsed time: 0 seconds

281
pom.xml
View File

@ -1,281 +0,0 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<parent>
<groupId>org.sonatype.oss</groupId>
<artifactId>oss-parent</artifactId>
<version>7</version>
</parent>
<modelVersion>4.0.0</modelVersion>
<groupId>net.engio</groupId>
<artifactId>mbassador</artifactId>
<version>1.2.1-SNAPSHOT</version>
<packaging>bundle</packaging>
<name>mbassador</name>
<description>
Mbassador is a fast and flexible message bus system following the publish subscribe pattern.
It is designed for ease of use and aims to be feature rich and extensible
while preserving resource efficiency and performance.
It features:
declarative handler definition via annotations,
sync and/or async message delivery,
weak-references,
message filtering,
ordering of message handlers etc.
</description>
<url>https://github.com/bennidi/mbassador</url>
<licenses>
<license>
<name>MIT license</name>
<url>http://www.opensource.org/licenses/mit-license.php</url>
</license>
</licenses>
<scm>
<url>git@github.com:bennidi/mbassador.git</url>
<connection>scm:git:git@github.com:bennidi/mbassador.git</connection>
<tag>mbassador-1.2.0</tag>
<developerConnection>scm:git:git@github.com:bennidi/mbassador.git</developerConnection>
</scm>
<developers>
<developer>
<id>bennidi</id>
<name>Benjamin Diedrichsen</name>
<timezone>+1</timezone>
<email>b.diedrichsen@googlemail.com</email>
</developer>
</developers>
<properties>
<nazgul-codestyle.version>2.0.1</nazgul-codestyle.version>
<jdk.version>1.6</jdk.version>
<pmd.plugin.version>3.0.1</pmd.plugin.version>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<project.build.java.version>1.6</project.build.java.version>
<github.url>file://${project.basedir}/mvn-local-repo</github.url>
</properties>
<dependencies>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.10</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-log4j12</artifactId>
<version>1.7.5</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>javax.el</groupId>
<artifactId>el-api</artifactId>
<scope>provided</scope>
<version>2.2</version>
</dependency>
<dependency>
<groupId>de.odysseus.juel</groupId>
<artifactId>juel-impl</artifactId>
<version>2.2.7</version>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
<dependency>
<groupId>de.odysseus.juel</groupId>
<artifactId>juel-spi</artifactId>
<version>2.2.7</version>
<scope>runtime</scope>
<optional>true</optional>
</dependency>
</dependencies>
<!-- Local repository (for testing)
<distributionManagement>
<repository>
<id>mbassador-github-repo</id>
<url>${github.url}</url>
</repository>
</distributionManagement>
-->
<build>
<plugins>
<!-- plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-pmd-plugin</artifactId>
<version>${pmd.plugin.version}</version>
<configuration>
<excludeRoots>
<excludeRoot>src/main/generated</excludeRoot>
<excludeRoot>src/test</excludeRoot>
</excludeRoots>
<rulesets>
<ruleset>/codestyle/pmd-rules.xml</ruleset>
</rulesets>
<targetJdk>${jdk.version}</targetJdk>
<sourceEncoding>${project.build.sourceEncoding}</sourceEncoding>
</configuration>
<executions>
<execution>
<goals>
<goal>check</goal>
<goal>cpd-check</goal>
</goals>
</execution>
</executions>
<dependencies>
<dependency>
<groupId>se.jguru.nazgul.tools.codestyle</groupId>
<artifactId>nazgul-codestyle</artifactId>
<version>${nazgul-codestyle.version}</version>
</dependency>
</dependencies>
</plugin -->
<plugin>
<groupId>org.apache.felix</groupId>
<artifactId>maven-bundle-plugin</artifactId>
<version>2.3.7</version>
<extensions>true</extensions>
<configuration>
<instructions>
<Bundle-SymbolicName>${project.groupId}.${project.artifactId}</Bundle-SymbolicName>
<Export-Package>{local-packages}</Export-Package>
</instructions>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<source>${project.build.java.version}</source>
<target>${project.build.java.version}</target>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-release-plugin</artifactId>
<version>2.4</version>
<configuration>
<mavenExecutorId>forked-path</mavenExecutorId>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
<configuration>
<skipTests>false</skipTests>
<excludes>
<!-- exclude the suite which is a convenience class for running all tests from IDE or using scripts -->
<exclude>AllTests.java</exclude>
</excludes>
</configuration>
</plugin>
<!-- bind the source attaching to package phase -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-source-plugin</artifactId>
<executions>
<execution>
<id>attach-sources</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<executions>
<execution>
<id>attach-javadocs</id>
<goals>
<goal>jar</goal>
</goals>
</execution>
</executions>
</plugin>
<!--
These two plugins take care of building and publishing the javadoc, using
mvn clean javadoc:javadoc scm-publish:publish-scm
-->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-javadoc-plugin</artifactId>
<version>2.9.1</version>
<configuration>
<aggregate>true</aggregate>
<show>public</show>
<nohelp>true</nohelp>
<header>mbassador, ${project.version}</header>
<footer>mbassador, ${project.version}</footer>
<doctitle>mbassador, ${project.version}</doctitle>
</configuration>
</plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-scm-publish-plugin</artifactId>
<version>1.0-beta-2</version>
<configuration>
<checkoutDirectory>${project.build.directory}/scmpublish</checkoutDirectory>
<checkinComment>Publishing javadoc for ${project.artifactId}:${project.version}</checkinComment>
<content>${project.reporting.outputDirectory}/apidocs</content>
<skipDeletedFiles>true</skipDeletedFiles>
<pubScmUrl>scm:git:git@github.com:bennidi/mbassador.git</pubScmUrl>
<scmBranch>gh-pages</scmBranch> <!-- branch with static site on github-->
</configuration>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>release-sign-artifacts</id>
<activation>
<property>
<name>performRelease</name>
<value>true</value>
</property>
</activation>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-gpg-plugin</artifactId>
<version>1.4</version>
<executions>
<execution>
<id>sign-artifacts</id>
<phase>verify</phase>
<goals>
<goal>sign</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
</profile>
</profiles>
</project>

View File

@ -68,7 +68,7 @@ public class DisruptorQueue {
try {
// System.err.println("+(" + seq + ") " + message1);
MessageHolder eventJob = ringBuffer.get(seq);
eventJob.messageType = MessageType.ONE;
eventJob.messageType = MessageTypeOLD.ONE;
eventJob.message1 = message1;
// eventJob.message2 = message2;
// eventJob.message3 = message3;

View File

@ -21,7 +21,7 @@ public class EventProcessor2 implements WorkHandlerEarlyRelease<MessageHolder> {
@Override
public void onEvent(long sequence, MessageHolder event) throws Exception {
MessageType messageType = event.messageType;
MessageTypeOLD messageType = event.messageType;
switch (messageType) {
case ONE: {
Object message1 = event.message1;

View File

@ -5,7 +5,7 @@ package com.lmax.disruptor;
* @author dorkbox, llc Date: 2/2/15
*/
public class MessageHolder {
public MessageType messageType = MessageType.ONE;
public MessageTypeOLD messageType = MessageTypeOLD.ONE;
public Object message1 = null;
public Object message2 = null;

View File

@ -3,6 +3,9 @@ package com.lmax.disruptor;
* @author dorkbox, llc
* Date: 2/2/15
*/
public enum MessageType {
ONE, TWO, THREE, ARRAY
public class MessageType {
public static final short ONE = 1;
public static final short TWO = 2;
public static final short THREE = 3;
public static final short ARRAY = 4;
}

View File

@ -0,0 +1,8 @@
package com.lmax.disruptor;
/**
* @author dorkbox, llc
* Date: 2/2/15
*/
public enum MessageTypeOLD {
ONE, TWO, THREE, ARRAY
}

View File

@ -13,6 +13,7 @@ import dorkbox.util.messagebus.common.NamedThreadFactory;
import dorkbox.util.messagebus.common.StrongConcurrentSetV8;
import dorkbox.util.messagebus.common.simpleq.HandlerFactory;
import dorkbox.util.messagebus.common.simpleq.SimpleQueue;
import dorkbox.util.messagebus.common.simpleq.jctools.Pow2;
import dorkbox.util.messagebus.error.IPublicationErrorHandler;
import dorkbox.util.messagebus.error.PublicationError;
import dorkbox.util.messagebus.subscription.Subscription;
@ -83,9 +84,11 @@ public class MultiMBassador implements IMessageBus {
* @param numberOfThreads how many threads to have for dispatching async messages
*/
public MultiMBassador(boolean forceExactMatches, int numberOfThreads) {
if (numberOfThreads < 1) {
numberOfThreads = 1; // at LEAST 1 thread
if (numberOfThreads < 2) {
numberOfThreads = 2; // at LEAST 2 threads
}
numberOfThreads = Pow2.roundToPowerOfTwo(numberOfThreads);
this.numberOfThreads = numberOfThreads;
// this.dispatchQueue = new DisruptorQueue(this, numberOfThreads);
@ -97,7 +100,7 @@ public class MultiMBassador implements IMessageBus {
}
};
this.dispatchQueue = new SimpleQueue(numberOfThreads, 1<<14);
this.dispatchQueue = new SimpleQueue(numberOfThreads);
this.subscriptionManager = new SubscriptionManager(numberOfThreads);
this.threads = new ArrayDeque<Thread>(numberOfThreads);
@ -108,31 +111,18 @@ public class MultiMBassador implements IMessageBus {
Runnable runnable = new Runnable() {
@Override
public void run() {
// TransferQueue<Runnable> IN_QUEUE = MultiMBassador.this.dispatchQueue;
SimpleQueue IN_QUEUE = MultiMBassador.this.dispatchQueue;
// Runnable event = null;
int spins;
Object message1;
while (true) {
// spins = maxSpins;
// while ((event = IN_QUEUE.poll()) == null) {
// if (spins > 100) {
// --spins;
// } else if (spins > 0) {
// --spins;
// LockSupport.parkNanos(1L);
// } else {
try {
while (true) {
message1 = IN_QUEUE.take();
// break;
// }
// }
// IN_QUEUE.release(event);
// event.run();
publish(message1);
}
}
} catch (InterruptedException e) {
Thread.interrupted();
return;
}
}
};
@ -441,8 +431,15 @@ public class MultiMBassador implements IMessageBus {
// }
// };
// this.dispatchQueue.transfer(runnable);
this.dispatchQueue.put(message);
try {
// this.dispatchQueue.transfer(runnable);
this.dispatchQueue.put(message);
} catch (InterruptedException e) {
handlePublicationError(new PublicationError()
.setMessage("Error while adding an asynchronous message")
.setCause(e)
.setPublishedObject(message));
}
}
}
@ -456,7 +453,14 @@ public class MultiMBassador implements IMessageBus {
}
};
try {
this.dispatchQueue.put(runnable);
} catch (InterruptedException e) {
handlePublicationError(new PublicationError()
.setMessage("Error while adding an asynchronous message")
.setCause(e)
.setPublishedObject(message1, message2));
}
}
}
@ -471,7 +475,14 @@ public class MultiMBassador implements IMessageBus {
};
try {
this.dispatchQueue.put(runnable);
} catch (InterruptedException e) {
handlePublicationError(new PublicationError()
.setMessage("Error while adding an asynchronous message")
.setCause(e)
.setPublishedObject(message1, message2, message3));
}
}
}

View File

@ -12,8 +12,8 @@ public final class MpmcExchangerQueue extends MpmcArrayQueueConsumerField<Node>
static {
try {
TYPE_OFFSET = UNSAFE.objectFieldOffset(Node.class.getField("nodeType"));
ITEM1_OFFSET = UNSAFE.objectFieldOffset(Node.class.getField("item1"));
THREAD = UNSAFE.objectFieldOffset(Node.class.getField("thread"));
ITEM1_OFFSET = UNSAFE.objectFieldOffset(Node.class.getField("item1"));
THREAD = UNSAFE.objectFieldOffset(Node.class.getField("thread"));
} catch (NoSuchFieldException e) {
throw new RuntimeException(e);
}

View File

@ -1,32 +1,33 @@
package dorkbox.util.messagebus.common.simpleq;
import com.lmax.disruptor.MessageType;
// mpmc sparse.shift = 2, for this to be fast.
abstract class PrePad {
volatile long z0, z1, z2, z4, z5, z6 = 7L;
// volatile long z0, z1, z2, z4, z5, z6 = 7L;
}
abstract class ColdItems extends PrePad {
public short nodeType = NodeType.FREE;
abstract class ColdItems {
public short type = MessageType.ONE;
public boolean isConsumer = false;
public Object item1 = null;
public Object item2 = null;
public Object item3 = null;
public Object[] item4 = null;
}
abstract class Pad1 extends ColdItems {
volatile long z0, z1, z2, z4, z5, z6 = 7L;
abstract class Pad0 extends ColdItems {
// volatile long z0, z1, z2, z4, z5, z6 = 7L;
}
abstract class HotItem1 extends Pad1 {
public Thread thread;
abstract class HotItem1 extends ColdItems {
public volatile Thread thread;
}
public class Node extends HotItem1 {
// post-padding
volatile long z0, z1, z2, z4, z5, z6 = 7L;
// volatile long z0, z1, z2, z4, z5, z6 = 7L;
public Node() {
}

View File

@ -0,0 +1,8 @@
package dorkbox.util.messagebus.common.simpleq;
public class NodeState {
public static final short FREE = 0;
public static final short CANCELLED = 1;
private NodeState() {}
}

View File

@ -1,13 +0,0 @@
package dorkbox.util.messagebus.common.simpleq;
public class NodeType {
public static final short FREE = 0;
public static final short P_INIT = 1;
public static final short P_DONE = 2;
public static final short C_INIT = 1;
public static final short C_DONE = 2;
private NodeType() {}
}

View File

@ -5,30 +5,95 @@ import static dorkbox.util.messagebus.common.simpleq.jctools.UnsafeAccess.UNSAFE
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;
import com.lmax.disruptor.MessageHolder;
import dorkbox.util.messagebus.common.simpleq.jctools.MpmcArrayQueueConsumerField;
public final class SimpleQueue extends MpmcArrayQueueConsumerField<Node> {
private final static long THREAD;
private static final long ITEM1_OFFSET;
private static final long TYPE;
private static final long CONSUMER;
// private final static long NODE_OFFSET;
private final static long MESSAGE1_OFFSET;
static {
try {
// NODE_OFFSET = UNSAFE.objectFieldOffset(Node.class.getField("n"));
MESSAGE1_OFFSET = UNSAFE.objectFieldOffset(Node.class.getField("item1"));
} catch (NoSuchFieldException e) {
throw new RuntimeException(e);
}
// Prevent rare disastrous classloading in first call to LockSupport.park.
// See: https://bugs.openjdk.java.net/browse/JDK-8074773
@SuppressWarnings("unused")
Class<?> ensureLoaded = LockSupport.class;
LockSupport.unpark(Thread.currentThread());
static {
try {
CONSUMER = UNSAFE.objectFieldOffset(Node.class.getField("isConsumer"));
THREAD = UNSAFE.objectFieldOffset(Node.class.getField("thread"));
TYPE = UNSAFE.objectFieldOffset(Node.class.getField("type"));
ITEM1_OFFSET = UNSAFE.objectFieldOffset(Node.class.getField("item1"));
} catch (NoSuchFieldException e) {
throw new RuntimeException(e);
}
}
private static final void spIsConsumer(Object node, boolean value) {
UNSAFE.putBoolean(node, CONSUMER, value);
}
private static final boolean lpIsConsumer(Object node) {
return UNSAFE.getBoolean(node, CONSUMER);
}
private static final boolean lvIsConsumer(Object node) {
return UNSAFE.getBooleanVolatile(node, CONSUMER);
}
private static final void spType(Object node, short type) {
UNSAFE.putShort(node, TYPE, type);
}
private static final short lpType(Object node) {
return UNSAFE.getShort(node, TYPE);
}
private static final void soItem1(Object node, Object item) {
UNSAFE.putOrderedObject(node, ITEM1_OFFSET, item);
}
private static final void spItem1(Object node, Object item) {
UNSAFE.putObject(node, ITEM1_OFFSET, item);
}
private static final Object lvItem1(Object node) {
return UNSAFE.getObjectVolatile(node, ITEM1_OFFSET);
}
private static final Object lpItem1(Object node) {
return UNSAFE.getObject(node, ITEM1_OFFSET);
}
private static final Thread lvThread(Object node) {
return (Thread) UNSAFE.getObjectVolatile(node, THREAD);
}
private static final Thread lpThread(Object node) {
return (Thread) UNSAFE.getObject(node, THREAD);
}
private static final Object cancelledMarker = new Object();
private static final boolean lpIsCanceled(Object node) {
return cancelledMarker == UNSAFE.getObject(node, THREAD);
}
private static final void spIsCancelled(Object node) {
UNSAFE.putObject(node, THREAD, cancelledMarker);
}
private static final void soThread(Object node, Thread newValue) {
UNSAFE.putOrderedObject(node, THREAD, newValue);
}
private static final void spThread(Object node, Thread newValue) {
UNSAFE.putObject(node, THREAD, newValue);
}
private static final boolean casThread(Object node, Object expect, Object newValue) {
return UNSAFE.compareAndSwapObject(node, THREAD, expect, newValue);
}
/** The number of CPUs */
private static final int NCPU = Runtime.getRuntime().availableProcessors();
@ -41,28 +106,36 @@ public final class SimpleQueue extends MpmcArrayQueueConsumerField<Node> {
*/
private static final int SPINS = NCPU == 1 ? 0 : 600; // orig: 2000
private static final int SIZE = 1<<14;
/**
* The number of times to spin before blocking in timed waits.
* The value is empirically derived -- it works well across a
* variety of processors and OSes. Empirically, the best value
* seems not to vary with number of CPUs (beyond 2) so is just
* a constant.
*/
static final int maxTimedSpins = NCPU < 2 ? 0 : 32;
static final int negMaxTimedSpins = -maxTimedSpins;
/**
* The number of times to spin before blocking in untimed waits.
* This is greater than timed value because untimed waits spin
* faster since they don't need to check times on each spin.
*/
static final int maxUntimedSpins = maxTimedSpins * 16;
/**
* The number of nanoseconds for which it is faster to spin
* rather than to use timed park. A rough estimate suffices.
*/
static final long spinForTimeoutThreshold = 1000L;
long p40, p41, p42, p43, p44, p45, p46;
long p30, p31, p32, p33, p34, p35, p36, p37;
// // EMPTY == TRUE
// if (currentConsumerIndex == currentProducerIndex) {
// // automatically park, since we are the first one on the Q
// }
//
//
// // other consumers may have grabbed the element, or queue might be empty
// Node fbject = lpElement(calcElementOffset(currentConsumerIndex));
//
//
private final int numberConsumerThreads;
public SimpleQueue(int numberConsumerThreads, int size) {
public SimpleQueue(final int size) {
super(size);
this.numberConsumerThreads = numberConsumerThreads;
// pre-fill our data structures
@ -77,105 +150,41 @@ public final class SimpleQueue extends MpmcArrayQueueConsumerField<Node> {
}
}
/**
* PRODUCER
*/
public void put(Object item) {
public void put(Object item) throws InterruptedException {
xfer(item, false, 0);
}
/**
* CONSUMER
*/
public Object take() throws InterruptedException {
return xfer(null, false, 0);
}
private Object xfer(Object item, boolean timed, long nanos) throws InterruptedException {
boolean isConsumer = item == null;
// local load of field to avoid repeated loads after volatile reads
final long mask = this.mask;
final long capacity = this.mask + 1;
final long[] sBuffer = this.sequenceBuffer;
// long currentConsumerIndex;
long currentProducerIndex;
long pSeqOffset;
long cIndex = Long.MAX_VALUE;// start with bogus value, hope we don't need it
while (true) {
// Order matters!
// Loading consumer before producer allows for producer increments after consumer index is read.
// This ensures this method is conservative in it's estimate. Note that as this is an MPMC there is
// nothing we can do to make this an exact method.
// currentConsumerIndex = lvConsumerIndex(); // LoadLoad
currentProducerIndex = lvProducerIndex(); // LoadLoad
pSeqOffset = calcSequenceOffset(currentProducerIndex, mask);
final long seq = lvSequence(sBuffer, pSeqOffset); // LoadLoad
final long delta = seq - currentProducerIndex;
if (delta == 0) {
// this is expected if we see this first time around
if (casProducerIndex(currentProducerIndex, currentProducerIndex + 1)) {
// Successful CAS: full barrier
// on 64bit(no compressed oops) JVM this is the same as seqOffset
final long offset = calcElementOffset(currentProducerIndex, mask);
Node lpElement = lpElement(offset);
setMessage1(lpElement, item);
// lpspElement(offset, fakeVal, NODE_OFFSET, MESSAGE1_OFFSET);
// spElement(offset, item);
// lpElement(offset);
// ((Node)e).setMessage1(null);
// node.setMessage1(((Node)item).getMessage1());
// Object lpElement = lpElement(offset);
// setMessage1(lpElement, 445);
// this.fakeNode.setMessage1(Integer.valueOf(12));
// e.setMessage1(((Node)item).item);
////// NodeState2 message1 = e.getMessage1();
// increment sequence by 1, the value expected by consumer
// (seeing this value from a producer will lead to retry 2)
soSequence(sBuffer, pSeqOffset, currentProducerIndex + 1); // StoreStore
return;
}
// failed cas, retry 1
} else if (delta < 0 && // poll has not moved this value forward
currentProducerIndex - capacity <= cIndex && // test against cached cIndex
currentProducerIndex - capacity <= (cIndex = lvConsumerIndex())) { // test against latest cIndex
// Extra check required to ensure [Queue.offer == false iff queue is full]
// return null;
busySpin();
}
// another producer has moved the sequence by one, retry 2
// only producer will busySpin if contention
// busySpin();
}
}
private static final void setMessage1(Object node, Object item) {
// final Object o = UNSAFE.getObject(node, NODE_OFFSET);
UNSAFE.putObject(node, MESSAGE1_OFFSET, item);
}
private static final Object getMessage1(Object node) {
// final Object o = UNSAFE.getObject(node, NODE_OFFSET);
return UNSAFE.getObject(node, MESSAGE1_OFFSET);
}
/**
* CONSUMER
* @return null iff empty
*/
public Object take() {
// local load of field to avoid repeated loads after volatile reads
final long mask = this.mask;
final long[] sBuffer = this.sequenceBuffer;
long currentConsumerIndex;
// long currentProducerIndex;
long currentProducerIndex;
long cSeqOffset;
long pIndex = -1; // start with bogus value, hope we don't need it
long pSeqOffset;
long pSeq;
long pDelta = -1;
boolean sameMode = false;
boolean empty = false;
while (true) {
// Order matters!
@ -183,224 +192,195 @@ public final class SimpleQueue extends MpmcArrayQueueConsumerField<Node> {
// This ensures this method is conservative in it's estimate. Note that as this is an MPMC there is
// nothing we can do to make this an exact method.
currentConsumerIndex = lvConsumerIndex(); // LoadLoad
// currentProducerIndex = lvProducerIndex(); // LoadLoad
currentProducerIndex = lvProducerIndex(); // LoadLoad
// empty or same mode
// check what was last placed on the queue
if (currentProducerIndex == currentConsumerIndex) {
empty = true;
} else {
final long previousProducerIndex = currentProducerIndex - 1;
cSeqOffset = calcSequenceOffset(currentConsumerIndex, mask);
final long seq = lvSequence(sBuffer, cSeqOffset); // LoadLoad
final long delta = seq - (currentConsumerIndex + 1);
final long ppSeqOffset = calcSequenceOffset(previousProducerIndex, mask);
final long ppSeq = lvSequence(sBuffer, ppSeqOffset); // LoadLoad
final long ppDelta = ppSeq - previousProducerIndex;
if (delta == 0) {
if (casConsumerIndex(currentConsumerIndex, currentConsumerIndex + 1)) {
// Successful CAS: full barrier
if (ppDelta == 1) {
// same mode check
// on 64bit(no compressed oops) JVM this is the same as seqOffset
final long offset = calcElementOffset(currentConsumerIndex, mask);
final Object e = lpElement(offset);
Object item = getMessage1(e);
// final Node node = (Node)e;
//
// Object item = node.getMessage1();
// soElement(offset, null);
final long offset = calcElementOffset(previousProducerIndex, mask);
Object element = lpElement(offset);
sameMode = lpIsConsumer(element) == isConsumer;
} else if (ppDelta < 1 && // slot has not been moved by producer
currentConsumerIndex >= currentProducerIndex && // test against cached pIndex
currentConsumerIndex == (currentProducerIndex = lvProducerIndex())) { // update pIndex if we must
// Move sequence ahead by capacity, preparing it for next offer
// (seeing this value from a consumer will lead to retry 2)
soSequence(sBuffer, cSeqOffset, currentConsumerIndex + mask + 1); // StoreStore
return item;
// is empty
empty = true;
} else {
// hasn't been moved yet. retry 2
busySpin();
continue;
}
// failed cas, retry 1
} else if (delta < 0 && // slot has not been moved by producer
currentConsumerIndex >= pIndex && // test against cached pIndex
currentConsumerIndex == (pIndex = lvProducerIndex())) { // update pIndex if we must
// strict empty check, this ensures [Queue.poll() == null iff isEmpty()]
// return null;
// contention. we WILL have data in the Q, we just got to it too quickly
busySpin();
}
// another consumer beat us and moved sequence ahead, retry 2
// only producer busyspins
}
}
if (empty || sameMode) {
// push+park onto queue
// we add ourselves to the queue and wait
pSeqOffset = calcSequenceOffset(currentProducerIndex, mask);
pSeq = lvSequence(sBuffer, pSeqOffset); // LoadLoad
pDelta = pSeq - currentProducerIndex;
if (pDelta == 0) {
// this is expected if we see this first time around
final long nextProducerIndex = currentProducerIndex + 1;
if (casProducerIndex(currentProducerIndex, nextProducerIndex)) {
// Successful CAS: full barrier
// it is possible that two threads check the queue at the exact same time,
// BOTH can think that the queue is empty, resulting in a deadlock between threads
// it is ALSO possible that the consumer pops the previous node, and so we thought it was not-empty, when
// in reality, it is.
currentConsumerIndex = lvConsumerIndex();
if (empty && currentProducerIndex != currentConsumerIndex) {
// RESET the push of this element.
empty = false;
casProducerIndex(nextProducerIndex, currentProducerIndex);
continue;
} else if (sameMode && currentProducerIndex == currentConsumerIndex) {
sameMode = false;
casProducerIndex(nextProducerIndex, currentProducerIndex);
continue;
}
// on 64bit(no compressed oops) JVM this is the same as seqOffset
final long offset = calcElementOffset(currentProducerIndex, mask);
final Object element = lpElement(offset);
spIsConsumer(element, isConsumer);
spThread(element, Thread.currentThread());
if (isConsumer) {
// increment sequence by 1, the value expected by consumer
// (seeing this value from a producer will lead to retry 2)
soSequence(sBuffer, pSeqOffset, nextProducerIndex); // StoreStore
public void putOLD(Object message1) throws InterruptedException {
// // decrement count
// // <0: no consumers available, add to Q, park and wait
// // >=0: consumers available, get one from the parking lot
//
// Thread myThread = Thread.currentThread();
// for (;;) {
// final int count = this.currentCount.get();
// if (this.currentCount.compareAndSet(count, count - 1)) {
// if (count <= 0) {
// // <=0: no consumers available (PUSH_P, PARK_P)
// Node<M> producer = this.producersWaiting.put();
// if (producer == null || producer.item == null) {
// System.err.println("KAPOW");
// }
// producer.item.message1 = message1;
//
// if (!park(producer, myThread)) {
// throw new InterruptedException();
// }
//
// return;
// } else {
// // >0: consumers available (TAKE_C, UNPARK_C)
// Node<M> consumer = this.consumersWaiting.take();
// while (consumer == null) {
//// busySpin();
// consumer = this.consumersWaiting.take();
// }
//
// consumer.item.message1 = message1;
//
// unpark(consumer, myThread);
// return;
// }
// }
//
// // contention
// busySpin();
// }
}
public void takeOLD(MessageHolder item) throws InterruptedException {
// // increment count
// // >=0: no producers available, park and wait
// // <0: producers available, get one from the Q
//
// Thread myThread = Thread.currentThread();
// for (;;) {
// final int count = this.currentCount.get();
// if (this.currentCount.compareAndSet(count, count + 1)) {
// if (count >= 0) {
// // >=0: no producers available (PUT_C, PARK_C)
// Node<M> consumer = this.consumersWaiting.put();
//
// if (!park(consumer, myThread)) {
// throw new InterruptedException();
// }
// if (consumer.item == null || consumer.item.message1 == null) {
// System.err.println("KAPOW");
// }
// item.message1 = consumer.item.message1;
//
// return;
// } else {
// // <0: producers available (TAKE_P, UNPARK_P)
// Node<M> producer = this.producersWaiting.take();
// while (producer == null) {
//// busySpin();
// producer = this.producersWaiting.take();
// }
//
// item.message1 = producer.item.message1;
// unpark(producer, myThread);
//
// if (item.message1 == null) {
// System.err.println("KAPOW");
// }
//
// return;
// }
// }
//
// // contention
// busySpin();
// }
}
/**
* @param myThread
* @return false if we were interrupted, true if we were unparked by another thread
*/
private boolean park(Node myNode, Thread myThread) {
PaddedObject<Thread> waiter = myNode.waiter;
Thread thread;
for (;;) {
thread = waiter.get();
if (waiter.compareAndSet(thread, myThread)) {
if (thread == null) {
// busy spin for the amount of time (roughly) of a CPU context switch
int spins = SPINS;
for (;;) {
if (spins > 0) {
--spins;
} else if (waiter.get() != myThread) {
break;
// now we wait
park(element, timed, nanos);
return lvItem1(element);
} else {
// park can return for NO REASON. Subsequent loops will hit this if it has not been ACTUALLY unlocked.
LockSupport.park();
if (myThread.isInterrupted()) {
waiter.set(null);
return false;
}
break;
spItem1(element, item);
// increment sequence by 1, the value expected by consumer
// (seeing this value from a producer will lead to retry 2)
soSequence(sBuffer, pSeqOffset, nextProducerIndex); // StoreStore
// now we wait
park(element, timed, nanos);
return null;
}
}
// do {
// // park can return for NO REASON. Subsequent loops will hit this if it has not been ACTUALLY unlocked.
// LockSupport.park();
// if (myThread.isInterrupted()) {
// myNode.waiter.set(null);
// return false;
// }
// } while (myNode.waiter.get() == myThread);
waiter.set(null);
return true;
} else if (thread != myThread) {
// no parking
return true;
} else {
// contention
// failed cas, retry 1
} else if (pDelta < 0 && // poll has not moved this value forward
currentProducerIndex - capacity <= currentConsumerIndex && // test against cached cIndex
currentProducerIndex - capacity <= (currentConsumerIndex = lvConsumerIndex())) { // test against latest cIndex
// Extra check required to ensure [Queue.offer == false iff queue is full]
// return;
busySpin();
}
}
}
}
} else {
// complimentary mode
/**
* Unparks the other node (if it was waiting)
*/
private void unpark(Node otherNode, Thread myThread) {
PaddedObject<Thread> waiter = otherNode.waiter;
Thread thread;
// get item
cSeqOffset = calcSequenceOffset(currentConsumerIndex, mask);
final long cSeq = lvSequence(sBuffer, cSeqOffset); // LoadLoad
final long nextConsumerIndex = currentConsumerIndex + 1;
final long cDelta = cSeq - nextConsumerIndex;
for (;;) {
thread = waiter.get();
if (waiter.compareAndSet(thread, myThread)) {
if (thread == null) {
// no parking
return;
} else if (thread != myThread) {
// park will always set the waiter back to null
LockSupport.unpark(thread);
return;
} else {
// contention
if (cDelta == 0) {
// on 64bit(no compressed oops) JVM this is the same as seqOffset
final long offset = calcElementOffset(currentConsumerIndex, mask);
final Object element = lpElement(offset);
final Thread thread = lpThread(element);
if (isConsumer) {
if (thread == null || // is cancelled/fulfilled already
!casThread(element, thread, null)) { // failed cas state
// pop off queue
if (casConsumerIndex(currentConsumerIndex, nextConsumerIndex)) {
// Successful CAS: full barrier
soSequence(sBuffer, cSeqOffset, nextConsumerIndex + mask); // StoreStore
}
continue;
}
// success
Object item1 = lpItem1(element);
// pop off queue
if (casConsumerIndex(currentConsumerIndex, nextConsumerIndex)) {
// Successful CAS: full barrier
LockSupport.unpark(thread);
soSequence(sBuffer, cSeqOffset, nextConsumerIndex + mask); // StoreStore
return item1;
}
continue;
} else {
soItem1(element, item);
if (thread == null || // is cancelled/fulfilled already
!casThread(element, thread, null)) { // failed cas state
// pop off queue
if (casConsumerIndex(currentConsumerIndex, nextConsumerIndex)) {
// Successful CAS: full barrier
soSequence(sBuffer, cSeqOffset, nextConsumerIndex + mask); // StoreStore
}
continue;
}
// success
// pop off queue
if (casConsumerIndex(currentConsumerIndex, nextConsumerIndex)) {
// Successful CAS: full barrier
LockSupport.unpark(thread);
soSequence(sBuffer, cSeqOffset, nextConsumerIndex + mask); // StoreStore
return null;
}
// lost CAS
busySpin();
continue;
}
} else if (cDelta < 0 && // slot has not been moved by producer
currentConsumerIndex >= currentProducerIndex && // test against cached pIndex
currentConsumerIndex == (currentProducerIndex = lvProducerIndex())) { // update pIndex if we must
// strict empty check, this ensures [Queue.poll() == null iff isEmpty()]
// return null;
busySpin();
}
}
// contention.
busySpin();
}
}
private static final void busySpin() {
// busy spin for the amount of time (roughly) of a CPU context switch
int spins = SPINS;
int spins = maxUntimedSpins;
for (;;) {
if (spins > 0) {
--spins;
@ -410,19 +390,73 @@ public final class SimpleQueue extends MpmcArrayQueueConsumerField<Node> {
}
}
public boolean hasPendingMessages() {
// count the number of consumers waiting, it should be the same as the number of threads configured
// return this.consumersWaiting.size() == this.numberConsumerThreads;
/**
* @param myThread
* @return
* @return false if we were interrupted, true if we were unparked by another thread
*/
private static final boolean park(Object myNode, boolean timed, long nanos) throws InterruptedException {
// long lastTime = timed ? System.nanoTime() : 0;
// int spins = timed ? maxTimedSpins : maxUntimedSpins;
int spins = maxUntimedSpins;
Thread myThread = Thread.currentThread();
// Order matters!
// Loading consumer before producer allows for producer increments after consumer index is read.
// This ensures this method is conservative in it's estimate. Note that as this is an MPMC there is
// nothing we can do to make this an exact method.
return lvConsumerIndex() != lvProducerIndex();
// if (timed) {
// long now = System.nanoTime();
// nanos -= now - lastTime;
// lastTime = now;
// if (nanos <= 0) {
//// s.tryCancel(e);
// continue;
// }
// }
// busy spin for the amount of time (roughly) of a CPU context switch
// then park (if necessary)
int spin = spins;
for (;;) {
if (lvThread(myNode) == null) {
return true;
} else if (spin > 0) {
--spin;
} else if (spin > negMaxTimedSpins) {
LockSupport.parkNanos(1);
} else {
// park can return for NO REASON. Subsequent loops will hit this if it has not been ACTUALLY unlocked.
LockSupport.park();
if (myThread.isInterrupted()) {
casThread(myNode, myThread, null);
return false;
}
}
}
}
public void tryTransfer(Runnable runnable, long timeout, TimeUnit unit) throws InterruptedException {
}
// /**
// * Unparks the other node (if it was waiting)
// */
// private static final void unpark(Object otherNode) {
// Thread myThread = Thread.currentThread();
// Thread thread;
//
// for (;;) {
// thread = getThread(otherNode);
// if (threadCAS(otherNode, thread, myThread)) {
// if (thread == null) {
// // no parking (UNPARK won the race)
// return;
// } else if (thread != myThread) {
// // park will always set the waiter back to null
// LockSupport.unpark(thread);
// return;
// } else {
// // contention
// busySpin();
// }
// }
// }
// }
@Override
public boolean offer(Node message) {
@ -443,4 +477,24 @@ public final class SimpleQueue extends MpmcArrayQueueConsumerField<Node> {
public int size() {
return 0;
}
@Override
public boolean isEmpty() {
// Order matters!
// Loading consumer before producer allows for producer increments after consumer index is read.
// This ensures this method is conservative in it's estimate. Note that as this is an MPMC there is
// nothing we can do to make this an exact method.
return lvConsumerIndex() == lvProducerIndex();
}
public boolean hasPendingMessages() {
// count the number of consumers waiting, it should be the same as the number of threads configured
// return this.consumersWaiting.size() == this.numberConsumerThreads;
return false;
}
public void tryTransfer(Runnable runnable, long timeout, TimeUnit unit) throws InterruptedException {
// TODO Auto-generated method stub
}
}

View File

@ -41,7 +41,7 @@ abstract class ConcurrentCircularArrayQueueL0Pad<E> extends AbstractQueue<E> imp
* @param <E>
*/
public abstract class ConcurrentCircularArrayQueue<E> extends ConcurrentCircularArrayQueueL0Pad<E> {
protected static final int SPARSE_SHIFT = Integer.getInteger("sparse.shift", 2);
protected static final int SPARSE_SHIFT = Integer.getInteger("sparse.shift", 0);
protected static final int BUFFER_PAD;
private static final long REF_ARRAY_BASE;
private static final int REF_ELEMENT_SHIFT;

View File

@ -101,11 +101,11 @@ public class MpmcArrayQueue<T> extends MpmcArrayQueueConsumerField<T> {
currentProducerIndex - capacity <= cIndex && // test against cached cIndex
currentProducerIndex - capacity <= (cIndex = lvConsumerIndex())) { // test against latest cIndex
// Extra check required to ensure [Queue.offer == false iff queue is full]
// return false;
busySpin();
return false;
}
// another producer has moved the sequence by one, retry 2
busySpin();
}
}
@ -151,11 +151,11 @@ public class MpmcArrayQueue<T> extends MpmcArrayQueueConsumerField<T> {
currentConsumerIndex >= pIndex && // test against cached pIndex
currentConsumerIndex == (pIndex = lvProducerIndex())) { // update pIndex if we must
// strict empty check, this ensures [Queue.poll() == null iff isEmpty()]
// return null;
busySpin();
return null;
}
// another consumer beat us and moved sequence ahead, retry 2
// only producer will busy spin
}
}

View File

@ -18,12 +18,13 @@ package dorkbox.util.messagebus;
import org.openjdk.jol.info.ClassLayout;
import org.openjdk.jol.util.VMSupport;
import dorkbox.util.messagebus.common.simpleq.MpmcExchangerQueue;
import dorkbox.util.messagebus.common.simpleq.Node;
import dorkbox.util.messagebus.common.simpleq.jctools.MpmcArrayQueue;
public class MpmcQueueAltPerfTest {
// 15 == 32 * 1024
public static final int REPETITIONS = Integer.getInteger("reps", 50) * 1000 * 1000;
public static final Integer TEST_VALUE = Integer.valueOf(777);
public static final int QUEUE_CAPACITY = 1 << Integer.getInteger("pow2.capacity", 17);
@ -31,10 +32,8 @@ public class MpmcQueueAltPerfTest {
System.out.println(VMSupport.vmDetails());
System.out.println(ClassLayout.parseClass(Node.class).toPrintable());
System.out.println("capacity:" + QUEUE_CAPACITY + " reps:" + REPETITIONS);
final MpmcExchangerQueue queue = new MpmcExchangerQueue(QUEUE_CAPACITY);
final MpmcArrayQueue<Node> queue = new MpmcArrayQueue<Node>(QUEUE_CAPACITY);
final long[] results = new long[20];
for (int i = 0; i < 20; i++) {
@ -49,19 +48,20 @@ public class MpmcQueueAltPerfTest {
System.out.format("summary,QueuePerfTest,%s,%d\n", queue.getClass().getSimpleName(), sum / 10);
}
private static long performanceRun(int runNumber, MpmcArrayQueue<Node> queue) throws Exception {
Producer p = new Producer(queue);
Thread thread = new Thread(p);
thread.start(); // producer will timestamp start
private static long performanceRun(int runNumber, MpmcExchangerQueue queue) throws Exception {
// for (int i=0;i<CONCURRENCY_LEVEL;i++) {
Producer p = new Producer(queue);
Thread thread = new Thread(p);
thread.start(); // producer will timestamp start
// }
MpmcExchangerQueue consumer = queue;
Object result;
MpmcArrayQueue<Node> consumer = queue;
Node result;
int i = REPETITIONS;
int queueEmpty = 0;
do {
result = consumer.take();
while (null == (result = consumer.poll())) {
queueEmpty++;
Thread.yield();
}
} while (0 != --i);
long end = System.nanoTime();
@ -69,30 +69,46 @@ public class MpmcQueueAltPerfTest {
long duration = end - p.start;
long ops = REPETITIONS * 1000L * 1000L * 1000L / duration;
String qName = queue.getClass().getSimpleName();
System.out.format("%d - ops/sec=%,d - %s result=%d\n", runNumber, ops, qName, result);
System.out.format("%d - ops/sec=%,d - %s result=%d failed.poll=%d failed.offer=%d\n", runNumber, ops,
qName, result.item1, queueEmpty, p.queueFull);
return ops;
}
private static final Integer val = Integer.valueOf(234);
@SuppressWarnings("rawtypes")
public static class Producer implements Runnable {
private final MpmcExchangerQueue queue;
private final MpmcArrayQueue queue;
int queueFull = 0;
long start;
public Producer(MpmcExchangerQueue queue) {
public Producer(MpmcArrayQueue queue) {
this.queue = queue;
}
@Override
public void run() {
MpmcExchangerQueue producer = this.queue;
int i = REPETITIONS;
long s = System.nanoTime();
do {
producer.put(val);
} while (0 != --i);
this.start = s;
MpmcArrayQueue producer = this.queue;
int i = REPETITIONS;
int f = 0;
long s = System.nanoTime();
MpmcArrayQueue<Node> pool = new MpmcArrayQueue<Node>(2);
pool.offer(new Node());
pool.offer(new Node());
Node node;
do {
node = pool.poll();
node.item1 = TEST_VALUE;
while (!producer.offer(node)) {
Thread.yield();
f++;
}
pool.offer(node);
} while (0 != --i);
this.queueFull = f;
this.start = s;
}
}
}

View File

@ -30,7 +30,7 @@ public class MpmcQueueBaselineNodePerfTest {
public static void main(final String[] args) throws Exception {
System.out.println(VMSupport.vmDetails());
System.out.println(ClassLayout.parseClass(Integer.class).toPrintable());
System.out.println(ClassLayout.parseClass(Node.class).toPrintable());
System.out.println("capacity:" + QUEUE_CAPACITY + " reps:" + REPETITIONS);
final MpmcArrayQueue<Node> queue = new MpmcArrayQueue<Node>(QUEUE_CAPACITY);

View File

@ -6,7 +6,6 @@ package dorkbox.util.messagebus;
import junit.framework.Assert;
import dorkbox.util.messagebus.annotations.Handler;
import dorkbox.util.messagebus.common.ConcurrentExecutor;
import dorkbox.util.messagebus.common.simpleq.SimpleQueue;
import dorkbox.util.messagebus.error.IPublicationErrorHandler;
import dorkbox.util.messagebus.error.PublicationError;
@ -19,7 +18,7 @@ public class PerformanceTest {
public static final int QUEUE_CAPACITY = 1 << Integer.getInteger("pow2.capacity", 17);
public static final int CONCURRENCY_LEVEL = 1;
public static final int CONCURRENCY_LEVEL = 2;
private static long count = 0;
@ -32,72 +31,6 @@ public class PerformanceTest {
};
public static void main(String[] args) throws Exception {
// testSpeed();
tesCorrectness();
}
private static void testSpeed() throws Exception {
System.out.println("capacity:" + QUEUE_CAPACITY + " reps:" + REPETITIONS);
final SimpleQueue queue = new SimpleQueue (QUEUE_CAPACITY, 1 << 14);
final long[] results = new long[20];
for (int i = 0; i < 20; i++) {
System.gc();
results[i] = performanceRun(i, queue);
}
// only average last 10 results for summary
long sum = 0;
for (int i = 10; i < 20; i++) {
sum += results[i];
}
System.out.format("summary,QueuePerfTest,%s,%d\n", queue.getClass().getSimpleName(), sum / 10);
}
private static long performanceRun(int runNumber, SimpleQueue queue) throws Exception {
// for (int i=0;i<CONCURRENCY_LEVEL;i++) {
Producer p = new Producer(queue);
Thread thread = new Thread(p);
thread.start(); // producer will timestamp start
// }
SimpleQueue consumer = queue;
int i = REPETITIONS;
Object result;
do {
result = consumer.take();
} while (0 != --i);
long end = System.nanoTime();
thread.join();
long duration = end - p.start;
long ops = REPETITIONS * 1000L * 1000L * 1000L / duration;
String qName = queue.getClass().getSimpleName();
System.out.format("%d - ops/sec=%,d - %s result=%d\n", runNumber, ops, qName, result);
return ops;
}
public static class Producer implements Runnable {
private final SimpleQueue queue;
long start;
public Producer(SimpleQueue queue) {
this.queue = queue;
}
@Override
public void run() {
SimpleQueue producer = this.queue;
int i = REPETITIONS;
long s = System.nanoTime();
do {
producer.put(i);
} while (0 != --i);
this.start = s;
}
}
private static void tesCorrectness() {
final MultiMBassador bus = new MultiMBassador(CONCURRENCY_LEVEL);
bus.addErrorHandler(TestFailingHandler);

View File

@ -0,0 +1,97 @@
/*
* Copyright 2012 Real Logic Ltd.
*
* 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 dorkbox.util.messagebus;
import org.openjdk.jol.info.ClassLayout;
import org.openjdk.jol.util.VMSupport;
import dorkbox.util.messagebus.common.simpleq.Node;
import dorkbox.util.messagebus.common.simpleq.SimpleQueue;
public class SimpleQueueAltPerfTest {
// 15 == 32 * 1024
public static final int REPETITIONS = Integer.getInteger("reps", 50) * 1000 * 10;
public static final Integer TEST_VALUE = Integer.valueOf(777);
public static final int QUEUE_CAPACITY = 1 << Integer.getInteger("pow2.capacity", 17);
public static void main(final String[] args) throws Exception {
System.out.println(VMSupport.vmDetails());
System.out.println(ClassLayout.parseClass(Node.class).toPrintable());
System.out.println("capacity:" + QUEUE_CAPACITY + " reps:" + REPETITIONS);
final SimpleQueue queue = new SimpleQueue(1<<17);
final long[] results = new long[20];
for (int i = 0; i < 20; i++) {
System.gc();
results[i] = performanceRun(i, queue);
}
// only average last 10 results for summary
long sum = 0;
for (int i = 10; i < 20; i++) {
sum += results[i];
}
System.out.format("summary,QueuePerfTest,%s,%d\n", queue.getClass().getSimpleName(), sum / 10);
}
private static long performanceRun(int runNumber, SimpleQueue queue) throws Exception {
Producer p = new Producer(queue);
Thread thread = new Thread(p);
thread.start(); // producer will timestamp start
SimpleQueue consumer = queue;
Object result;
int i = REPETITIONS;
do {
result = consumer.take();
} while (0 != --i);
long end = System.nanoTime();
thread.join();
long duration = end - p.start;
long ops = REPETITIONS * 1000L * 1000L * 1000L / duration;
String qName = queue.getClass().getSimpleName();
System.out.format("%d - ops/sec=%,d - %s result=%d\n", runNumber, ops, qName, result);
return ops;
}
public static class Producer implements Runnable {
private final SimpleQueue queue;
int queueFull = 0;
long start;
public Producer(SimpleQueue queue) {
this.queue = queue;
}
@Override
public void run() {
SimpleQueue producer = this.queue;
int i = REPETITIONS;
long s = System.nanoTime();
try {
do {
producer.put(TEST_VALUE);
} while (0 != --i);
} catch (InterruptedException e) {
e.printStackTrace();
}
this.start = s;
}
}
}