{"action":"created","issue":{"url":"https://api.github.com/repos/reactor/reactor/issues/496","repository_url":"https://api.github.com/repos/reactor/reactor","labels_url":"https://api.github.com/repos/reactor/reactor/issues/496/labels{/name}","comments_url":"https://api.github.com/repos/reactor/reactor/issues/496/comments","events_url":"https://api.github.com/repos/reactor/reactor/issues/496/events","html_url":"https://github.com/reactor/reactor/issues/496","id":90626884,"number":496,"title":"RingBufferWorkProcessor stuck","user":{"login":"christian-laakmann","id":379279,"avatar_url":"https://avatars.githubusercontent.com/u/379279?v=3","gravatar_id":"","url":"https://api.github.com/users/christian-laakmann","html_url":"https://github.com/christian-laakmann","followers_url":"https://api.github.com/users/christian-laakmann/followers","following_url":"https://api.github.com/users/christian-laakmann/following{/other_user}","gists_url":"https://api.github.com/users/christian-laakmann/gists{/gist_id}","starred_url":"https://api.github.com/users/christian-laakmann/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/christian-laakmann/subscriptions","organizations_url":"https://api.github.com/users/christian-laakmann/orgs","repos_url":"https://api.github.com/users/christian-laakmann/repos","events_url":"https://api.github.com/users/christian-laakmann/events{/privacy}","received_events_url":"https://api.github.com/users/christian-laakmann/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":5,"created_at":"2015-06-24T09:45:26Z","updated_at":"2016-06-29T05:22:40Z","closed_at":"2015-09-07T16:14:30Z","body":"Hi all,\r\n\r\ndue to #479 we moved most of our implementation over to RingBufferWorkProcesser, which perfect for most of our use cases.\r\n\r\nHowever, we have a problem doing this:\r\n\r\n```\r\nRingBufferWorkProcessor<Integer> processor = RingBufferWorkProcessor.create(\"test\", 8192);\r\n\r\n      Stream<Integer> stream = Streams.wrap(processor);\r\n\r\n      CountDownLatch latch = new CountDownLatch(4);\r\n\r\n      Consumer<Void> completeConsumer = c -> {\r\n        latch.countDown();\r\n        System.out.println(Thread.currentThread().getName() + \" completed\");\r\n      };\r\n\r\n      stream.buffer(1000).consume(e -> {}, t -> {}, completeConsumer);\r\n      stream.buffer(1000).consume(e -> {}, t -> {}, completeConsumer);\r\n      stream.buffer(1000).consume(e -> {}, t -> {}, completeConsumer);\r\n      stream.buffer(1000).consume(e -> {}, t -> {}, completeConsumer);\r\n\r\n      processor.onNext(1);\r\n      processor.onComplete();\r\n\r\n      if (!latch.await(30, TimeUnit.SECONDS)) {\r\n        System.err.println(\"not completed\");\r\n        throw new RuntimeException(\"not completed\");\r\n      }\r\n```\r\n\r\nWe create 4 worker threads but have fewer values to be send over the processor. From time to time, not all worker threads will complete correctly (i.e. latch.await(...) fails).\r\n\r\nHere's a thread dump taken at a breakpoint in the if-clause:\r\n\r\n```\r\n\"main@1\" prio=5 tid=0x1 nid=NA runnable\r\n  java.lang.Thread.State: RUNNABLE\r\n\t  at com.novomind.iport.process.StreamTest.processorTest(StreamTest.java:53)\r\n\t  at sun.reflect.NativeMethodAccessorImpl.invoke0(NativeMethodAccessorImpl.java:-1)\r\n\t  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\t  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\t  at java.lang.reflect.Method.invoke(Method.java:483)\r\n\t  at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)\r\n\t  at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)\r\n\t  at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)\r\n\t  at mockit.integration.junit4.internal.JUnit4TestRunnerDecorator.executeTestMethod(JUnit4TestRunnerDecorator.java:131)\r\n\t  at mockit.integration.junit4.internal.JUnit4TestRunnerDecorator.invokeExplosively(JUnit4TestRunnerDecorator.java:71)\r\n\t  at mockit.integration.junit4.internal.MockFrameworkMethod.invokeExplosively(MockFrameworkMethod.java:42)\r\n\t  at sun.reflect.NativeMethodAccessorImpl.invoke0(NativeMethodAccessorImpl.java:-1)\r\n\t  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\t  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\t  at java.lang.reflect.Method.invoke(Method.java:483)\r\n\t  at mockit.internal.util.MethodReflection.invokeWithCheckedThrows(MethodReflection.java:112)\r\n\t  at mockit.internal.mockups.MockMethodBridge.callMock(MockMethodBridge.java:85)\r\n\t  at mockit.internal.mockups.MockMethodBridge.invoke(MockMethodBridge.java:44)\r\n\t  at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)\r\n\t  at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)\r\n\t  at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)\r\n\t  at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)\r\n\t  at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)\r\n\t  at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)\r\n\t  at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)\r\n\t  at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)\r\n\t  at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)\r\n\t  at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)\r\n\t  at org.junit.runners.ParentRunner.run(ParentRunner.java:363)\r\n\t  at org.junit.runner.JUnitCore.run(JUnitCore.java:137)\r\n\t  at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:78)\r\n\t  at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:212)\r\n\t  at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:68)\r\n\t  at sun.reflect.NativeMethodAccessorImpl.invoke0(NativeMethodAccessorImpl.java:-1)\r\n\t  at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)\r\n\t  at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)\r\n\t  at java.lang.reflect.Method.invoke(Method.java:483)\r\n\t  at com.intellij.rt.execution.CommandLineWrapper.main(CommandLineWrapper.java:130)\r\n\r\n\"test-18@1259\" prio=5 tid=0x20 nid=NA waiting\r\n  java.lang.Thread.State: WAITING\r\n\t  at sun.misc.Unsafe.park(Unsafe.java:-1)\r\n\t  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\r\n\t  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\r\n\t  at reactor.jarjar.com.lmax.disruptor.LiteBlockingWaitStrategy.waitFor(LiteBlockingWaitStrategy.java:56)\r\n\t  at reactor.jarjar.com.lmax.disruptor.ProcessingSequenceBarrier.waitFor(ProcessingSequenceBarrier.java:55)\r\n\t  at reactor.core.processor.RingBufferWorkProcessor$WorkSignalProcessor.replay(RingBufferWorkProcessor.java:802)\r\n\t  at reactor.core.processor.RingBufferWorkProcessor$WorkSignalProcessor.run(RingBufferWorkProcessor.java:726)\r\n\t  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n\t  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n\t  at java.lang.Thread.run(Thread.java:745)\r\n\r\n\"test-20@1260\" prio=5 tid=0x22 nid=NA waiting\r\n  java.lang.Thread.State: WAITING\r\n\t  at sun.misc.Unsafe.park(Unsafe.java:-1)\r\n\t  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\r\n\t  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\r\n\t  at reactor.jarjar.com.lmax.disruptor.LiteBlockingWaitStrategy.waitFor(LiteBlockingWaitStrategy.java:56)\r\n\t  at reactor.jarjar.com.lmax.disruptor.ProcessingSequenceBarrier.waitFor(ProcessingSequenceBarrier.java:55)\r\n\t  at reactor.core.processor.RingBufferWorkProcessor$WorkSignalProcessor.run(RingBufferWorkProcessor.java:760)\r\n\t  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n\t  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n\t  at java.lang.Thread.run(Thread.java:745)\r\n\r\n\"test-17@1257\" prio=5 tid=0x1f nid=NA waiting\r\n  java.lang.Thread.State: WAITING\r\n\t  at sun.misc.Unsafe.park(Unsafe.java:-1)\r\n\t  at java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)\r\n\t  at java.util.concurrent.locks.AbstractQueuedSynchronizer$ConditionObject.await(AbstractQueuedSynchronizer.java:2039)\r\n\t  at reactor.jarjar.com.lmax.disruptor.LiteBlockingWaitStrategy.waitFor(LiteBlockingWaitStrategy.java:56)\r\n\t  at reactor.jarjar.com.lmax.disruptor.ProcessingSequenceBarrier.waitFor(ProcessingSequenceBarrier.java:55)\r\n\t  at reactor.core.processor.RingBufferWorkProcessor$WorkSignalProcessor.run(RingBufferWorkProcessor.java:760)\r\n\t  at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)\r\n\t  at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)\r\n\t  at java.lang.Thread.run(Thread.java:745)\r\n\r\n\"Finalizer@1406\" daemon prio=8 tid=0x3 nid=NA waiting\r\n  java.lang.Thread.State: WAITING\r\n\t  at java.lang.Object.wait(Object.java:-1)\r\n\t  at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:142)\r\n\t  at java.lang.ref.ReferenceQueue.remove(ReferenceQueue.java:158)\r\n\t  at java.lang.ref.Finalizer$FinalizerThread.run(Finalizer.java:209)\r\n\r\n\"Reference Handler@1407\" daemon prio=10 tid=0x2 nid=NA waiting\r\n  java.lang.Thread.State: WAITING\r\n\t  at java.lang.Object.wait(Object.java:-1)\r\n\t  at java.lang.Object.wait(Object.java:502)\r\n\t  at java.lang.ref.Reference$ReferenceHandler.run(Reference.java:157)\r\n\r\n\"Attach Listener@745\" daemon prio=5 tid=0x5 nid=NA runnable\r\n  java.lang.Thread.State: RUNNABLE\r\n\r\n\"Signal Dispatcher@1405\" daemon prio=9 tid=0x4 nid=NA runnable\r\n  java.lang.Thread.State: RUNNABLE\r\n```\r\n\r\nAny ideas?"},"comment":{"url":"https://api.github.com/repos/reactor/reactor/issues/comments/229259310","html_url":"https://github.com/reactor/reactor/issues/496#issuecomment-229259310","issue_url":"https://api.github.com/repos/reactor/reactor/issues/496","id":229259310,"user":{"login":"udayabaski","id":1947490,"avatar_url":"https://avatars.githubusercontent.com/u/1947490?v=3","gravatar_id":"","url":"https://api.github.com/users/udayabaski","html_url":"https://github.com/udayabaski","followers_url":"https://api.github.com/users/udayabaski/followers","following_url":"https://api.github.com/users/udayabaski/following{/other_user}","gists_url":"https://api.github.com/users/udayabaski/gists{/gist_id}","starred_url":"https://api.github.com/users/udayabaski/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/udayabaski/subscriptions","organizations_url":"https://api.github.com/users/udayabaski/orgs","repos_url":"https://api.github.com/users/udayabaski/repos","events_url":"https://api.github.com/users/udayabaski/events{/privacy}","received_events_url":"https://api.github.com/users/udayabaski/received_events","type":"User","site_admin":false},"created_at":"2016-06-29T05:22:40Z","updated_at":"2016-06-29T05:22:40Z","body":"Hi,\r\n   I have the same problem. \r\nprivate static RingBufferWorkProcessor<TestEvent> processor = RingBufferWorkProcessor.<TestEvent> create(\"test\", 128);\r\nprocessor.subscribe(testSubscriber);\r\n\r\n@Component\r\npublic class TestSubscriber implements Subscriber<testEvent> {\r\n    private static Log log = new Log(TestSubscriber.class);\r\n\r\n    @Override\r\n    public void onSubscribe(final Subscription s) {\r\n        if (s != null) {\r\n            s.request(Long.MAX_VALUE);\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void onNext(final TestEvent testEvent) {\r\n        if (testEvent == null) {\r\n            return;\r\n        }\r\n        if ((testEvent.getBroadcastType() != null) && testEvent.getBroadcastType().isTrue()) {\r\n            ReactorResources.publish(testEvent.getType().name(), testEvent);\r\n        } else {\r\n            ReactorResources.publish(testEvent.getName(), testEvent);\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void onError(final Throwable t) {\r\n        if (t != null) {\r\n            log.error(t.getMessage());\r\n        }\r\n    }\r\n\r\n    @Override\r\n    public void onComplete() {\r\n        log.info(\"TestSubscriber is done.\");\r\n    }\r\n}\r\n\r\nPosting event,\r\n\r\nwhile(it.hasNext()) {\r\n  processor.onNext(it.next());\r\n}\r\n\r\nIt stucks after processing few hundred messages.\r\n\r\n\"workQueueDispatcher-3\" daemon prio=5 tid=42 TIMED_WAITING\r\n\tat sun.misc.Unsafe.park(Native Method)\r\n\tat java.util.concurrent.locks.LockSupport.parkNanos(LockSupport.java:338)\r\n\tat reactor.jarjar.com.lmax.disruptor.MultiProducerSequencer.next(MultiProducerSequencer.java:136)\r\n\tat reactor.jarjar.com.lmax.disruptor.MultiProducerSequencer.next(MultiProducerSequencer.java:105)\r\n\tat reactor.jarjar.com.lmax.disruptor.RingBuffer.next(RingBuffer.java:246)\r\n\tat reactor.core.dispatch.WorkQueueDispatcher.allocateTask(WorkQueueDispatcher.java:172)\r\n\tat reactor.core.dispatch.MultiThreadDispatcher.allocateRecursiveTask(MultiThreadDispatcher.java:59)\r\n\tat reactor.core.dispatch.AbstractLifecycleDispatcher.dispatch(AbstractLifecycleDispatcher.java:116)\r\n\tat reactor.bus.EventBus.notify(EventBus.java:368)\r\n\t   Local Variable: reactor.bus.Event#6"}}
{"actor":{"display_login":"udayabaski"}}