From 64de02adab290cb813969447fd143d3589d6603b Mon Sep 17 00:00:00 2001 From: cxl Date: Fri, 8 Jan 2016 21:57:22 +0000 Subject: [PATCH] Core: CoWork optimized git-svn-id: svn://ultimatepp.org/upp/trunk@9374 f0d560ea-af0d-0410-9eb7-867de7ffcac7 --- uppsrc/Core/CoWork.cpp | 60 +++++++++++++++++++--------- uppsrc/Core/CoWork.h | 20 ++++++---- uppsrc/Core/Debug.cpp | 1 - uppsrc/Core/Profile.h | 5 +-- uppsrc/Core/src.tpp/Thread$en-us.tpp | 8 +++- 5 files changed, 63 insertions(+), 31 deletions(-) diff --git a/uppsrc/Core/CoWork.cpp b/uppsrc/Core/CoWork.cpp index 710115a84..3bcff0872 100644 --- a/uppsrc/Core/CoWork.cpp +++ b/uppsrc/Core/CoWork.cpp @@ -7,6 +7,8 @@ NAMESPACE_UPP #define LLOG(x) // DLOG(x) #define LDUMP(x) // DDUMP(x) +#define LHITCOUNT(x) // RHITCOUNT(x) + CoWork::Pool& CoWork::pool() { static Pool pool; @@ -16,6 +18,7 @@ CoWork::Pool& CoWork::pool() CoWork::Pool::Pool() { LLOG("CoWork INIT pool: " << CPU_Cores() + 2); + scheduled = 0; for(int i = 0; i < CPU_Cores() + 2; i++) threads.Add().Run(callback1(&ThreadRun, i)); } @@ -23,11 +26,9 @@ CoWork::Pool::Pool() CoWork::Pool::~Pool() { LLOG("Quit"); - MJob job; - job.work = NULL; lock.Enter(); - jobs.Clear(); - jobs.Add(job); + jobs[0].work = NULL; + scheduled = 1; lock.Leave(); for(int i = 0; i < threads.GetCount(); i++) waitforjob.Release(); @@ -39,13 +40,23 @@ CoWork::Pool::~Pool() bool CoWork::Pool::DoJob() { Pool& p = pool(); - if(p.jobs.Top().work == NULL) { + MJob& job = p.jobs[p.scheduled - 1]; + if(job.work == NULL) { LLOG("Quit thread"); return true; } - MJob job = p.jobs.Pop(); - p.lock.Leave(); - job.cb(); + std::function fn = std::move(job.fn); + if(fn) { + p.scheduled--; + p.lock.Leave(); + fn(); + } + else { + Callback cb = job.cb; + p.scheduled--; + p.lock.Leave(); + cb(); + } p.lock.Enter(); if(--job.work->todo == 0) { LLOG("Releasing waitforfinish of (CoWork " << FormatIntHex(job.work) << ")"); @@ -61,7 +72,8 @@ void CoWork::Pool::ThreadRun(int tno) Pool& p = pool(); p.lock.Enter(); for(;;) { - while(p.jobs.GetCount() == 0) { + while(p.scheduled == 0) { + LHITCOUNT("CoWork: Parking thread to Wait"); p.waiting_threads++; p.lock.Leave(); LLOG("#" << tno << " Waiting for job"); @@ -70,6 +82,7 @@ void CoWork::Pool::ThreadRun(int tno) p.lock.Enter(); } LLOG("#" << tno << " Job acquired"); + LHITCOUNT("CoWork: Running new job"); if(DoJob()) break; LLOG("#" << tno << " Job finished"); @@ -78,20 +91,28 @@ void CoWork::Pool::ThreadRun(int tno) LLOG("CoWork thread #" << tno << " finished"); } -void CoWork::Do(Callback cb) { +void CoWork::Do(const Callback *cb, const std::function *fn) +{ + LHITCOUNT("CoWork: Sheduling callback"); #ifdef _MULTITHREADED Pool& p = pool(); p.lock.Enter(); - if(p.jobs.GetCount() > 128) { - LLOG("Stack full: running in the main thread"); + if(p.scheduled >= SCHEDULED_MAX) { + LLOG("Stack full: running in the originating thread"); + LHITCOUNT("CoWork: Stack full: Running in originating thread"); p.lock.Leave(); - cb(); + if(fn) + (*fn)(); + else + (*cb)(); return; } - MJob job; - job.cb = cb; + MJob& job = p.jobs[p.scheduled++]; job.work = this; - p.jobs.Add(job); + if(fn) + job.fn = std::move(*fn); + else + job.cb = *cb; todo++; LLOG("Adding job; todo: " << todo << " (CoWork " << FormatIntHex(this) << ")"); if(p.waiting_threads) { @@ -101,7 +122,10 @@ void CoWork::Do(Callback cb) { } p.lock.Leave(); #else - cb(); + if(fn) + (*fn)(); + else + (*cb)(); #endif } @@ -113,7 +137,7 @@ void CoWork::Finish() { LLOG("Finish: todo: " << todo << " (CoWork " << FormatIntHex(this) << ")"); if(todo == 0) break; - if(p.jobs.GetCount()) + if(p.scheduled) Pool::DoJob(); else { p.lock.Leave(); diff --git a/uppsrc/Core/CoWork.h b/uppsrc/Core/CoWork.h index 0450f381e..944d4aef6 100644 --- a/uppsrc/Core/CoWork.h +++ b/uppsrc/Core/CoWork.h @@ -4,12 +4,16 @@ class CoWork : NoCopy { typedef StaticCriticalSection Lock; struct MJob : Moveable { - Callback cb; - CoWork *work; + std::function fn; + Callback cb; + CoWork *work; }; + + enum { SCHEDULED_MAX = 2048 }; struct Pool { - Vector jobs; + int scheduled; + MJob jobs[SCHEDULED_MAX]; int waiting_threads; Array threads; @@ -30,15 +34,17 @@ class CoWork : NoCopy { Semaphore waitforfinish; int todo; + void Do(const Callback *cb, const std::function *fn); + public: - void Do(Callback cb); + void Do(const Callback& cb) { Do(&cb, NULL); } #ifdef CPP_11 - void Do(std::function fn) { Do(Callback(lambda(fn))); } + void Do(const std::function& fn) { Do(NULL, &fn); } #endif - CoWork& operator&(Callback cb) { Do(cb); return *this; } + CoWork& operator&(const Callback& cb) { Do(&cb, NULL); return *this; } #ifdef CPP_11 - CoWork& operator&(std::function fn) { Do(Callback(lambda(fn))); return *this; } + CoWork& operator&(const std::function& fn) { Do(NULL, &fn); return *this; } #endif void Finish(); diff --git a/uppsrc/Core/Debug.cpp b/uppsrc/Core/Debug.cpp index 31b096852..273f69c21 100644 --- a/uppsrc/Core/Debug.cpp +++ b/uppsrc/Core/Debug.cpp @@ -123,7 +123,6 @@ String TimingInspector::Dump() { HitCountInspector::~HitCountInspector() { - Mutex::Lock __(mutex); RLOG("HITCOUNT " << name << ": hit count = " << hitcount); } diff --git a/uppsrc/Core/Profile.h b/uppsrc/Core/Profile.h index 2a40f79f7..b9b5ba04e 100644 --- a/uppsrc/Core/Profile.h +++ b/uppsrc/Core/Profile.h @@ -42,15 +42,14 @@ public: HitCountInspector(const char *name, int hitcount = 0) : name(name), hitcount(hitcount) {} ~HitCountInspector(); - void Step() { Mutex::Lock __(mutex); hitcount++; } - void Add(int i) { Mutex::Lock __(mutex); hitcount += i; } + void Step() { hitcount++; } + void Add(int i) { hitcount += i; } void operator ++ () { Step(); } void operator += (int i) { Add(i); } private: const char *name; int hitcount; - Mutex mutex; }; #define RTIMING(x) \ diff --git a/uppsrc/Core/src.tpp/Thread$en-us.tpp b/uppsrc/Core/src.tpp/Thread$en-us.tpp index 870c3cbb0..d6d0b830c 100644 --- a/uppsrc/Core/src.tpp/Thread$en-us.tpp +++ b/uppsrc/Core/src.tpp/Thread$en-us.tpp @@ -81,12 +81,16 @@ is running so far).&] [s4; &] [s5;:Thread`:`:ShutdownThreads`(`): [@(0.0.255) static] [@(0.0.255) void]_[* ShutdownThread s]()&] -[s2;%% Sets the `"Shutdown`" flag on.&] +[s2;%% Sets the `"Shutdown`" flag on, waits before all threads terminate, +then sets flag off again. It is meant to be used together with +IsShutdownThreads to terminate long running secondary service +threads. Main thread calls ShutdownThreads, secondary threads +test IsShutdownThreads and if true, exit.&] [s3; &] [s4; &] [s5;:Thread`:`:IsShutdownThreads`(`): [@(0.0.255) static] [@(0.0.255) bool]_[* IsShutdownTh reads]()&] -[s2;%% True if ShutdownThreads was called.&] +[s2;%% True if ShutdownThreads is active.&] [s3; &] [s4; &] [s5;:Thread`:`:AtExit: [@(0.0.255) static]_[@(0.0.255) void]_(`*[* AtExit]([@(0.0.255) void]_