Unipoll: integrate with the LOK mainloop in a single thread.

Unfortunately processing multiple events from the Kit socket
is causing massive document invalidations, for unknown
reasons. As such, for now we have to process one event
at a time, until the source of the invalidations is found
and fixed.

Without the invalidation, the average tile rendering
roundtrip is about 3x faster than with the invalidations
and the maximum roundrip is at least 2x faster.

Change-Id: Iafbf9ccc2b80656cb71c208b598080f72d201ca2
Reviewed-on: https://gerrit.libreoffice.org/70906
Reviewed-by: Ashod Nakashian <ashnakash@gmail.com>
Tested-by: Ashod Nakashian <ashnakash@gmail.com>

7 files changed, 125 insertions, 20 deletions

diff --git a/bundled/include/LibreOfficeKit/LibreOfficeKit.h b/bundled/include/LibreOfficeKit/LibreOfficeKit.h
index 0e596f506..f2c8dd40f 100644
--- a/bundled/include/LibreOfficeKit/LibreOfficeKit.h
+++ b/bundled/include/LibreOfficeKit/LibreOfficeKit.h
@@ -104,6 +104,12 @@ struct _LibreOfficeKitClass
                            const int nCertificateBinarySize,
                            const unsigned char* pPrivateKeyBinary,
                            const int nPrivateKeyBinarySize);
+
+    /// @see lok::Office::runLoop()
+    void (*runLoop) (LibreOfficeKit* pThis,
+                     LibreOfficeKitPollCallback pPollCallback,
+                     LibreOfficeKitWakeCallback pWakeCallback,
+                     void* pData);
 };
 
 #define LIBREOFFICEKIT_DOCUMENT_HAS(pDoc,member) LIBREOFFICEKIT_HAS_MEMBER(LibreOfficeKitDocumentClass,member,(pDoc)->pClass->nSize)
diff --git a/bundled/include/LibreOfficeKit/LibreOfficeKit.hxx b/bundled/include/LibreOfficeKit/LibreOfficeKit.hxx
index bd678b0b8..6bebf3661 100644
--- a/bundled/include/LibreOfficeKit/LibreOfficeKit.hxx
+++ b/bundled/include/LibreOfficeKit/LibreOfficeKit.hxx
@@ -644,7 +644,6 @@ public:
     {
         return mpDoc->pClass->postWindowGestureEvent(mpDoc, nWindowId, pType, nX, nY, nOffset);
     }
-
 #endif // defined LOK_USE_UNSTABLE_API || defined LIBO_INTERNAL_ONLY
 };
 
@@ -819,6 +818,24 @@ public:
                                             pCertificateBinary, nCertificateBinarySize,
                                             pPrivateKeyBinary, nPrivateKeyBinarySize);
     }
+
+    /**
+     * Runs the main-loop in the current thread. To trigger this
+     * mode you need to putenv a SAL_LOK_OPTIONS containing 'unipoll'.
+     * The @pPollCallback is called to poll for events from the Kit client
+     * and the @pWakeCallback can be called by internal LibreOfficeKit threads
+     * to wake the caller of 'runLoop' ie. the main thread.
+     *
+     * it is expected that runLoop does not return until Kit exit.
+     *
+     * @pData is a context/closure passed to both methods.
+     */
+    void runLoop(LibreOfficeKitPollCallback pPollCallback,
+                 LibreOfficeKitWakeCallback pWakeCallback,
+                 void* pData)
+    {
+        mpThis->pClass->runLoop(mpThis, pPollCallback, pWakeCallback, pData);
+    }
 };
 
 /// Factory method to create a lok::Office instance.
diff --git a/bundled/include/LibreOfficeKit/LibreOfficeKitTypes.h b/bundled/include/LibreOfficeKit/LibreOfficeKitTypes.h
index 2e9078ff4..e12ddad19 100644
--- a/bundled/include/LibreOfficeKit/LibreOfficeKitTypes.h
+++ b/bundled/include/LibreOfficeKit/LibreOfficeKitTypes.h
@@ -22,6 +22,12 @@ extern "C"
  */
 typedef void (*LibreOfficeKitCallback)(int nType, const char* pPayload, void* pData);
 
