[Libreoffice-commits] online.git: kit/Kit.cpp

[Ashod Nakashian]

 kit/Kit.cpp |   47 ++++++++++++++++++++++++++++++++++++++++-------
 1 file changed, 40 insertions(+), 7 deletions(-)

New commits:
commit 0fe580d9ab3abe198e3143d30c83f1cffb589832
Author: Ashod Nakashian <ashod.nakashian at collabora.co.uk>
Date:   Fri Jan 20 00:14:29 2017 -0500

    wsd: cap batches by duration
    
    Batched loKit calls are now capped by time
    to limit the maximum latency.
    
    Change-Id: Ife354228f9837dbc4c2e4f216e7f9d85487bac14
    Reviewed-on: https://gerrit.libreoffice.org/33423
    Reviewed-by: Ashod Nakashian <ashnakash at gmail.com>
    Tested-by: Ashod Nakashian <ashnakash at gmail.com>

diff --git a/kit/Kit.cpp b/kit/Kit.cpp
index dbe25a1..3f14c19 100644
--- a/kit/Kit.cpp
+++ b/kit/Kit.cpp
@@ -1379,22 +1379,50 @@ private:
 
         LOG_DBG("Thread started.");
 
+        static const unsigned MaxBatchDurationMs = 125;
         size_t batched = 0;
+        std::chrono::steady_clock::time_point timeBeginBatch;
         try
         {
             while (!_stop && !TerminationFlag)
             {
-                // End if we have no more.
+                unsigned timeoutMs = 0;
                 if (batched)
                 {
-                    LOG_TRC("Ending batch of " << batched << " messages.");
-                    std::unique_lock<std::mutex> lock(_documentMutex);
-                    batched = 0;
-                    getLOKitDocument()->endBatch();
+                    // Cap the wait so we eventually process the batch.
+                    const auto duration = std::chrono::steady_clock::now() - timeBeginBatch;
+                    const auto durationMs = std::chrono::duration_cast<std::chrono::milliseconds>(duration).count();
+                    if (durationMs >= MaxBatchDurationMs)
+                    {
+                        LOG_TRC("Ending batch of " << batched << " messages.");
+                        std::unique_lock<std::mutex> lock(_documentMutex);
+                        batched = 0;
+                        getLOKitDocument()->endBatch();
+                        timeoutMs = 0;
+                    }
+                    else
+                    {
+                        timeoutMs = MaxBatchDurationMs - durationMs;
+                    }
                 }
 
-                const TileQueue::Payload input = _tileQueue->get();
-                LOG_TRC("Kit Recv " << LOOLProtocol::getAbbreviatedMessage(input));
+                LOG_TRC("Kit dequeue with max timeout of " << timeoutMs << " ms. Batched: " << batched);
+                const TileQueue::Payload input = _tileQueue->get(timeoutMs);
+                LOG_TRC("Kit dequeued " << LOOLProtocol::getAbbreviatedMessage(input));
+                if (input.empty())
+                {
+                    // End if we have no more.
+                    if (batched)
+                    {
+                        LOG_TRC("Ending batch of " << batched << " messages.");
+                        std::unique_lock<std::mutex> lock(_documentMutex);
+                        batched = 0;
+                        getLOKitDocument()->endBatch();
+                    }
+
+                    // Nothing to process.
+                    continue;
+                }
 
                 if (_stop || TerminationFlag)
                 {
@@ -1430,6 +1458,11 @@ private:
                 else if (LOOLProtocol::getFirstToken(tokens[0], '-') == "child")
                 {
                     batched = forwardToChild(tokens[0], input, batched);
+                    if (batched == 1)
+                    {
+                        // New batch started.
+                        timeBeginBatch = std::chrono::steady_clock::now();
+                    }
                 }
                 else if (tokens[0] == "callback")
                 {