[+] AuAsync::IThreadPool::GetIOProcessor

[+] AuAsync::IThreadPool::GetIONetInterface
[+] AuAsync::IThreadPool::GetIONetWorker
[+] AuAsync::IWorkItem::SetSchedByLoopSource

Include/Aurora/Async/IThreadPool.hpp

@@ -13,6 +13,12 @@ namespace Aurora::IO::Loop
     struct ILoopSource;
 }
 
+namespace Aurora::IO::Net
+{
+    struct INetInterface;
+    struct INetWorker;
+}
+
 namespace Aurora::Async
 {
     struct IThreadPool
@@ -77,6 +83,11 @@ namespace Aurora::Async
         //
         virtual WorkerId_t GetCurrentThread()                                             = 0;
 
+        //
+        virtual AuSPtr<IO::IIOProcessor>       GetIOProcessor(WorkerId_t id)              = 0;
+        virtual AuSPtr<IO::Net::INetInterface> GetIONetInterface(WorkerId_t id)           = 0;
+        virtual AuSPtr<IO::Net::INetWorker>    GetIONetWorker(WorkerId_t id)              = 0;
+
         // Synchronization
         // Note: syncing to yourself will nullify requireSignal to prevent deadlock conditions
         virtual bool Sync(WorkerId_t workerId,

Include/Aurora/Async/IWorkItem.hpp

@@ -22,6 +22,8 @@ namespace Aurora::Async
         virtual AuSPtr<IWorkItem> WaitFor(const AuSPtr<IWorkItem> &pWorkItem)   = 0;
         virtual AuSPtr<IWorkItem> WaitFor(const AuList<AuSPtr<IWorkItem>> &workItems) = 0;
         
+        virtual AuSPtr<IWorkItem> SetSchedByLoopSource(const AuSPtr<IO::Loop::ILoopSource> &pLoopSource) = 0;
+
         // ms = time relative to the current time
         virtual AuSPtr<IWorkItem> SetSchedTime(AuUInt32 ms)                     = 0;
 

Include/Aurora/Async/IWorkItemHandler.hpp

@@ -33,6 +33,7 @@ namespace Aurora::Async
             AuUInt32 reschedClockAbsMs {};
             AuUInt64 reschedClockAbsNs {};
             AuUInt64 reschedSteadyClockAbsNs {};
+            AuSPtr<IO::Loop::ILoopSource> pLoopSource {};
             // @hideinitializer
             IThreadPool *pool;
         };

Source/Async/AsyncApp.cpp

@@ -228,6 +228,21 @@ namespace Aurora::Async
         ThreadPool::AssertWorker(id);
     }
 
