HPCC-30085 Batch and write dali transactions asynchronously #17671

jakesmith · 2023-08-14T09:59:12Z

Also add an external file caching mechanism to reduce the
expense of frequently reloading recent externals.

Signed-off-by: Jake Smith [email protected]

Type of change:

This change is a bug fix (non-breaking change which fixes an issue).
This change is a new feature (non-breaking change which adds functionality).
This change improves the code (refactor or other change that does not change the functionality)
This change fixes warnings (the fix does not alter the functionality or the generated code)
This change is a breaking change (fix or feature that will cause existing behavior to change).
This change alters the query API (existing queries will have to be recompiled)

Checklist:

Smoketest:

Send notifications about my Pull Request position in Smoketest queue.
Test my draft Pull Request.

Testing:

jakesmith · 2023-08-16T15:18:55Z

@ghalliday - please give this a 1st review.

ghalliday

@jakesmith a few minor comments from an initial scan. I will revisit again and review properly.

ghalliday · 2023-08-21T15:55:56Z

initfiles/componentfiles/configschema/xsd/dali.xsd

+                                  hpcc:displayName="Seconds between transactions being committed to disk" hpcc:presetValue="0"
+                                  hpcc:tooltip="The maximum time between commit pending transactions to disk (default=0, meaning commit immediately)"/>
+                    <xs:attribute name="deltaTransactionQueueLimit" type="xs:nonNegativeInteger"
+                                  hpcc:displayName="Maximum number of pending uncommited transaction" hpcc:presetValue="10000"


trivial: uncommitted

ghalliday · 2023-08-21T16:01:25Z

dali/base/dasds.cpp

+        type = f_delta;
+        name = strdup(path);
+    }
+    CTransactionItem(char *_name, size32_t _dataLength, void *_data) : name(_name), dataLength(_dataLength), data(_data)


Unusual that one constructor clones the name, and the other takes ownership.

agree, I'll make it consistently take ownership, and clone the string outside.

ghalliday · 2023-08-21T16:03:18Z

dali/base/dasds.cpp

+        deltaFilename.clear().append(backupPath);
+        iStoreHelper->getCurrentDeltaFilename(deltaFilename);
+        OwnedIFile iFileDeltaBackup = createIFile(deltaFilename.str());
+        if (!compareFiles(iFileDeltaBackup, iFileDelta, false))


minor: A better function name would be filesMatch/filesAreIdentical - more specific, and reads better in the if

not new, but agreed, I'll change it.

ghalliday · 2023-08-21T16:22:53Z

dali/base/dasds.cpp

+        StringBuffer fname(dataPath);
+        OwnedIFile deltaIPIFile = createIFile(fname.append(DELTAINPROGRESS).str());
+        OwnedIFileIO deltaIPIFileIO = deltaIPIFile->open(IFOcreate);
+        deltaIPIFileIO.clear();


Why the close? Worth a comment at least.

not new code, but relocated. This code is effectively a touch()
Just looked, and there is a jfile touchFile function, but not for an IFile, nor a removeFile for a filename - I may add, to clarify what this is doing.

ghalliday · 2023-08-22T07:43:22Z

dali/base/dasds.cpp

