1 files changed, 45 insertions, 3 deletions

diff --git a/src/profiling/FileOnlyProfilingConnection.hpp b/src/profiling/FileOnlyProfilingConnection.hpp
index d4477b6883..12ac27333d 100644
--- a/src/profiling/FileOnlyProfilingConnection.hpp
+++ b/src/profiling/FileOnlyProfilingConnection.hpp
@@ -5,15 +5,19 @@
 
 #pragma once
 
-#include "CounterDirectory.hpp"
+#include <armnn/profiling/ILocalPacketHandler.hpp>
 #include "DirectoryCaptureCommandHandler.hpp"
 #include "IProfilingConnection.hpp"
+#include "Packet.hpp"
 #include "ProfilingUtils.hpp"
 #include "Runtime.hpp"
 
+#include <atomic>
 #include <condition_variable>
 #include <fstream>
+#include <mutex>
 #include <queue>
+#include <thread>
 
 namespace armnn
 {
@@ -42,7 +46,20 @@ public:
         : m_Options(options)
         , m_QuietOp(quietOp)
         , m_Endianness(TargetEndianness::LeWire)    // Set a sensible default. WaitForStreamMeta will set a real value.
-        {};
+        , m_IsRunning(false)
+        , m_KeepRunning(false)
+        , m_Timeout(1000)
+    {
+        for (ILocalPacketHandlerSharedPtr localPacketHandler : options.m_LocalPacketHandlers)
+        {
+            AddLocalPacketHandler(localPacketHandler);
+        }
+        if (!options.m_LocalPacketHandlers.empty())
+        {
+            StartProcessingThread();
+        }
+        // NOTE: could add timeout to the external profiling options
+    };
 
     ~FileOnlyProfilingConnection();
 
@@ -57,6 +74,11 @@ public:
     Packet ReadPacket(uint32_t timeout) override;
 
 private:
+    void AddLocalPacketHandler(ILocalPacketHandlerSharedPtr localPacketHandler);
+    void StartProcessingThread();
+    void ClearReadableList();
+    void DispatchPacketToHandlers(const Packet& packet);
+
     bool WaitForStreamMeta(const unsigned char* buffer, uint32_t length);
 
     uint32_t ToUint32(const unsigned char* data, TargetEndianness endianness);
@@ -65,10 +87,13 @@ private:
 
     bool SendCounterSelectionPacket();
 
-    PackageActivity GetPackageActivity(const unsigned char* buffer, uint32_t headerAsWords[2]);
+    PackageActivity GetPackageActivity(const Packet& packet, uint32_t headerAsWords[2]);
 
     void Fail(const std::string& errorMessage);
 
+    void ForwardPacketToHandlers(Packet& packet);
+    void ServiceLocalHandlers();
+
     static const uint32_t PIPE_MAGIC = 0x45495434;
 
     Runtime::CreationOptions::ExternalProfilingOptions m_Options;
@@ -79,6 +104,23 @@ private:
 
     std::mutex m_PacketAvailableMutex;
     std::condition_variable m_ConditionPacketAvailable;
+
+    std::vector<ILocalPacketHandlerSharedPtr> m_PacketHandlers;
+    std::map<uint32_t, std::vector<ILocalPacketHandlerSharedPtr>> m_IndexedHandlers;
+    std::vector<ILocalPacketHandlerSharedPtr> m_UniversalHandlers;
+
+    // List of readable packets for the local packet handlers
+    std::queue<Packet> m_ReadableList;
+    // Mutex and condition variable for the readable packet list
+    std::mutex m_ReadableMutex;
+    std::condition_variable m_ConditionPacketReadable;
+    // thread that takes items from the readable list and dispatches them
+    // to the handlers.
+    std::thread m_LocalHandlersThread;
+    // atomic booleans that control the operation of the local handlers thread
+    std::atomic<bool> m_IsRunning;
+    std::atomic<bool> m_KeepRunning;
+    int m_Timeout;
 };
 
 }    // namespace profiling