QFJ-943: optional watermarks-based back pressure propagation from inbo… #168

mrbald · 2018-01-28T21:53:27Z

The background for this change is described in the QFJ-943

To make sure writes to the inbound FIX Message queue don't block the MINA IO loop a blocking queue can be replaced with watermarks-based (configured with lower and upper watermark values instead of fixed queue capacity.

When using watermarks the queue is no longer limited in size.
Instead, when queue grows above the upper watermark, the socket reads of the respective session are suspended by unsubscribing from reads with MINA selector (IoSession.suspendRead()), nicely propagating back pressure onto the remote end of the TCP session by the TCP flow control.

When the queue drains below the lower watermark the socket is resubscribed for reads (IoSession.resumeRead()and the back pressure is relieved.

The functionality is implemented with a standalone class WatermarkTracker.

Watermarks - enabled queue in SPSC scenario shows somewhat near 300ns enqueue/dequeue times, which should not be a problem considering general latencies of QuickFIXj, so I left it with JDK collections.

The implementation is sandboxed / benchmarked in https://github.com/mrbald/ufwj. Additionally, the threaded initiator tested under load in a prod-like environment

NB: watermarks do not save the day if one side of the fix session outperforms the other side on the average. If socket reads get disabled for long enough the receiver starts missing heartbeats and eventually drops the link.

…und queue to the socket

chrjohn · 2018-01-30T09:32:28Z

Hi @mrbald , thanks for the PR and the benchmarks you did.
Will take a more thorough look at it soon.
Cheers
Chris.

chrjohn · 2018-01-30T09:33:28Z

quickfixj-core/src/main/java/quickfix/mina/WatermarkTracker.java

+            long lowerWatermark, long upperWatermark,
+            Function<E, S> classifier,
+            Consumer<S> onLowerWatermarkCrossed, Consumer<S> onUpperWatermarkCrossed) {
+        assert lowerWatermark >= 0 && lowerWatermark < upperWatermark;


I am not a fan of assertions in non-test code. Assertions are disabled by default anyway. Could you please remove them? Thanks

It's a documenting assertion (as you said, they are disabled in prod). Anyway, I'll remove in the next commit.

chrjohn · 2018-01-30T09:33:52Z

quickfixj-core/src/main/java/quickfix/mina/WatermarkTracker.java

+            BlockingQueue<E> queue,
+            long lowerWatermark, long upperWatermark,
+            Runnable onLowerWatermarkCrossed, Runnable onUpperWatermarkCrossed) {
+        assert lowerWatermark >= 0 && lowerWatermark < upperWatermark;


I am not a fan of assertions in non-test code. Assertions are disabled by default anyway. Could you please remove them? Thanks

It's a documenting assertion (as you said, they are disabled in prod). Anyway, I'll remove in the next commit.

chrjohn · 2018-01-30T09:34:45Z

quickfixj-core/src/main/java/quickfix/mina/WatermarkTracker.java

+ ******************************************************************************/
+
+/*
+ * Copyright (c) 2018 Vladimir Lysyy (mrbald@github)


I am not sure if we can mix different licenses and copyrights here. Will have to check.

Happy to "donate" it to any license, and strip this comment. Will include in the next commit.

chrjohn · 2018-01-30T09:37:42Z

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

+                    }
+                });
+    }
+    private static IoSession lookupIoSession(Session qfSession) {


There is a similar method in the other EventHandlingStrategy. Can this be moved to the super-interface maybe?

should be possible. static or default method on the EventHandlingStrategy interface sounds ok?

chrjohn · 2018-01-30T09:38:36Z

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

+                    final IoSession ioSession = lookupIoSession(qfSession);
+                    if (ioSession != null && ioSession.isReadSuspended()) {
+                        ioSession.resumeRead();
+                        LOG.info("{}: inbound queue size < lower watermark ({}), socket reads resumed",


IMHO this should be DEBUG logging.

They point to a slow consumer problem - something to be picked up by prod monitoring and actioned upon, therefore definitely not debug. Also, if/when they are logged frequently (up/down/up/down) it is a sign that they should be reconfigured. But I understand there may be established rules for what to log when in QuickFIX, so happy to change. Let me know.

Agreed. But could you please use qfSession.getLog().onEvent( logMsg )? That way you don't have to specify the sessionID and the message gets written to the correct event log file. Please note that you cannot use "{ }" in the log String then.

chrjohn · 2018-01-30T09:38:44Z

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

+                    final IoSession ioSession = lookupIoSession(qfSession);
+                    if (ioSession != null && !ioSession.isReadSuspended()) {
+                        ioSession.suspendRead();
+                        LOG.info("{}: inbound queue size > upper watermark ({}), socket reads suspended",


IMHO this should be DEBUG logging.

They point to a slow consumer problem - something to be picked up by prod monitoring and actioned upon, therefore definitely not debug. Also, if/when they are logged frequently (up/down/up/down) it is a sign that they should be reconfigured. But I understand there may be established rules for what to log when in QuickFIX, so happy to change. Let me know.

Agreed. But could you please use qfSession.getLog().onEvent( logMsg )? That way you don't have to specify the sessionID and the message gets written to the correct event log file. Please note that you cannot use "{ }" in the log String then.

chrjohn · 2018-01-30T09:49:47Z

quickfixj-core/pom.xml

+                        </goals>
+                        <configuration>
+                            <sources>
+                                <source>${project.build.directory}/generated-sources</source>


Does this add all the generated source files to the JAR file? Is this needed? They are generated on a fresh build anyway.

this helps IntelliJ automatically detect the generated files (this was the reason), if they also get added to the *-sources.jar I'd love it too as generated or not, sometimes it's nice to click down to sources to see what they do.

Having said the above - if there are reservations about this change, let's drop it. Let me know then - I'll update the pull request.

mrbald · 2018-01-30T23:21:46Z

@chrjohn - pushed the requested changes. My production code is on 1.6.x - I'd like it in a 1.6.x patch, if you plan to do more of those. Would that need a pull request into the 1.6 branch?

chrjohn · 2018-02-01T09:48:44Z

Hi @mrbald , there will not be any 1.6.x release anymore. Even this change is probably more a 2.1.x change (than 2.0.x) since it introduces new functionality.

chrjohn · 2018-02-05T14:28:22Z

Hi @mrbald ,
regarding the benchmarks. How do I run them? I tried the usual JMH way with building the JAR and then doing java -jar <benchmarkfile> but it did not work.
Thanks,
Chris.

mrbald · 2018-02-06T11:57:23Z

Hi @chrjohn,

Benchmarks - I just run them from the IDE - open the benchmark source code (say net/bobah/ufwj/queue/WatermarkTrackerBenchmark.java, and "Run as Java Application...". It forks the JVM, so the fact or running from the IDE should not disturb numbers.

Importantly, JMH does code generation during compilation so the IDE should be asked to do the same when building (IntelliJ does if the repo is imported as Maven project).

It should be possible to run the same class from the command line, or with an appropriate maven plugin as well, just that I haven't done it that way in that repo.

Let me know if you have problems running it, I'll get something command-line based to work.

chrjohn · 2018-02-14T13:32:14Z

quickfixj-core/src/main/java/quickfix/mina/ThreadPerSessionEventHandlingStrategy.java

+                            if (ioSession != null && ioSession.isReadSuspended()) {
+                                ioSession.resumeRead();
+                                LOG.info("{}: inbound queue size < lower watermark ({}), socket reads resumed",
+                                        quickfixSession.getSessionID(), queueLowerWatermark);


See comments on logging in SingleThreadedEventHandlingStrategy.

chrjohn · 2018-02-14T13:39:18Z

Hi @mrbald ,
I use Netbeans but I had to change the POM for it to work but looking good now.
Thanks

mrbald · 2018-02-16T19:05:43Z

@chrjohn, logging updated, please review. if you don't mind, can you please share the changes you needed to do to the pom file to run the benchmarks?

chrjohn

Hi, did a review once more. Thanks for checking.

chrjohn · 2018-02-14T13:37:03Z

quickfixj-core/src/test/java/quickfix/mina/WatermarkTrackerTest.java

+ ******************************************************************************/
+
+/*
+ * Copyright (c) 2018 Vladimir Lysyy (mrbald@github)


Could you please remove this copyright/license? You can use "@author" if desired. Thanks

chrjohn · 2018-02-16T22:20:57Z

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

 /**
 * Processes messages for all sessions in a single thread.
 */
 public class SingleThreadedEventHandlingStrategy implements EventHandlingStrategy {
+    private static final Logger LOG = LoggerFactory.getLogger(EventHandlingStrategy.class);


LOG shouldn't be required anymore now.

chrjohn · 2018-02-16T22:21:54Z

quickfixj-core/src/main/java/quickfix/mina/ThreadPerSessionEventHandlingStrategy.java

 /**
 * Processes messages in a session-specific thread.
 */
 public class ThreadPerSessionEventHandlingStrategy implements EventHandlingStrategy {
+    private static final Logger LOG = LoggerFactory.getLogger(EventHandlingStrategy.class);


LOG shouldn't be required anymore now.

chrjohn · 2018-02-16T22:38:36Z

quickfixj-core/src/main/java/quickfix/mina/SingleThreadedEventHandlingStrategy.java

+        queueTracker = WatermarkTracker.newMulti(eventQueue, queueLowerWatermark, queueUpperWatermark,
+                evt -> evt.quickfixSession,
+                qfSession -> { // lower watermark crossed down, while reads suspended
+                    final IoSession ioSession = lookupIoSession(qfSession);


The onWatermarkCrossed actions are duplicated in SingleThreadedEventHandlingStrategy and ThreadPerSessionEventHandlingStrategy. Can this be unified somehow?

chrjohn · 2018-02-16T23:01:35Z

Hi @mrbald ,
I had to add the following to the benchmark POM:

<plugin>
               <groupId>org.codehaus.mojo</groupId>
               <artifactId>exec-maven-plugin</artifactId>
               <executions>
                   <execution>
                       <id>run-benchmarks</id>
                       <phase>integration-test</phase>
                       <goals>
                           <goal>exec</goal>
                       </goals>
                       <configuration>
                           <classpathScope>test</classpathScope>
                           <executable>java</executable>
                           <arguments>
                               <argument>-classpath</argument>
                               <classpath />
                               <argument>org.openjdk.jmh.Main</argument>
                               <argument>.*</argument>
                           </arguments>
                       </configuration>
                   </execution>
               </executions>
           </plugin>

This ran the benchmarks on build. I could then also run the benchmarks alone because the classes and information in META-INF were generated. I had the following error without these changes:
Unable to find the resource: /META-INF/BenchmarkList

Another question: which one of the benchmarks is the one where the "threaded initiator tested under load in a prod-like environment" (as mentioned in description of this PR)?

Thanks in advance!

…eftover mrbald copyright

chrjohn · 2018-03-07T12:03:02Z

Thanks @mrbald

QFJ-943: optional watermarks-based back pressure propagaion from inbo…

2ea92ac

…und queue to the socket

chrjohn reviewed Jan 30, 2018

View reviewed changes

chrjohn requested changes Jan 30, 2018

View reviewed changes

chrjohn added this to the QFJ 2.1.0 milestone Jan 30, 2018

QFJ-943: code review feedback

b6a7131

chrjohn changed the title ~~QFJ-943: optional watermarks-based back pressure propagaion from inbo…~~ QFJ-943: optional watermarks-based back pressure propagation from inbo… Feb 1, 2018

chrjohn reviewed Feb 14, 2018

View reviewed changes

QFJ-943: code review feedback - watermarks logging normalized

9cccdb3

chrjohn requested changes Feb 16, 2018

View reviewed changes

QFJ-943: code review feedback - cleaned up last bits of cut/paste + l…

a65915e

…eftover mrbald copyright

chrjohn approved these changes Mar 6, 2018

View reviewed changes

chrjohn merged commit 746a35b into quickfix-j:master Mar 6, 2018

mrbald deleted the jira/QFJ-943 branch November 9, 2018 22:11

QFJ-943: optional watermarks-based back pressure propagation from inbo… #168

QFJ-943: optional watermarks-based back pressure propagation from inbo… #168

Uh oh!

Conversation

mrbald commented Jan 28, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chrjohn commented Jan 30, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald commented Jan 30, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chrjohn commented Feb 1, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chrjohn commented Feb 5, 2018

Uh oh!

mrbald commented Feb 6, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chrjohn Feb 14, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chrjohn commented Feb 14, 2018

Uh oh!

mrbald commented Feb 16, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

chrjohn left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mrbald commented Jan 28, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald Jan 30, 2018 •

edited

Loading

mrbald commented Jan 30, 2018 •

edited

Loading

chrjohn commented Feb 1, 2018 •

edited

Loading

mrbald commented Feb 6, 2018 •

edited

Loading

chrjohn Feb 14, 2018 •

edited

Loading

mrbald commented Feb 16, 2018 •

edited

Loading