HPCC-24069 Fix changeToWrite transaction bug

Do not hold the SDS transaction mutex (dataRWLock) whilst trying
to establish a write lock in the RTM_DELETE_ON_DISCONNECT logic.

Signed-off-by: Jake Smith <jake.smith@lexisnexisrisk.com>

dali/base/dasds.cpp

@@ -220,7 +220,8 @@ public:
                     break;
             }
             PROGLOG("CLCLockBlock(write=%d) timeout %s(%d), took %d ms",!readLock,fname,lnum,got-msTick());
-            PrintStackReport();
+            if (readWriteStackTracing)
+                PrintStackReport();
         }
         got = msTick();
     };
@@ -238,10 +239,7 @@ public:
                 PrintStackReport();
         }
     }
-    void changeToWrite(unsigned timeout)
-    {
-        lock.changeToWrite(timeout);
-    }
+    void changeToWrite(const char *msg);
 };
 #else
 class LinkingCriticalBlock : public CriticalBlock, public CInterface, implements IInterface
@@ -265,13 +263,33 @@ public:
     {
         lock.unlock();
     }
-    void changeToWrite(unsigned timeout)
-    {
-        lock.changeToWrite(timeout);
-    }
+    void changeToWrite(const char *msg);
 };
 #endif
 
+void CLCLockBlock::changeToWrite(const char *msg)
+{
+    /* NB: this intentionally clears the lock, before re-establishing a write lock
+     * because we do not want to keep this read lock whilst waiting for a write,
+     * as other write transactions would pile up and/or cause deadlock.
+     */
+
+    // Must be read locked when called.
+    dbgassertex(lock.queryReadLockCount() > 0);
+
+    lock.unlock();
+
+    CCycleTimer timer;
+    while (true)
+    {
+        if (lock.lockWrite(readWriteTimeout))
+            break;
+        PROGLOG("changeToWrite timeout [%s], time taken so far %u ms", msg, timer.elapsedMs());
+        if (readWriteStackTracing)
+            PrintStackReport();
+    }
+}
+
 #ifdef USECHECKEDCRITICALSECTIONS
 #define CHECKEDDALIREADLOCKBLOCK(l, timeout)  Owned<CLCLockBlock> glue(block,__LINE__) = new CLCLockBlock(l, true, timeout, __FILE__, __LINE__)
 #define CHECKEDDALIWRITELOCKBLOCK(l, timeout)  Owned<CLCLockBlock> glue(block,__LINE__) = new CLCLockBlock(l, false, timeout, __FILE__, __LINE__)