+    AuSPtr<AuIO::IIOProcessor> AsyncApp::GetIOProcessor(WorkerId_t pid)
+    {
+        return ThreadPool::GetIOProcessor(pid);
+    }
+
+    AuSPtr<AuIO::Net::INetInterface> AsyncApp::GetIONetInterface(WorkerId_t pid)
+    {
+        return ThreadPool::GetIONetInterface(pid);
+    }
+
+    AuSPtr<AuIO::Net::INetWorker> AsyncApp::GetIONetWorker(WorkerId_t pid)
+    {
+        return ThreadPool::GetIONetWorker(pid);
+    }
+
     AuSPtr<AuLoop::ILoopQueue> AsyncApp::ToKernelWorkQueue()
     {
         return ThreadPool::ToKernelWorkQueue();

Source/Async/AsyncApp.hpp

@@ -42,7 +42,9 @@ namespace Aurora::Async
         void AddFeature(WorkerId_t id, AuSPtr<Threading::Threads::IThreadFeature> feature, bool async)  override;
         void AssertInThreadGroup(ThreadGroup_t group)                           override;
         void AssertWorker(WorkerId_t id)                                        override;
-
+        AuSPtr<AuIO::IIOProcessor> GetIOProcessor(WorkerId_t pid)               override;
+        AuSPtr<AuIO::Net::INetInterface> GetIONetInterface(WorkerId_t pid)      override;
+        AuSPtr<AuIO::Net::INetWorker> GetIONetWorker(WorkerId_t pid)            override;
         AuSPtr<AuLoop::ILoopQueue> ToKernelWorkQueue()                          override;
         AuSPtr<AuLoop::ILoopQueue> ToKernelWorkQueue(WorkerId_t workerId)       override;
 

Source/Async/AuThreadState.hpp

@@ -7,6 +7,8 @@
 ***/
 #pragma once
 
+#include "AuThreadStateSingletons.hpp"
+
 namespace Aurora::Async
 {
     struct AsyncLoop;
@@ -68,6 +70,7 @@ namespace Aurora::Async
         AuSPtr<AsyncLoop> asyncLoop;
         ThreadStateStack stackState;
         ThreadStateFeatureCallbacks tlsFeatures;
+        ThreadStateSingletons singletons;
 
         bool Init();
     };

Source/Async/AuThreadStateSingletons.cpp

@@ -10,5 +10,59 @@
 
 namespace Aurora::Async
 {
-
+    AuSPtr<AuIO::IIOProcessor> ThreadStateSingletons::GetIOProcessor(AuWorkerPId_t pid)
+    {
+        if (this->pIOProcessors)
+        {
+            return this->pIOProcessors;
+        }
+
+        {
+            AU_LOCK_GUARD(this->mutex);
+
+            if (this->pIOProcessors)
+            {
+                return this->pIOProcessors;
+            }
+
+            return this->pIOProcessors = AuIO::NewIOProcessorOnThread(false, pid);
+        }
+    }
+
+    void ThreadStateSingletons::TryInitNet(AuWorkerPId_t pid)
+    {
+        if (this->pNetInterface)
+        {
+            return;
+        }
+
+        AU_LOCK_GUARD(this->mutex);
+
+        auto pNetProcessor = AuNet::NewNetworkInterface();
+        if (!pNetProcessor)
+        {
+            return;
+        }
+
+        auto pNetWorker = pNetProcessor->GetWorkersService()->Attach(this->GetIOProcessor(pid));
+        if (!pNetWorker)
+        {
+            return;
+        }
+
+        this->pNetWorker = pNetWorker;
+        this->pNetInterface = pNetProcessor;
+    }
+
+    AuSPtr<AuIO::Net::INetInterface> ThreadStateSingletons::GetIONetInterface(AuWorkerPId_t pid)
+    {
+        this->TryInitNet(pid);
+        return this->pNetInterface;
+    }
+
+    AuSPtr<AuIO::Net::INetWorker> ThreadStateSingletons::GetIONetWorker(AuWorkerPId_t pid)
+    {
+        this->TryInitNet(pid);
+        return this->pNetWorker;
+    }
 }

Source/Async/AuThreadStateSingletons.hpp

@@ -7,7 +7,23 @@
 ***/
 #pragma once
 
+#include <Aurora/IO/Net/NetExperimental.hpp>
+
 namespace Aurora::Async
 {
+    struct ThreadStateSingletons
+    {
+        AuCriticalSection mutex;
 
+        AuSPtr<AuIO::IIOProcessor> pIOProcessors;
+
+        AuSPtr<AuIO::IIOProcessor> pIOProcessor;
+        AuSPtr<AuIO::Net::INetInterface> pNetInterface;
+        AuSPtr<AuIO::Net::INetWorker> pNetWorker;
+
+        AuSPtr<AuIO::IIOProcessor> GetIOProcessor(AuWorkerPId_t pid);
+        AuSPtr<AuIO::Net::INetInterface> GetIONetInterface(AuWorkerPId_t pid);
+        AuSPtr<AuIO::Net::INetWorker> GetIONetWorker(AuWorkerPId_t pid);
+        void TryInitNet(AuWorkerPId_t pid);
+    };
 }

Source/Async/ThreadPool.cpp

@@ -739,6 +739,36 @@ namespace Aurora::Async
         return tlsWorkerId;
     }
 
