Calculating orderedFieldsArray eagerly to avoid concurrency problems (QFJ-971) (#241)

* #240 Calculating orderedFieldsArray eagerly to avoid races when dictionary is shared between multiple sessions
* Added unit test to verify race condition is fixed.
* Added missing constructor to PausableThreadPoolExecutor.
Co-authored-by: Christoph John <[email protected]>

Author: AndreyNudko

quickfixj-core/src/main/java/quickfix/DataDictionary.java

@@ -87,6 +87,7 @@ public class DataDictionary {
     private final IntegerStringMap<String> valueNames = new IntegerStringMap<>();
     private final StringIntegerMap<GroupInfo> groups = new StringIntegerMap<>();
     private final Map<String, Node> components = new HashMap<>();
+    private int[] orderedFieldsArray;
 
     private DataDictionary() {
     }
@@ -529,6 +530,8 @@ private void copyFrom(DataDictionary rhs) {
         setCheckUserDefinedFields(rhs.checkUserDefinedFields);
         setCheckUnorderedGroupFields(rhs.checkUnorderedGroupFields);
         setAllowUnknownMessageFields(rhs.allowUnknownMessageFields);
+
+        calculateOrderedFields();
     }
 
     @SuppressWarnings("unchecked")
@@ -1017,6 +1020,8 @@ private void load(InputStream inputStream) throws ConfigError {
                 load(document, msgtype, messageNode);
             }
         }
+
+        calculateOrderedFields();
     }
 
     public int getNumMessageCategories() {
@@ -1069,18 +1074,22 @@ private void load(Document document, String msgtype, Node node) throws ConfigErr
         }
     }
 