@@ -7848,43 +7866,66 @@ void CCovenSDSManager::disconnect(ConnectionId id, bool deleteRoot, CLCLockBlock
     else
         DBGLOG("Disconnecting orphaned connection: %s, deleteRoot=%s", connection->queryXPath(), deleteRoot?"true":"false");
     // Still want disconnection to be performed & recorded, if orphaned
-    if (!deleteRoot && unlock(tree->queryServerId(), id, true)) // unlock returns true if last unlock and there was a setDROLR on it
+    bool last = !deleteRoot && unlock(tree->queryServerId(), id, true); // unlock returns true if last unlock and there was a setDROLR on it
+    if (last)
         deleteRoot = true;
     if (deleteRoot)
     {
-        if (lockBlock)
-            lockBlock->changeToWrite(readWriteTimeout);
-        connection->queryParent()->removeTree(tree);
-        writeTransactions++;
-        if (!orphaned)
+        if (lockBlock) // NB: only passed if might need read->write
         {
-            Owned<IPropertyTree> changeTree = createPTree(RESERVED_CHANGE_NODE);
-            IPropertyTree *d = changeTree->setPropTree(DELETE_TAG, createPTree());
-            d->setProp("@name", tree->queryName());
-            d->setPropInt("@pos", index+1);
+            lockBlock->changeToWrite(connection->queryXPath());
 
-            Owned<CBranchChange> branchChange = new CBranchChange(*tree);
-            branchChange->noteChange(PDS_Deleted, PDS_Deleted);
-            CPTStack stack = connection->queryPTreePath();
-            stack.pop();
-            if (connection->queryRootUnvalidated() == SDSManager->queryRoot())
-                stack.pop();
-
-            if (!RTM_MODE(connection->queryMode(), RTM_INTERNAL))
+            /* If it was last and deleteRoot was triggered, check again (now that have exclusive read write lock),
+             * that I am last, i.e. that no-one else has established a lock in the changeToWrite window.
+             * If they have, do not deleteRoot, as others are using it.
+             * 
+             * NB: that will mean the write lock was not actually needed.
+             * The lock will be released by the caller (of disconnect()), when 'lockBlock' goes out of scope.
+             */
+            if (last)
             {
-                connection->notify();
-                SDSManager->startNotification(*changeTree, stack, *branchChange);
+                /* check if lock has been re-established, in window within changeToWrite
+                 * If it has, then I am no longer the last link, do not delete root
+                 */
+                CLock *lock = queryLock(tree->queryServerId());
+                if (lock && lock->lockCount())
+                    deleteRoot = false;
             }
+        }
+        if (deleteRoot)
+        {
+            connection->queryParent()->removeTree(tree);
+            writeTransactions++;
+            if (!orphaned)
+            {
+                Owned<IPropertyTree> changeTree = createPTree(RESERVED_CHANGE_NODE);
+                IPropertyTree *d = changeTree->setPropTree(DELETE_TAG, createPTree());
+                d->setProp("@name", tree->queryName());
+                d->setPropInt("@pos", index+1);
 
-            StringBuffer head;
-            const char *tail = splitXPath(path.str(), head);
-            CHECKEDCRITICALBLOCK(blockedSaveCrit, fakeCritTimeout);
-            if (NotFound != index)
-                saveDelta(head.str(), *changeTree);
-            else
-            { // NB: don't believe this can happen, but last thing want to do is save duff delete delta.
-                IERRLOG("** CCovenSDSManager::disconnect - index position lost **");
-                PrintStackReport();
+                Owned<CBranchChange> branchChange = new CBranchChange(*tree);
+                branchChange->noteChange(PDS_Deleted, PDS_Deleted);
+                CPTStack stack = connection->queryPTreePath();
+                stack.pop();
+                if (connection->queryRootUnvalidated() == SDSManager->queryRoot())
+                    stack.pop();
+
+                if (!RTM_MODE(connection->queryMode(), RTM_INTERNAL))
+                {
+                    connection->notify();
+                    SDSManager->startNotification(*changeTree, stack, *branchChange);
+                }
+
+                StringBuffer head;
+                const char *tail = splitXPath(path.str(), head);
+                CHECKEDCRITICALBLOCK(blockedSaveCrit, fakeCritTimeout);
+                if (NotFound != index)
+                    saveDelta(head.str(), *changeTree);
+                else
+                { // NB: don't believe this can happen, but last thing want to do is save duff delete delta.
+                    IERRLOG("** CCovenSDSManager::disconnect - index position lost **");
+                    PrintStackReport();
+                }
             }
         }
     }

system/jlib/jmutex.cpp

@@ -320,191 +320,7 @@ void Monitor::notifyAll()
 
 //==================================================================================
 
-#ifndef USE_PTHREAD_RWLOCK
-
-bool ReadWriteLock::lockRead(bool timed, unsigned timeout)
-{  
-    cs.enter(); 
-    if (writeLocks == 0) 
-    {
-        readLocks++;
-        cs.leave();
-    }
-    else
-    {
-        readWaiting++;
-        cs.leave();
-        if (timed)
-        {
-            if (!readSem.wait(timeout))
-            {
-                cs.enter(); 
-                if (!readSem.wait(0))
-                {
-                    readWaiting--;
-                    cs.leave();
-                    return false;
-                }
-                cs.leave();
-            }
-        }
-        else
-            readSem.wait();
-        //NB: waiting and locks adjusted before the signal occurs.
-    }
-    return true;
-}
-
-bool ReadWriteLock::lockWrite(bool timed, unsigned timeout)
-{
-    cs.enter();
-    if ((readLocks == 0) && (writeLocks == 0))
-    {
-        writeLocks++;
-        cs.leave();
-    }
-    else
-    {
-        writeWaiting++;
-        cs.leave();
-        if (timed)
-        {
-            if (!writeSem.wait(timeout))
-            {
-                cs.enter(); 
-                if (!writeSem.wait(0))
-                {
-                    writeWaiting--;
-                    cs.leave();
-                    return false;
-                }
-                cs.leave();
-            }
-        }
-        else
-            writeSem.wait();
-        //NB: waiting and locks adjusted before the signal occurs.
-    }
-#ifdef _DEBUG
-    exclWriteOwner = GetCurrentThreadId();
-#endif
-    return true;
-}
-
-bool ReadWriteLock::changeToWrite(bool timed, unsigned timeout)
-{
-    cs.enter();
-
-    // invalid use cases, changeToWrite must only be called when read locked.
-    if (writeLocks)
-    {
-        cs.leave();
-        throw makeStringException(9999, "Internal Error: ReadWriteLock::changeToWrite - already write locked");
-    }
-    else if (0 == readLocks)
-    {
-        cs.leave();
-        throw makeStringException(9999, "Internal Error: ReadWriteLock::changeToWrite - not read locked");
-    }
-
-    if (readLocks == 1)
-    {
-        readLocks--;
-        writeLocks++;
-        cs.leave();
-    }
-    else // readLocks>1
-    {
-        readToWriteWaiting++;
-        cs.leave();
-        if (timed)
-        {
-            if (!readToWriteSem.wait(timeout))
-            {
-                cs.enter(); 
-                if (!readToWriteSem.wait(0))
-                {
-                    readToWriteWaiting--;
-                    cs.leave();
-                    return false;
-                }
-                cs.leave();
-            }
-        }
-        else
-            readToWriteSem.wait();
-        //NB: waiting and locks adjusted before the signal occurs.
-    }
-#ifdef _DEBUG
-    exclWriteOwner = GetCurrentThreadId();
-#endif
-    return true;
-}
-
-bool ReadWriteLock::changeToRead()
-{
-    cs.enter();
-    if (!writeLocks)
-    {
-        cs.leave();
-        throw makeStringException(9999, "Internal Error: ReadWriteLock::changeToRead - not write locked");
-
-    }
-    --writeLocks;
-    dbgassertex(!readLocks);
-    readLocks++;
-    cs.leave();
-#ifdef _DEBUG
-    exclWriteOwner = 0;
-#endif
-    return true;
-}
-
-void ReadWriteLock::unlock()
-{
-    cs.enter(); 
-    if (readLocks) // implies this unlock() is paired with a previous read lock
-    {
-        readLocks--;
-        if (readToWriteWaiting && (1 == readLocks))
-        {
-            readToWriteWaiting--;
-            readLocks--;
-            writeLocks++;
-            readToWriteSem.signal();
-        }
-        else if (writeWaiting && (0 == readLocks))
-        {
-            writeWaiting--;
-            writeLocks++;
-            writeSem.signal();
-        }
-    }
-    else // implies this unlock() is paired with a previous write lock
-    {
-        writeLocks--;
-        dbgassertex(writeLocks == 0);
-        if (readWaiting)
-        {
-            unsigned numWaiting = readWaiting;
-            readWaiting = 0;
-            readLocks += numWaiting;
-            readSem.signal(numWaiting);
-        }
-        else if (writeWaiting)
-        {
-            writeWaiting--;
-            writeLocks++;
-            writeSem.signal();
-        }
-#ifdef _DEBUG
-        exclWriteOwner = 0;
-#endif
-    }
-    cs.leave();
-}
-
-#else
+#ifdef USE_PTHREAD_RWLOCK
 
 bool ReadWriteLock::lockRead(unsigned timeout)
 {

system/jlib/jmutex.hpp

@@ -586,14 +586,71 @@ public:
 
 class jlib_decl ReadWriteLock
 {
-    bool lockRead(bool timed, unsigned timeout);
-    bool lockWrite(bool timed, unsigned timeout);
-    bool changeToWrite(bool timed, unsigned timeout);
+    bool lockRead(bool timed, unsigned timeout) { 
+                                cs.enter(); 
+                                if (writeLocks == 0) 
+                                {
+                                    readLocks++;
+                                    cs.leave();
+                                }
+                                else
+                                {
+                                    readWaiting++;
+                                    cs.leave();
+                                    if (timed)
+                                    {
+                                        if (!readSem.wait(timeout)) {
+                                            cs.enter(); 
+                                            if (!readSem.wait(0)) {
+                                                readWaiting--;
+                                                cs.leave();
+                                                return false;
+                                            }
+                                            cs.leave();
+                                        }
+                                    }
+                                    else
+                                        readSem.wait();
+                                    //NB: waiting and locks adjusted before the signal occurs.
+                                }
+                                return true;
+                            }
+    bool lockWrite(bool timed, unsigned timeout) { 
+                                cs.enter(); 
+                                if ((readLocks == 0) && (writeLocks == 0))
+                                {
+                                    writeLocks++;
+                                    cs.leave();
+                                }
+                                else
+                                {
+                                    writeWaiting++;
+                                    cs.leave();
+                                    if (timed)
+                                    {
+                                        if (!writeSem.wait(timeout)) {
+                                            cs.enter(); 
+                                            if (!writeSem.wait(0)) {
+                                                writeWaiting--;
+                                                cs.leave();
+                                                return false;
+                                            }
+                                            cs.leave();
+                                        }
+                                    }
+                                    else
+                                        writeSem.wait();
+                                    //NB: waiting and locks adjusted before the signal occurs.
+                                }
+#ifdef _DEBUG
+                                exclWriteOwner = GetCurrentThreadId();
+#endif
+                                return true;
+                            }
 public:
     ReadWriteLock()
     {
         readLocks = 0; writeLocks = 0; readWaiting = 0; writeWaiting = 0;
-        readToWriteWaiting = 0;
 #ifdef _DEBUG
         exclWriteOwner = 0;
 #endif
@@ -604,16 +661,36 @@ public:
     void lockWrite()        { lockWrite(false, 0); }
     bool lockRead(unsigned timeout) { return lockRead(true, timeout); }
     bool lockWrite(unsigned timeout) { return lockWrite(true, timeout); }
-    void changeToWrite()
-    {
-        changeToWrite(false, 0);
-    }
-    bool changeToWrite(unsigned timeout)
-    {
-        return changeToWrite(true, timeout);
-    }
-    bool changeToRead();
-    void unlock();
+    unsigned queryReadLockCount() const { return readLocks; }
+    void unlock()           { 
+                                cs.enter(); 
+                                if (readLocks) readLocks--;
+                                else
+                                {
+                                    writeLocks--;
+#ifdef _DEBUG
+                                    exclWriteOwner = 0;
+#endif
+                                }
+                                assertex(writeLocks == 0);
+                                if (readLocks == 0)
+                                {
+                                    if (readWaiting)
+                                    {
+                                        unsigned numWaiting = readWaiting;
+                                        readWaiting = 0;
+                                        readLocks += numWaiting;
+                                        readSem.signal(numWaiting);
+                                    }
+                                    else if (writeWaiting)
+                                    {
+                                        writeWaiting--;
+                                        writeLocks++;
+                                        writeSem.signal();
+                                    }
+                                }
+                                cs.leave();
+                            }
     bool queryWriteLocked() { return (writeLocks != 0); }
     void unlockRead()       { unlock(); }
     void unlockWrite()      { unlock(); }
@@ -623,12 +700,10 @@ protected:
     CriticalSection     cs;
     Semaphore           readSem;
     Semaphore           writeSem;
-    Semaphore           readToWriteSem;
     unsigned            readLocks;
     unsigned            writeLocks;
     unsigned            readWaiting;
     unsigned            writeWaiting;
-    unsigned            readToWriteWaiting;
 #ifdef _DEBUG
     ThreadId            exclWriteOwner;
 #endif

testing/unittests/jlibtests.cpp

@@ -2067,16 +2067,16 @@ public:
         COUNTER extraValues[NUMVALUES];
     };
 
-    #define DO_TEST(LOCK, CLOCK, COUNTER, NUMVALUES, NUMLOCKS, NUMITERATIONS)   \
+    #define DO_TEST(LOCK, CLOCK, COUNTER, NUMVALUES, NUMLOCKS)   \
     { \
-        const char * title = #LOCK "," #CLOCK "," #COUNTER;\
+        const char * title = #LOCK "," #COUNTER;\
         LockTester<LOCK, CLOCK, COUNTER, NUMVALUES, NUMLOCKS> tester;\
-        uncontendedTimes.append(tester.run(title, 1, NUMITERATIONS));\
-        minorTimes.append(tester.run(title, 2, NUMITERATIONS));\
-        typicalTimes.append(tester.run(title, numCores / 2, NUMITERATIONS));\
-        tester.run(title, numCores, NUMITERATIONS);\
-        tester.run(title, numCores + 1, NUMITERATIONS);\
-        contendedTimes.append(tester.run(title, numCores * 2, NUMITERATIONS));\
+        uncontendedTimes.append(tester.run(title, 1, numIterations));\
+        minorTimes.append(tester.run(title, 2, numIterations));\
+        typicalTimes.append(tester.run(title, numCores / 2, numIterations));\
+        tester.run(title, numCores, numIterations);\
+        tester.run(title, numCores + 1, numIterations);\
+        contendedTimes.append(tester.run(title, numCores * 2, numIterations));\
     }
 
     //Use to common out a test
@@ -2096,89 +2096,35 @@ public:
     const unsigned numCores = getAffinityCpus();
     void runAllTests()
     {
-        class WriteToReadLockBlock
-        {
-            ReadWriteLock *lock;
-        public:
-            WriteToReadLockBlock(ReadWriteLock &l) : lock(&l)
-            {
-                lock->lockWrite();
-            }
-            ~WriteToReadLockBlock()
-            {
-                if (lock)
-                {
-                    lock->changeToRead();
-                    lock->unlockRead();
-                }
-            }
-        };
-
-        class ReadToWriteLockBlock
-        {
-            ReadWriteLock *lock;
-        public:
-            ReadToWriteLockBlock(ReadWriteLock &l) : lock(&l)
-            {
-                lock->lockRead();
-            }
-            ~ReadToWriteLockBlock()
-            {
-                if (lock)
-                {
-                    // try to swap to write lock, a lot threads will fail to as highly contended.
-                    unsigned attempts = 5;
-                    while (true)
-                    {
-                        if (lock->changeToWrite(1))
-                            break;
-                        else if (0 == --attempts)
-                            break;
-                    }
-                    lock->unlock();
-                }
-            }
-        };
-
-        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 1, 1, numIterations);
-        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 2, 1, numIterations);
-        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 5, 1, numIterations);
-        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 1, 2, numIterations);
-        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 1, 1, numIterations);
-        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 2, 1, numIterations);
-        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 5, 1, numIterations);
-        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 1, 2, numIterations);
-        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 1, 1, numIterations);
-        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 2, 1, numIterations);
-        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 5, 1, numIterations);
-        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 1, 2, numIterations);
-        DO_TEST(Null, Null, RelaxedAtomic<unsigned __int64>, 1, 1, numIterations);
-        DO_TEST(Null, Null, RelaxedAtomic<unsigned __int64>, 5, 1, numIterations);
-        DO_TEST(Null, Null, CasCounter, 1, 1, numIterations);
-        DO_TEST(Null, Null, CasCounter, 5, 1, numIterations);
-        DO_TEST(Null, Null, unsigned __int64, 1, 1, numIterations);
-        DO_TEST(Null, Null, unsigned __int64, 2, 1, numIterations);
-        DO_TEST(Null, Null, unsigned __int64, 5, 1, numIterations);
+        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 1, 1);
+        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 2, 1);
+        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 5, 1);
+        DO_TEST(CriticalSection, CriticalBlock, unsigned __int64, 1, 2);
+        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 1, 1);
+        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 2, 1);
+        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 5, 1);
+        DO_TEST(SpinLock, SpinBlock, unsigned __int64, 1, 2);
+        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 1, 1);
+        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 2, 1);
+        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 5, 1);
+        DO_TEST(Null, Null, std::atomic<unsigned __int64>, 1, 2);
+        DO_TEST(Null, Null, RelaxedAtomic<unsigned __int64>, 1, 1);
+        DO_TEST(Null, Null, RelaxedAtomic<unsigned __int64>, 5, 1);
+        DO_TEST(Null, Null, CasCounter, 1, 1);
+        DO_TEST(Null, Null, CasCounter, 5, 1);
+        DO_TEST(Null, Null, unsigned __int64, 1, 1);
+        DO_TEST(Null, Null, unsigned __int64, 2, 1);
+        DO_TEST(Null, Null, unsigned __int64, 5, 1);
 
         //Read locks will fail to prevent values being lost, but the timings are useful in comparison with CriticalSection