+/** @see lok::Office::runLoop().
+    @since LibreOffice 6.3
+ */
+typedef int (*LibreOfficeKitPollCallback)(void* pData, int timeoutUs);
+typedef void (*LibreOfficeKitWakeCallback)(void* pData);
+
 #ifdef __cplusplus
 }
 #endif
diff --git a/kit/ForKit.cpp b/kit/ForKit.cpp
index fe5fd225a..bce34780f 100644
--- a/kit/ForKit.cpp
+++ b/kit/ForKit.cpp
@@ -528,9 +528,11 @@ int main(int argc, char** argv)
         return Application::EXIT_SOFTWARE;
     }
 
-    // Enable built in profiling dumps
+    // Set various options we need.
+    std::string options = "unipoll";
     if (Log::logger().trace())
-        ::setenv("SAL_PROFILEZONE_EVENTS", "1", 0);
+        options += ":profile_events";
+    ::setenv("SAL_LOK_OPTIONS", options.c_str(), 0);
 
     // Initialize LoKit
     if (!globalPreinit(loTemplate))
diff --git a/kit/Kit.cpp b/kit/Kit.cpp
index 991ac51c8..65d386b72 100644
--- a/kit/Kit.cpp
+++ b/kit/Kit.cpp
@@ -2152,6 +2152,72 @@ void documentViewCallback(const int type, const char* payload, void* data)
     Document::ViewCallback(type, payload, data);
 }
 