-    private int[] orderedFieldsArray;
-
     public int[] getOrderedFields() {
-        if (orderedFieldsArray == null) {
-            orderedFieldsArray = new int[fields.size()];
-            int i = 0;
-            for (Integer field : fields) {
-                orderedFieldsArray[i++] = field;
-            }
+        return orderedFieldsArray;
+    }
+
+    private void calculateOrderedFields() {
+        orderedFieldsArray = new int[fields.size()];
+        int i = 0;
+        for (Integer field : fields) {
+            orderedFieldsArray[i++] = field;
         }
 
-        return orderedFieldsArray;
+        for (Map<Integer, GroupInfo> gm : groups.values()) {
+            for (GroupInfo gi : gm.values()) {
+                gi.dataDictionary.calculateOrderedFields();
+            }
+        }
     }
 
     private int lookupXMLFieldNumber(Document document, Node node) throws ConfigError {

quickfixj-core/src/test/java/quickfix/DataDictionaryTest.java

@@ -34,9 +34,15 @@
 import quickfix.field.MsgType;
 import quickfix.field.NoHops;
 import quickfix.field.NoPartyIDs;
+import quickfix.field.NoPartySubIDs;
 import quickfix.field.NoRelatedSym;
 import quickfix.field.OrdType;
 import quickfix.field.OrderQty;
+import quickfix.field.PartyID;
+import quickfix.field.PartyIDSource;
+import quickfix.field.PartyRole;
+import quickfix.field.PartySubID;
+import quickfix.field.PartySubIDType;
 import quickfix.field.Price;
 import quickfix.field.QuoteReqID;
 import quickfix.field.SenderCompID;
@@ -55,9 +61,15 @@
 import java.math.BigDecimal;
 import java.net.URL;
 import java.net.URLClassLoader;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.Matchers.hasProperty;
+import static org.junit.Assert.assertArrayEquals;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertNotNull;
@@ -973,6 +985,8 @@ public void testCopy() throws Exception {
         assertEquals(ddCopy.isCheckFieldsOutOfOrder(),dataDictionary.isCheckFieldsOutOfOrder());
         assertEquals(ddCopy.isCheckUnorderedGroupFields(),dataDictionary.isCheckUnorderedGroupFields());
         assertEquals(ddCopy.isCheckUserDefinedFields(),dataDictionary.isCheckUserDefinedFields());
+        assertArrayEquals(getDictionary().getOrderedFields(),ddCopy.getOrderedFields());
+        assertArrayEquals(getDictionary().getOrderedFields(),dataDictionary.getOrderedFields());
 
         DataDictionary.GroupInfo groupFromDDCopy = ddCopy.getGroup(NewOrderSingle.MSGTYPE, NoPartyIDs.FIELD);
         assertTrue(groupFromDDCopy.getDataDictionary().isAllowUnknownMessageFields());
@@ -984,6 +998,31 @@ public void testCopy() throws Exception {
         assertTrue(ddCopy.isAllowUnknownMessageFields());
         groupFromDDCopy = ddCopy.getGroup(NewOrderSingle.MSGTYPE, NoPartyIDs.FIELD);
         assertTrue(groupFromDDCopy.getDataDictionary().isAllowUnknownMessageFields());
+
+        DataDictionary originalGroupDictionary = getDictionary().getGroup(NewOrderSingle.MSGTYPE, NoPartyIDs.FIELD).getDataDictionary();
+        DataDictionary groupDictionary = dataDictionary.getGroup(NewOrderSingle.MSGTYPE, NoPartyIDs.FIELD).getDataDictionary();
+        DataDictionary copyGroupDictionary = ddCopy.getGroup(NewOrderSingle.MSGTYPE, NoPartyIDs.FIELD).getDataDictionary();
+        assertArrayEquals(originalGroupDictionary.getOrderedFields(), groupDictionary.getOrderedFields());
+        assertArrayEquals(originalGroupDictionary.getOrderedFields(), copyGroupDictionary.getOrderedFields());
+
+        DataDictionary originalNestedGroupDictionary = originalGroupDictionary.getGroup(NewOrderSingle.MSGTYPE, NoPartySubIDs.FIELD).getDataDictionary();
+        DataDictionary nestedGroupDictionary = groupDictionary.getGroup(NewOrderSingle.MSGTYPE, NoPartySubIDs.FIELD).getDataDictionary();
+        DataDictionary copyNestedGroupDictionary = copyGroupDictionary.getGroup(NewOrderSingle.MSGTYPE, NoPartySubIDs.FIELD).getDataDictionary();
+        assertArrayEquals(originalNestedGroupDictionary.getOrderedFields(), nestedGroupDictionary.getOrderedFields());
+        assertArrayEquals(originalNestedGroupDictionary.getOrderedFields(), copyNestedGroupDictionary.getOrderedFields());
+    }
+
+    @Test
+    public void testOrderedFields() throws Exception {
+        final DataDictionary dataDictionary = getDictionary();
+
+        final DataDictionary partyIDsDictionary = dataDictionary.getGroup(NewOrderSingle.MSGTYPE, NoPartyIDs.FIELD).getDataDictionary();
+        int[] expectedPartyIDsFieldOrder = new int[] {PartyID.FIELD, PartyIDSource.FIELD, PartyRole.FIELD, NoPartySubIDs.FIELD};
+        assertArrayEquals(expectedPartyIDsFieldOrder, partyIDsDictionary.getOrderedFields());
+
+        final DataDictionary partySubIDsDictionary = partyIDsDictionary.getGroup(NewOrderSingle.MSGTYPE, NoPartySubIDs.FIELD).getDataDictionary();
+        int[] expectedPartySubIDsFieldOrder = new int[] {PartySubID.FIELD, PartySubIDType.FIELD};
+        assertArrayEquals(expectedPartySubIDsFieldOrder, partySubIDsDictionary.getOrderedFields());
     }
 
     /**
@@ -1299,6 +1338,52 @@ public void testAllowingBlankValuesDisablesFieldValidation() throws Exception {
         dictionary.validate(newSingle, true);
     }
 
+
+    // QFJ-971
+    @Test
+    public void testConcurrentValidationFailure() throws Exception {
+        final String data = "8=FIX.4.4|9=284|35=F|49=TEST_49|56=TEST_56|34=420|52=20190302-07:31:57.079|"
+                + "115=TEST3|116=TEST_116|11=TEST_11|41=TEST_41|55=TEST_55|48=TEST_48|22=4|54=2|"
+                + "60=20190302-07:31:56.933|38=100|207=TEST_207|453=1|448=TEST_448|447=D|452=3|10=204|";
+        final String msgString = data.replace('|', (char) 1);
+
+        // use some more threads to make it more likely that the problem will occur
+        final int noOfThreads = 8;
+        final int noOfIterations = 500;
+
+        for (int i = 0; i < noOfIterations; i++) {
+            final DataDictionary dd = new DataDictionary("FIX44.xml");
+            final MessageFactory messageFactory = new quickfix.fix44.MessageFactory();
+            PausableThreadPoolExecutor ptpe = new PausableThreadPoolExecutor(noOfThreads);
+            // submit threads to pausable executor and try to let them start at the same time
+            ptpe.pause();
+            List<Future> resultList = new ArrayList<>();
+            for (int j = 0; j < noOfThreads; j++) {
+                final Callable messageParser = (Callable) () -> {
+                    Message msg = MessageUtils.parse(messageFactory, dd, msgString);
+                    Group partyGroup = msg.getGroups(quickfix.field.NoPartyIDs.FIELD).get(0);
+                    char partyIdSource = partyGroup.getChar(PartyIDSource.FIELD);
+                    assertEquals(PartyIDSource.PROPRIETARY_CUSTOM_CODE, partyIdSource);
+                    return msg;
+                };
+                resultList.add(ptpe.submit(messageParser));
+            }
+
+            // start all threads
+            ptpe.resume();
+            ptpe.shutdown();
+            ptpe.awaitTermination(10, TimeUnit.MILLISECONDS);
+
+            // validate results
+            for (Future future : resultList) {
+                // if unsuccessful, this will throw an ExecutionException
+                future.get();
+            }
+        }
+    }
+
+
+
     //
     // Group Validation Tests in RepeatingGroupTest
     //

quickfixj-core/src/test/java/quickfix/PausableThreadPoolExecutor.java

@@ -15,6 +15,11 @@ public PausableThreadPoolExecutor() {
         super(2, 2, 20, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000));
     }
 
+    public PausableThreadPoolExecutor(int noOfThreads) {
+        super(noOfThreads, noOfThreads, 20, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10000));
+    }
+
+    @Override
     protected void beforeExecute(Thread t, Runnable r) {
         super.beforeExecute(t, r);
         pauseLock.lock();