-        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 1, 1, numIterations);
-        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 2, 1, numIterations);
-        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 5, 1, numIterations);
-        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 1, 2, numIterations);
-        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 1, 1, numIterations);
-        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 2, 1, numIterations);
-        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 5, 1, numIterations);
-        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 1, 2, numIterations);
-
-        DO_TEST(ReadWriteLock, WriteToReadLockBlock, unsigned __int64, 1, 1, numIterations);
-        DO_TEST(ReadWriteLock, WriteToReadLockBlock, unsigned __int64, 2, 1, numIterations);
-        DO_TEST(ReadWriteLock, WriteToReadLockBlock, unsigned __int64, 5, 1, numIterations);
-        DO_TEST(ReadWriteLock, WriteToReadLockBlock, unsigned __int64, 1, 2, numIterations);
-
-        DO_TEST(ReadWriteLock, ReadToWriteLockBlock, unsigned __int64, 1, 1, 10000);
-        DO_TEST(ReadWriteLock, ReadToWriteLockBlock, unsigned __int64, 2, 1, 10000);
-        DO_TEST(ReadWriteLock, ReadToWriteLockBlock, unsigned __int64, 5, 1, 10000);
-        DO_TEST(ReadWriteLock, ReadToWriteLockBlock, unsigned __int64, 1, 2, 10000);
+        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 1, 1);
+        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 2, 1);
+        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 5, 1);
+        DO_TEST(ReadWriteLock, ReadLockBlock, unsigned __int64, 1, 2);
+        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 1, 1);
+        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 2, 1);
+        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 5, 1);
+        DO_TEST(ReadWriteLock, WriteLockBlock, unsigned __int64, 1, 2);
 
         printf("Summary\n");
         summariseTimings("Uncontended", uncontendedTimes);