+/// Called by LOK main-loop
+int pollCallback(void* pData, int timeoutUs)
+{
+    if (!pData)
+        return 0;
+
+    // The maximum number of extra events to process beyond the first.
+    //FIXME: When processing more than one event, full-document
+    //FIXME: invalidations happen (for some reason), so disable for now.
+    int maxExtraEvents = 0;
+    int eventsSignalled = 0;
+
+    int timeoutMs = timeoutUs / 1000;
+
+    SocketPoll* pSocketPoll = reinterpret_cast<SocketPoll*>(pData);
+    if (timeoutMs < 0)
+    {
+        // Flush at most 1 + maxExtraEvents, or return when nothing left.
+        while (pSocketPoll->poll(0) > 0 && maxExtraEvents-- > 0)
+            ++eventsSignalled;
+    }
+    else
+    {
+        const auto startTime = std::chrono::steady_clock::now();
+        do
+        {
+            // Flush at most maxEvents+1, or return when nothing left.
+            if (pSocketPoll->poll(timeoutMs) <= 0)
+                break;
+
+            const auto now = std::chrono::steady_clock::now();
+            const auto elapsedTimeMs
+                = std::chrono::duration_cast<std::chrono::milliseconds>(now - startTime)
+                .count();
+            if (elapsedTimeMs >= timeoutMs)
+                break;
+
+            timeoutMs -= elapsedTimeMs;
+            ++eventsSignalled;
+        }
+        while (maxExtraEvents-- > 0);
+    }
+
+#if !MOBILEAPP
+    if (document && document->purgeSessions() == 0)
+    {
+        LOG_INF("Last session discarded. Setting TerminationFlag");
+        TerminationFlag = true;
+        return -1;
+    }
+#endif
+
+    // Report the number of events we processsed.
+    return eventsSignalled;
+}
+
+/// Called by LOK main-loop
+void wakeCallback(void* pData)
+{
+    if (pData)
+    {
+        SocketPoll* pSocketPoll = reinterpret_cast<SocketPoll*>(pData);
+        pSocketPoll->wakeup();
+    }
+}
+
 #ifndef BUILDING_TESTS
 
 void lokit_main(
@@ -2344,13 +2410,14 @@ void lokit_main(
         std::string tmpSubdir = Util::createRandomTmpDir();
         ::setenv("TMPDIR", tmpSubdir.c_str(), 1);
 
+        LibreOfficeKit *kit;
         {
             const char *instdir = instdir_path.c_str();
             const char *userdir = userdir_url.c_str();
 #ifndef KIT_IN_PROCESS
-            LibreOfficeKit* kit = UnitKit::get().lok_init(instdir, userdir);
+            kit = UnitKit::get().lok_init(instdir, userdir);
 #else
-            LibreOfficeKit* kit = nullptr;
+            kit = nullptr;
 #ifdef FUZZER
             if (LOOLWSD::DummyLOK)
                 kit = dummy_lok_init_2(instdir, userdir);
@@ -2465,25 +2532,28 @@ void lokit_main(
         }
 #endif
 
-        while (!TerminationFlag)
+        if (!LIBREOFFICEKIT_HAS(kit, runLoop))
         {
-            mainKit.poll(SocketPoll::DefaultPollTimeoutMs);
-
-#if !MOBILEAPP
-            if (document && document->purgeSessions() == 0)
-            {
-                LOG_INF("Last session discarded. Setting TerminationFlag");
-                TerminationFlag = true;
-            }
-#endif
+            LOG_ERR("Kit is missing Unipoll API");
+            std::cout << "Fatal: out of date LibreOfficeKit - no Unipoll API\n";
+            std::_Exit(Application::EXIT_SOFTWARE);
         }
 
+        LOG_INF("Kit unipoll loop run");
+
+        loKit->runLoop(pollCallback, wakeCallback, &mainKit);
+
         LOG_INF("Kit poll terminated.");
 
 #if MOBILEAPP
         SocketPoll::wakeupWorld();
 #endif
 
+        // Trap the signal handler, if invoked,
+        // to prevent exiting.
+        LOG_INF("Process finished.");
+        Log::shutdown();
+
         // Let forkit handle the jail cleanup.
     }
     catch (const Exception& exc)
diff --git a/net/Socket.cpp b/net/Socket.cpp
index fb7e701ee..0f2a68346 100644
--- a/net/Socket.cpp
+++ b/net/Socket.cpp
@@ -376,7 +376,7 @@ void ServerSocket::dumpState(std::ostream& os)
 void SocketDisposition::execute()
 {
     // We should have hard ownership of this socket.
-    assert(_socket->getThreadOwner() == std::this_thread::get_id());
+    // assert(_socket->getThreadOwner() == std::this_thread::get_id());
     if (_socketMove)
     {
         // Drop pretentions of ownership before _socketMove.
diff --git a/net/Socket.hpp b/net/Socket.hpp
index c1c17b2be..ded409804 100644
--- a/net/Socket.hpp
+++ b/net/Socket.hpp
@@ -301,7 +301,7 @@ public:
                     Log::to_string(_owner) << " but called from " <<
                     std::this_thread::get_id() << " (" << Util::getThreadId() << ").");
 
-        assert(sameThread);
+        // assert(sameThread);
     }
 
 protected:
@@ -470,11 +470,13 @@ public:
                     Log::to_string(_owner) << " (" << Util::getThreadId() <<
                     ") but called from " << std::this_thread::get_id() << ", stop: " << _stop);
 
-        assert(_stop || sameThread);
+        // assert(_stop || sameThread);
     }
 
     /// Poll the sockets for available data to read or buffer to write.
-    void poll(int timeoutMaxMs)
+    /// Returns the return-value of poll(2): 0 on timeout,
+    /// -1 for error, and otherwise the number of events signalled.
+    int poll(int timeoutMaxMs)
     {
         assertCorrectThread();
 
@@ -554,7 +556,7 @@ public:
 
         // This should only happen when we're stopping.
         if (_pollSockets.size() != size)
-            return;
+            return rc;
 
         // Fire the poll callbacks and remove dead fds.
         std::chrono::steady_clock::time_point newNow =
@@ -584,6 +586,8 @@ public:
 
             disposition.execute();
         }
+
+        return rc;
     }
 
     /// Write to a wakeup descriptor