+    AuSPtr<AuIO::IIOProcessor> ThreadPool::GetIOProcessor(WorkerId_t pid)
+    {
+        if (auto pState = this->GetThreadHandle(pid))
+        {
+            return pState->singletons.GetIOProcessor({ this->SharedFromThis(), pid });
+        }
+
+        return {};
+    }
+
+    AuSPtr<AuIO::Net::INetInterface> ThreadPool::GetIONetInterface(WorkerId_t pid)
+    {
+        if (auto pState = this->GetThreadHandle(pid))
+        {
+            return pState->singletons.GetIONetInterface({ this->SharedFromThis(), pid });
+        }
+
+        return {};
+    }
+
+    AuSPtr<AuIO::Net::INetWorker> ThreadPool::GetIONetWorker(WorkerId_t pid)
+    {
+        if (auto pState = this->GetThreadHandle(pid))
+        {
+            return pState->singletons.GetIONetWorker({ this->SharedFromThis(), pid });
+        }
+
+        return {};
+    }
+
     bool ThreadPool::Sync(WorkerId_t workerId, AuUInt32 timeoutMs, bool requireSignal)
     {
         AU_LOCK_GUARD(this->pRWReadView);

Source/Async/ThreadPool.hpp

@@ -60,6 +60,10 @@ namespace Aurora::Async
                                                                                                        
         virtual WorkerId_t GetCurrentThread()                                                          override;
 
+        virtual AuSPtr<AuIO::IIOProcessor>       GetIOProcessor(WorkerId_t id)                         override;
+        virtual AuSPtr<AuIO::Net::INetInterface> GetIONetInterface(WorkerId_t pid)                     override;
+        virtual AuSPtr<AuIO::Net::INetWorker>    GetIONetWorker(WorkerId_t id)                         override;
+
         virtual bool Sync(WorkerId_t workerId, AuUInt32 timeoutMs, bool requireSignal)                 override;
         virtual void Signal(WorkerId_t workerId)                                                       override;
         virtual AuSPtr<AuLoop::ILoopSource> WorkerToLoopSource(WorkerId_t id)                          override;

Source/Async/WorkItem.cpp

@@ -43,8 +43,10 @@ namespace Aurora::Async
 
     WorkItem::~WorkItem()
     {
-
+        if (auto pIOWatch = AuExchange(this->pIOWatch, {}))
-        //Fail();
+        {
+            pIOWatch->StopWatch();
+        }
     }
 
     AuSPtr<IWorkItem> WorkItem::WaitFor(const AuSPtr<IWorkItem> &workItem)
@@ -172,6 +174,12 @@ namespace Aurora::Async
         return AU_SHARED_FROM_THIS;
     }
 
+    AuSPtr<IWorkItem> WorkItem::SetSchedByLoopSource(const AuSPtr<IO::Loop::ILoopSource> &pLoopSource)
+    {
+        this->pIOWatchLS = pLoopSource;
+        return AU_SHARED_FROM_THIS;
+    }
+
     AuSPtr<IWorkItem> WorkItem::SetSchedTime(AuUInt32 ms)
     {
         this->dispatchTimeNs_  = Time::SteadyClockNS() + AuMSToNS<AuUInt64>(ms);
@@ -232,6 +240,22 @@ namespace Aurora::Async
 
         this->dispatchPending_ = true;
 
+        if (this->pIOWatchLS)
+        {
+            if (!this->pIOWatchLS->IsSignaled())
+            {
+                if (!Schedule())
+                {
+                    this->Fail();
+                }
+                return;
+            }
+            else
+            {
+                AuResetMember(this->pIOWatchLS);
+            }
+        }
+
         if (Time::SteadyClockNS() < this->dispatchTimeNs_)
         {
             if (!Schedule())
@@ -324,11 +348,14 @@ namespace Aurora::Async
 
     AuSPtr<ThreadState> WorkItem::GetState()
     {
-        if (!this->worker_.HasValue())
+        if (this->worker_.HasValue())
+        {
+            return this->owner_->GetThreadHandle(this->worker_.value());
+        }
+        else
         {
             return {};
         }
-        return this->owner_->GetThreadHandle(this->worker_.value());
     }
 
     void WorkItem::RunAsyncLocked2()
@@ -398,12 +425,15 @@ namespace Aurora::Async
             {
                 SetSchedSteadyTimeNsAbs(info.reschedSteadyClockAbsNs);
             }
+            else if (info.pLoopSource)
+            {
+                SetSchedByLoopSource(info.pLoopSource);
+            }
 
             if (!WaitForLocked(info.waitFor))
             {
                 this->Fail();
             }
-
         }
         [[fallthrough]];
         case ETickType::eRerun:
@@ -419,21 +449,38 @@ namespace Aurora::Async
         }
 
         this->finished = true;
+
         if (this->finishedEvent_)
         {
             this->finishedEvent_->Set();
         }
 
-        for (auto &waiter : this->waiters_)
+        for (auto &waiter : AuExchange(this->waiters_, {}))
         {
             AuReinterpretCast<WorkItem>(waiter)->DispatchExLocked(true);
         }
+
+        this->waitOn_.clear();
+
+        if (auto pIOWatch = AuExchange(this->pIOWatch, {}))
+        {
+            pIOWatch->StopWatch();
+        }
+
+        AuResetMember(this->pIOWatchLS);
     }
 
     void WorkItem::Fail()
     {
         failed = true;
 
+        if (auto pIOWatch = AuExchange(this->pIOWatch, {}))
+        {
+            pIOWatch->StopWatch();
+        }
+
+        AuResetMember(this->pIOWatchLS);
+
         if (auto task_ = AuExchange(this->task_, {}))
         {
             task_->OnFailure();
@@ -530,9 +577,49 @@ namespace Aurora::Async
     }
     
     bool WorkItem::Schedule()
+    {
+        if (auto pLoopSource = this->pIOWatchLS)
+        {
+            if (this->pIOWatch)
+            {
+                return true;
+            }
+
+            auto pState = this->GetState();
+            if (!pState)
+            {
+                return false;
+            }
+
+            auto pIOProcessor = pState->singletons.GetIOProcessor(this->worker_.value());
+            if (!pIOProcessor)
+            {
+                return false;
+            }
+
+            this->pIOWatch = pIOProcessor->StartSimpleLSWatch(pLoopSource, AuMakeSharedThrow<AuIO::IIOSimpleEventListenerFunctional>([=]()
+            {
+                this->Dispatch();
+            }, [=]()
+            {
+                this->Dispatch();
+            }, [=]()
+            {
+                this->Dispatch();
+            }));
+
+            if (!this->pIOWatch)
+            {
+                return false;
+            }
+
+            return true;
+        }
+        else
         {
             return Async::Schedule(this->dispatchTimeNs_, this->owner_, this->worker_.value(), AuSharedFromThis());
         }
+    }
 
     void WorkItem::SendOff()
     {

Source/Async/WorkItem.hpp

@@ -32,6 +32,7 @@ namespace Aurora::Async
         AuSPtr<IWorkItem> SetSchedTimeAbs(AuUInt32 ms) override;
         AuSPtr<IWorkItem> SetSchedTimeNsAbs(AuUInt64 ns) override;
         AuSPtr<IWorkItem> SetSchedSteadyTimeNsAbs(AuUInt64 ns) override;
+        AuSPtr<IWorkItem> SetSchedByLoopSource(const AuSPtr<IO::Loop::ILoopSource> &pLoopSource) override;
         AuSPtr<IWorkItem> Then(const AuSPtr<IWorkItem> &next) override;
 
         AuSPtr<IWorkItem> Dispatch() override;
@@ -80,6 +81,8 @@ namespace Aurora::Async
         AuThreadPrimitives::Event finishedEvent_;
         AuUInt32 uShutdownCookie {};
         AuOptionalEx<AuUInt32> optOtherCookie {};
+        AuSPtr<AuIO::IIOProcessorItem> pIOWatch;
+        AuSPtr<IO::Loop::ILoopSource> pIOWatchLS;
 
         bool finished {};
         bool failed {};