+    size_t items = pending.size();
+    if ((pendingSz < transactionMaxMem) && (items < transactionQueueLimit))
+    {
+        if (nextTimeThreshold && (get_cycles_now() < nextTimeThreshold))


There is the potential for wrapping - if nextTimeThreshold gets to (cycles_type)-1 then this is unlikely to every be true. Always safer to test (get_cycles_now() - prevWriteTime) > thresholdDuration

right, will change.

ghalliday · 2023-08-22T07:46:51Z

dali/base/dasds.cpp

-        item->flags = BackupQueueItem::f_addext;
-        add(item);
+        CriticalBlock b(pendingCrit);
+        CTransactionItem *item = new CTransactionItem(name, length, content);


efficiency: create the item outside of the critical section.

yes, I'll move the creation of these.

ghalliday · 2023-08-22T07:47:04Z

dali/base/dasds.cpp

-        item->flags = BackupQueueItem::f_delext;
-        add(item);
+        CriticalBlock b(pendingCrit);
+        addToQueue(new CTransactionItem(path, delta));


efficiency: create the item outside of the critical section.

ghalliday · 2023-08-22T07:49:23Z

dali/base/dasds.cpp

-            e->Release();
+            aborted = true;
+            sem.signal();
+            pendingCrit.enter();


Worth a comment why you are doing this.

ghalliday · 2023-08-22T07:53:23Z

dali/base/dasds.cpp

+class CBinaryFileExternal : public CExternalFile, implements IExternalHandler
+{
+public:
+    IMPLEMENT_IINTERFACE;


Should really be
IMPLEMENT_IINTERFACE_USING(CExternalFile)

ghalliday · 2023-08-22T08:06:30Z

dali/base/dasds.cpp

+            ++throttleCounter;
+            if (timeThrottled >= queryOneSecCycles())
+            {
+                IWARNLOG("Transactions throttled - current items = %u, since last message throttled-time/tracactions = { %u ms, %u }, total hard limit hits = %u", (unsigned)items, (unsigned)cycle_to_millisec(timeThrottled), throttleCounter, totalQueueLimitHits);


typo?: tracactions->transactions?

will correct.

jakesmith · 2023-08-22T11:17:19Z

@ghalliday - please see new commit

ghalliday · 2023-08-30T14:17:42Z

dali/base/dasds.cpp

+    Semaphore sem;
+    cycle_t timeThrottled = 0;
+    unsigned throttleCounter = 0;
+    bool waiting = true;


This needs to be atomic

does it? It's always altered or read whilst pendingCrit is held.

ghalliday

@jakesmith a few relatively minor comments.

ghalliday · 2023-08-30T14:28:00Z

dali/base/dasds.cpp

+                // external file
+
+                // commit accumulated delta 1st (so write order is consistent)
+                if (deltaXml.length())


Could you possibly do this in two phases?
a) write new files, but don't flush xml
after all written
b) remove all external.

ghalliday · 2023-08-30T14:32:52Z

dali/base/dasds.cpp

+            StringBuffer deltaFilename(dataPath);
+            iStoreHelper->getCurrentDeltaFilename(deltaFilename);
+            OwnedIFile iFile = createIFile(deltaFilename.str());
+            writeDelta(deltaXml, *iFile);


crc calculation in that function is a bit expensive. Might be worth considering other options/implementations.

New JIRA: https://track.hpccsystems.com/browse/HPCC-30191

ghalliday · 2023-08-31T08:45:18Z

dali/base/dasds.cpp

+    unsigned throttleCounter = 0;
+    bool waiting = true;
+    bool backupOutOfSync = false;
+    bool aborted = false;


should be atomic

agreed, will change.

ghalliday · 2023-08-31T08:48:07Z

dali/base/dasds.cpp

+                cleanChangeTree(*changeTree);
+
+                // write out with header details (e.g. path)
+                Owned<IPropertyTree> header = createPTree("Header");


efficiency: Not as clean, but would it be more efficient to explicitly write out the text
"<Header path='%s'><Delta>"..."</Delta></Header>"
than the overhead of creating the tree nodes?

yes could do, cuts down on the 'Header' and 'Delta' trees.
I'll change.

ghalliday · 2023-08-31T09:07:39Z

dali/base/dasds.cpp

-        item->data = data; // take ownership
-        item->flags = BackupQueueItem::f_addext;
-        add(item);
+        CTransactionItem *item = new CTransactionItem(name, length, content);


Should this be checking aborted still? (And the other similar functions)

I think it highlights an issue. The deltaWriter is cleared when blockingSave is called (when Dali itself saves all in memory state, the pending 'to be saved' transactions are cleared). However, when Dali shuts down that is not done.
The delta writer should both be cleared and stopped, although by the time it (Dali) is stopping and saving, there will be no more transactions added.

I will make some changes.

ghalliday · 2023-08-31T09:17:14Z

dali/base/dasds.cpp

+    serializeVisibleAttributes(owner, mb);
+    StringBuffer extName;
+    getName(extName, name);
+    if (manager.extCache.lookup(extName, mb))


Should withValue be passed/tested before this? Is it something that occurs often?

Should withValue be passed/tested before this?

it does look that way, I need to double check.

Is it something that occurs often?

is hitting the cache something that happens often?
Yes, in testing, the same external is often re-read in quick succession.

ghalliday · 2023-08-31T09:17:41Z

dali/base/dasds.cpp

+        getFilename(filename, name);
+        Owned<IFile> iFile = createIFile(filename.str());
+        size32_t sz = (size32_t)iFile->size();
+        if ((unsigned)-1 == sz)


Better to cast to size32_t for consistency.

agree, will change.

ghalliday · 2023-08-31T09:18:07Z

dali/base/dasds.cpp

+    getFilename(filename, name);
+    Owned<IFile> iFile = createIFile(filename.str());
+    size32_t sz = (size32_t)iFile->size();
+    if ((unsigned)-1 == sz)


Better to cast to size32_t for consistency.

will change.

ghalliday · 2023-08-31T09:20:53Z

dali/base/dasds.cpp

+        else
+        {
+            Owned<IFileIO> fileIO = iFile->open(IFOread);
+            verifyex(sz == ::read(fileIO, 0, sz, mb));


read function unnecessarily checks size() again. I don't know if that will cause an extra delay (on remote storage), but may be worth avoiding.

looks like it would at least be worth modifying ::read to avoid the check:

const size32_t checkLengthLimit = 0x1000; if (len >= checkLengthLimit) { //Don't allocate a stupid amount of memory.... offset_t fileLength = in->size(); if (pos > fileLength) pos = fileLength; if ((len == (size32_t)-1) || (pos + len > fileLength)) len = (size32_t)(fileLength - pos); }

if pos == 0 and len != -1

makes sense to do in separate JIRA/PR : https://track.hpccsystems.com/browse/HPCC-30193

ghalliday · 2023-08-31T09:34:32Z

dali/base/dasds.cpp

+        // force a synchronous save
+        CCycleTimer timer;
+        CHECKEDCRITICALBLOCK(blockedSaveCrit, fakeCritTimeout); // because if Dali is saving state (::blockingSave), it will clear pending
+        PROGLOG("Forcing synchronous save of %u transactions", (unsigned)pending.size());


Move before the critsec? I suspect it is unlikely to be contended so not at all significant.

but yes, logical to move before

jakesmith · 2023-08-31T14:52:41Z

dali/base/dasds.cpp

            thresholdDuration = queryOneSecCycles() * saveThresholdSecs;
+            lastSaveTime = get_cycles_now();


NB: this is an incidental change. Not being set before meant it would save without checking threshold early than intended (not a big deal, but more correct to set time at start)

jakesmith · 2023-08-31T15:07:28Z

@ghalliday - please review new commit

ghalliday

@jakesmith looks close to ready. A few comments on naming, and I think one potential hole.

ghalliday · 2023-09-04T13:44:01Z

dali/base/dasds.cpp

+                if (pendingSz)
+                    signalWaiting();
+            }
+            if (!waiting)


trivial: else?

it's intentionally not an else, because signalWaiting (if called above) wil reset waiting to false

ghalliday · 2023-09-04T14:00:27Z

dali/base/dasds.cpp

@@ -1407,17 +1427,36 @@ class CDeltaWriter : implements IThreaded
        CriticalBlock b(pendingCrit);
        addToQueue(item);
    }
-    void clear()
+    void flush()


This should be commented because it isn't clear what the different flags mean etc. Something like
//ensure that all pending transactions have been flushed to the file.

if (waiting)
{
// The writer thread is actively waiting - if there is data to write ensure it is written
}
else
{
//delta is currently being written. Wait until all the data is written.

will update with some comments.

ghalliday · 2023-09-04T14:04:03Z

dali/base/dasds.cpp

+            if (waiting)
+            {
+                if (pendingSz)
+                    signalWaiting();


Shouldn't this also wait until until signalEmpty occurs? Othewise it might still be writing data (or not yet have written the data) before the function continues.

it will because sianglWaiting resets waiting to false (all within pendingCrit), and so it will fall into the if (!waiting) block below.

ghalliday · 2023-09-04T14:11:51Z

dali/base/dasds.cpp

+        bool waitOnEmptySem = false;
+        {
+            CriticalBlock b(pendingCrit);
+            if (waiting)


waiting is a very confusing variable name - especially since it is set from thread and cleared in another. It actually seems to mean.

needToSendSignalToRequestDataIsWritten

I think it would be worth coming up with a clearer name - it would help check the logic is correct.

I've renamed it deltaWriterIdle, which I think is more accurate and clarifies the logic IMO.

ghalliday · 2023-09-04T14:13:39Z

dali/base/dasds.cpp

+                waitOnEmptySem = true;
+            }
+        }
+        if (waitOnEmptySem)


"allWritten" would be a bit clearer than "empty"

agree, will rename this and some others to clarify meaning.

ghalliday · 2023-09-04T14:16:02Z

dali/base/dasds.cpp

@@ -1445,6 +1484,11 @@ class CDeltaWriter : implements IThreaded
                    if (semTimedout)


I think this can be
if (semTimedout && !needToSendSignalToRequestDataIsWritten)
{
sem.wait(); // Consume the signal that was sent
needToSendSignalToRequestDataIsWritten = true;
}

I think that's true, I'll modify.

ghalliday · 2023-09-04T14:17:20Z

dali/base/dasds.cpp

@@ -1445,6 +1484,11 @@ class CDeltaWriter : implements IThreaded
                    if (semTimedout)
                        sem.wait(0);
                    waiting = true;
+                    if (signalEmpty)


signalWhen...

will rename.

ghalliday

@jakesmith please squash. A couple of comments to clarify the code, but I think it is functional. Will scan once more when squashed.

ghalliday · 2023-09-05T11:16:44Z

dali/base/dasds.cpp

-        bool waitOnEmptySem = false;
+        // ensure all pending transactions are written out
+        // if necessary wait until CDeltaWriter thread has finished
+        bool allWritten = false;


picky: really waitUntilAllWritten

will change

ghalliday · 2023-09-05T11:19:42Z

dali/base/dasds.cpp

                    signalWaiting();
            }
-            if (!waiting)
+            if (!writerIdle) // i.e. if not idle, it is busy writing


Worth having a comment that this includes if signalWaiting() was called above. Otherwise it is very tempting to turn this if into an else, because it strange that writerIdle would be set by the requester.
writerRequested would be clearer than writerIdle, but would invert the condition, so safer not to make that change now.

signalWaiting() would similarly be better renamed as request[Async]Write()

have made these changes, but kept writerIdle -> writeRequested change in its own commit to allow easier review

ghalliday

@jakesmith please squash

Also add an external file caching mechanism to reduce the expense of frequently reloading recent externals. Signed-off-by: Jake Smith <[email protected]>

jakesmith · 2023-09-05T14:09:00Z

@ghalliday - squashed

jakesmith force-pushed the HPCC-30085-dali-async-batch branch 2 times, most recently from c12cc11 to bc5c527 Compare August 16, 2023 11:32

jakesmith changed the title ~~Hpcc 30085 dali async batch~~ HPCC-30085 Batch and write dali transactions asynchronously Aug 16, 2023

jakesmith force-pushed the HPCC-30085-dali-async-batch branch 2 times, most recently from ec2f450 to 32d924a Compare August 16, 2023 15:17

jakesmith marked this pull request as ready for review August 16, 2023 15:18

jakesmith force-pushed the HPCC-30085-dali-async-batch branch from 32d924a to 5ce8c5e Compare August 16, 2023 15:18

jakesmith requested a review from ghalliday August 16, 2023 15:18

ghalliday reviewed Aug 22, 2023

View reviewed changes

jakesmith requested a review from ghalliday August 22, 2023 11:16

ghalliday reviewed Aug 30, 2023

View reviewed changes

ghalliday reviewed Aug 31, 2023

View reviewed changes

jakesmith requested a review from ghalliday August 31, 2023 14:49

jakesmith commented Aug 31, 2023

View reviewed changes

jakesmith force-pushed the HPCC-30085-dali-async-batch branch from 70d928c to 8c2ceb0 Compare August 31, 2023 15:06

ghalliday reviewed Sep 4, 2023

View reviewed changes

jakesmith requested a review from ghalliday September 4, 2023 15:21

ghalliday approved these changes Sep 5, 2023

View reviewed changes

HPCC-30085 Batch and write dali transactions asynchronously

0a10a8a

Also add an external file caching mechanism to reduce the expense of frequently reloading recent externals. Signed-off-by: Jake Smith <[email protected]>

jakesmith force-pushed the HPCC-30085-dali-async-batch branch from b656757 to 0a10a8a Compare September 5, 2023 14:08

jakesmith requested a review from ghalliday September 5, 2023 14:08

ghalliday approved these changes Sep 5, 2023

View reviewed changes

ghalliday merged commit d502065 into hpcc-systems:master Sep 5, 2023
29 checks passed

		thresholdDuration = queryOneSecCycles() * saveThresholdSecs;
		lastSaveTime = get_cycles_now();

		@@ -1445,6 +1484,11 @@ class CDeltaWriter : implements IThreaded
		if (semTimedout)

HPCC-30085 Batch and write dali transactions asynchronously #17671

HPCC-30085 Batch and write dali transactions asynchronously #17671

Conversation

jakesmith commented Aug 14, 2023 • edited Loading

Type of change:

Checklist:

Smoketest:

Testing:

jakesmith commented Aug 16, 2023

ghalliday left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jakesmith commented Aug 22, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghalliday left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jakesmith Aug 31, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jakesmith commented Aug 31, 2023

ghalliday left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ghalliday left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jakesmith Sep 5, 2023 • edited Loading

Choose a reason for hiding this comment

jakesmith commented Aug 14, 2023 •

edited

Loading

jakesmith Aug 31, 2023 •

edited

Loading

jakesmith Sep 5, 2023 •

edited

Loading