github-starred/ultimatepp
2
0
Fork 0
mirror of https://github.com/ultimatepp/ultimatepp.git synced 2026-07-11 22:03:01 -06:00
Code Issues 33 Projects Packages Wiki Activity Actions

Core: CoWork optimized

Browse source 
git-svn-id: svn://ultimatepp.org/upp/trunk@9374 f0d560ea-af0d-0410-9eb7-867de7ffcac7
This commit is contained in:
cxl 2016-01-08 21:57:22 +00:00
parent 15000b4da7
commit 64de02adab
5 changed files with 63 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

60
uppsrc/Core/CoWork.cpp
View file

@ -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<void ()> 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<void ()> *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();

20
uppsrc/Core/CoWork.h
View file

@ -4,12 +4,16 @@ class CoWork : NoCopy {
typedef StaticCriticalSection Lock;
struct MJob : Moveable<MJob> {
Callback cb;
CoWork *work;
std::function<void ()> fn;
Callback cb;
CoWork *work;
};
enum { SCHEDULED_MAX = 2048 };
struct Pool {
Vector<MJob> jobs;
int scheduled;
MJob jobs[SCHEDULED_MAX];
int waiting_threads;
Array<Thread> threads;
@ -30,15 +34,17 @@ class CoWork : NoCopy {
Semaphore waitforfinish;
int todo;
void Do(const Callback *cb, const std::function<void ()> *fn);
public:
void Do(Callback cb);
void Do(const Callback& cb) { Do(&cb, NULL); }
#ifdef CPP_11
void Do(std::function<void ()> fn) { Do(Callback(lambda(fn))); }
void Do(const std::function<void ()>& 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<void ()> fn) { Do(Callback(lambda(fn))); return *this; }
CoWork& operator&(const std::function<void ()>& fn) { Do(NULL, &fn); return *this; }
#endif
void Finish();

1
uppsrc/Core/Debug.cpp
View file

@ -123,7 +123,6 @@ String TimingInspector::Dump() {
HitCountInspector::~HitCountInspector()
{
Mutex::Lock __(mutex);
RLOG("HITCOUNT " << name << ": hit count = " << hitcount);
}

5
uppsrc/Core/Profile.h
View file

@ -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) \

8
uppsrc/Core/src.tpp/Thread$en-us.tpp
View file

@ -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]_