@@ -2189,10 +2135,10 @@ public:
 
     void summariseTimings(const char * option, UInt64Array & times)
     {
-        printf("%11s 1x: cs(%3" I64F "u) spin(%3" I64F "u) atomic(%3" I64F "u) ratomic(%3" I64F "u) cas(%3" I64F "u) rd(%3" I64F "u) wr(%3" I64F "u) w2r(%3" I64F "u) r2w(%3" I64F "u) "
-                    "5x: cs(%3" I64F "u) spin(%3" I64F "u) atomic(%3" I64F "u) ratomic(%3" I64F "u) cas(%3" I64F "u) rd(%3" I64F "u) wr(%3" I64F "u) w2r(%3" I64F "u) r2w(%3" I64F "u)\n", option,
-                    times.item(0), times.item(4), times.item(8), times.item(12), times.item(14), times.item(19), times.item(23), times.item(27), times.item(31),
-                    times.item(2), times.item(6), times.item(10), times.item(13), times.item(15), times.item(21), times.item(25), times.item(29), times.item(33));
+        printf("%11s 1x: cs(%3" I64F "u) spin(%3" I64F "u) atomic(%3" I64F "u) ratomic(%3" I64F "u) cas(%3" I64F "u) rd(%3" I64F "u) wr(%3" I64F "u)   "
+                    "5x: cs(%3" I64F "u) spin(%3" I64F "u) atomic(%3" I64F "u) ratomic(%3" I64F "u) cas(%3" I64F "u) rd(%3" I64F "u) wr(%3" I64F "u)\n", option,
+                    times.item(0), times.item(4), times.item(8), times.item(12), times.item(14), times.item(19), times.item(23),
+                    times.item(2), times.item(6), times.item(10), times.item(13), times.item(15), times.item(21), times.item(25));
     }
 
 private: