From 58db8a02bc99f3f5a05125896af163996a51ee01 Mon Sep 17 00:00:00 2001 From: cxl Date: Mon, 5 Dec 2016 16:44:09 +0000 Subject: [PATCH] Core: CoWorkerResources git-svn-id: svn://ultimatepp.org/upp/trunk@10520 f0d560ea-af0d-0410-9eb7-867de7ffcac7 --- uppsrc/Core/CoWork.cpp | 12 ++++-- uppsrc/Core/CoWork.h | 27 +++++++++++-- uppsrc/Core/src.tpp/CoWork$en-us.tpp | 60 ++++++++++++++++++++++++++++ 3 files changed, 93 insertions(+), 6 deletions(-) diff --git a/uppsrc/Core/CoWork.cpp b/uppsrc/Core/CoWork.cpp index 1765c68b8..713024597 100644 --- a/uppsrc/Core/CoWork.cpp +++ b/uppsrc/Core/CoWork.cpp @@ -10,7 +10,7 @@ namespace Upp { #define LHITCOUNT(x) // RHITCOUNT(x) thread_local bool CoWork::Pool::finlock; -thread_local bool CoWork::is_worker; +thread_local int CoWork::worker_index = -1; CoWork::Pool& CoWork::GetPool() { @@ -28,7 +28,7 @@ void CoWork::Pool::InitThreads(int nthreads) { LLOG("Pool::InitThreads: " << nthreads); for(int i = 0; i < nthreads; i++) - CHECK(threads.Add().Run([=] { is_worker = true; ThreadRun(i); }, true)); + CHECK(threads.Add().Run([=] { worker_index = i; ThreadRun(i); }, true)); } void CoWork::Pool::ExitThreads() @@ -46,9 +46,15 @@ void CoWork::Pool::ExitThreads() lock.Leave(); } +int CoWork::GetPoolSize() +{ + int n = GetPool().threads.GetCount(); + return n ? n : CPU_Cores() + 2; +} + CoWork::Pool::Pool() { - ASSERT(!is_worker); + ASSERT(!IsWorker()); InitThreads(CPU_Cores() + 2); diff --git a/uppsrc/Core/CoWork.h b/uppsrc/Core/CoWork.h index 527c856ab..ae2e83c3c 100644 --- a/uppsrc/Core/CoWork.h +++ b/uppsrc/Core/CoWork.h @@ -40,9 +40,8 @@ public: static Pool& GetPool(); - static thread_local bool is_worker; + static thread_local int worker_index; -// byte magic[sizeof(ConditionVariable)]; ConditionVariable waitforfinish; Link jobs; // global stack and CoWork stack as double-linked lists int todo; @@ -72,13 +71,35 @@ public: bool IsFinished(); - static bool IsWorker() { return is_worker; } + static bool IsWorker() { return worker_index >= 0; } + static int GetWorkerIndex() { return worker_index; } + static int GetPoolSize(); static void SetPoolSize(int n); CoWork(); ~CoWork(); }; +template +class CoWorkerResources { + int workercount; + Buffer res; + +public: + int GetCount() const { return workercount + 1; } + T& operator[](int i) { return res[i]; } + + T& Get() { int i = CoWork::GetWorkerIndex(); return res[i < 0 ? workercount : i]; } + T& operator~() { return Get(); } + + CoWorkerResources() { workercount = CoWork::GetPoolSize(); res.Alloc(GetCount()); } + + CoWorkerResources(Event initializer) : CoWorkerResources() { + for(int i = 0; i < GetCount(); i++) + initializer(res[i]); + } +}; + #else class CoWork : NoCopy { diff --git a/uppsrc/Core/src.tpp/CoWork$en-us.tpp b/uppsrc/Core/src.tpp/CoWork$en-us.tpp index 949cf05c7..ce39503b8 100644 --- a/uppsrc/Core/src.tpp/CoWork$en-us.tpp +++ b/uppsrc/Core/src.tpp/CoWork$en-us.tpp @@ -108,8 +108,68 @@ non`-blocking variant of Finish).&] [s2;%% Returns true if current thread is CoWork worker thread.&] [s3; &] [s4; &] +[s5;:Upp`:`:CoWork`:`:GetWorkerIndex`(`): [@(0.0.255) static] [@(0.0.255) int]_[* GetWorker +Index]()&] +[s2;%% Returns the index of worker `- index is >`= 0 and < GetPoolSize(). +This is useful if there is a need for per`-thread resources.&] +[s3; &] +[s4; &] +[s5;:Upp`:`:CoWork`:`:GetPoolSize`(`): [@(0.0.255) static] [@(0.0.255) int]_[* GetPoolSize]( +)&] +[s2;%% Returns the current count of worker threads.&] +[s3; &] +[s4; &] [s5;:Upp`:`:CoWork`:`:SetPoolSize`(int`): [@(0.0.255) static void]_[* SetPoolSize]([@(0.0.255) i nt]_[*@3 n])&] [s2;%% Adjusts the thread pool size (default pool size is CPU`_Cores() `+ 2).&] +[s0; &] +[s0; &] +[ {{10000@(113.42.0) [s0;%% [*@7;4 CoWorkerResources]]}}&] +[s3; &] +[s1;:noref: [@(0.0.255)3 template][3 _<][@(0.0.255)3 class][3 _][*@4;3 T][3 >]&] +[s1;:Upp`:`:CoWorkerResources`:`:class: [@(0.0.255) class]_[* CoWorkerResources]&] +[s2;%% This is a simple helper class that provides per`-worker resources. +For example, certain calculation requires the instance of computation +model that is not immutable, but can be reused over iterations. +In single`-thread code, a single instance of such model would +be used over the whole loop, however in multi`-threaded code, +each worker thread, plus thread that created CoWork need its +own instance. CoWorkerResources helps to manage such situation.&] +[s0;%% &] +[ {{10000F(128)G(128)@1 [s0;%% [* Public Method List]]}}&] +[s3; &] +[s5;:Upp`:`:CoWorkerResources`:`:CoWorkerResources`(`): [* CoWorkerResources]()&] +[s2;%% Creates a required number of instances so that each sub`-job +of CoWork has its unique instance.&] +[s3; &] +[s4; &] +[s5;:Upp`:`:CoWorkerResources`:`:CoWorkerResources`(Upp`:`:Event``): [* CoWorkerR +esources]([_^Upp`:`:Event^ Event]<[*@4 T][@(0.0.255) `&]>_[*@3 initializer])&] +[s2;%% Creates a required number of instances so that each sub`-job +of CoWork has its unique instance.and initializes them using +[%-*@3 initializer].&] +[s3; &] +[s4; &] +[s5;:Upp`:`:CoWorkerResources`:`:GetCount`(`)const: [@(0.0.255) int]_[* GetCount]()_[@(0.0.255) c +onst]&] +[s2;%% Returns the number of instances. Note that this is equal to +CoWork`::GetPoolSize() if thread that created CoWorkerResources +is itself CoWork worker, or CoWork`::GetPoolSize() `+ 1 if it +is any other thread. The reason for this is that CoWork owner +can execute CoWork jobs too (while waiting in Finish).&] +[s3;%% &] +[s4; &] +[s5;:Upp`:`:CoWorkerResources`:`:operator`[`]`(int`): [*@4 T][@(0.0.255) `&]_[* operator`[`] +]([@(0.0.255) int]_[*@3 i])&] +[s2;%% Returns instance [%-*@3 i]. Together with GetCount can be used +to initialize resources (alternative method to using constructor +with [%-*@3 initializer].&] +[s3;%% &] +[s4; &] +[s5;:Upp`:`:CoWorkerResources`:`:Get`(`): [*@4 T][@(0.0.255) `&]_[* Get]()&] +[s5;:Upp`:`:CoWorkerResources`:`:operator`~`(`): [*@4 T][@(0.0.255) `&]_[* operator`~]()&] +[s2;%% Supposed to be called in CoWork job, returns a reference to +resources unique for the thread.&] +[s3; &] [s0; ]] \ No newline at end of file