Loading...
Searching...
No Matches
LibThread::Scheduler Class Reference

## Public Member Functions

| Scheduler (int n)
void | set_maxconcurrency (int n)
int | get_maxconcurrency ()
int | threadpool_size (ThreadPool *pool)
virtual | ~Scheduler ()
ThreadState * | getThread (int i)
void | shutdown (bool wait)
void | addThread (ThreadPool *owner, ThreadState *thread)
void | attachJob (ThreadPool *pool, Job *job)
void | detachJob (Job *job)
void | queueJob (Job *job)
void | broadcastJob (ThreadPool *pool, Job *job)
void | cancelDeps (Job *job)
void | cancelJob (Job *job)
void | waitJob (Job *job)
void | clearThreadState ()
Public Member Functions inherited from LibThread::SharedObject
| SharedObject ()
virtual | ~SharedObject ()
void | set_type (int type_init)
int | get_type ()
void | set_name (std::string &name_init)
void | set_name (const char *s)
std::string & | get_name ()
void | incref (int by=1)
long | decref ()
long | getref ()
virtual BOOLEAN | op2 (int op, leftv res, leftv a1, leftv a2)
virtual BOOLEAN | op3 (int op, leftv res, leftv a1, leftv a2, leftv a3)

## Static Public Member Functions

static void | notifyDeps (Scheduler *scheduler, Job *job)
static void * | main (ThreadState *ts, void *arg)

## Data Fields

Lock | lock

## Private Attributes

bool | single_threaded
size_t | jobid
int | nthreads
int | maxconcurrency
int | running
bool | shutting_down
int | shutdown_counter
vector< ThreadState * > | threads
vector< ThreadPool * > | thread_owners
priority_queue< Job *, vector< Job * >, JobCompare > | global_queue
vector< JobQueue * > | thread_queues
vector< Job * > | pending
ConditionVariable | cond
ConditionVariable | response

## Friends

class | Job

## Detailed Description

Definition at line 1649 of file shared.cc.

## Constructor & Destructor Documentation

## Scheduler()

LibThread::Scheduler::Scheduler | ( | int | n | ) |

inline

Definition at line 1668 of file shared.cc.

:
SharedObject(), threads(), thread_owners(), global_queue(), thread_queues(),
single_threaded(n==0), nthreads(n == 0 ? 1 : n),
lock(true), cond(&lock), response(&lock),
shutting_down(false), shutdown_counter(0), jobid(0),
maxconcurrency(n), running(0)
{
thread_queues.push_back(new JobQueue());
}
LibThread::Scheduler::maxconcurrency
int maxconcurrency
Definition shared.cc:1654
LibThread::Scheduler::thread_owners
vector< ThreadPool * > thread_owners
Definition shared.cc:1659
LibThread::Scheduler::thread_queues
vector< JobQueue * > thread_queues
Definition shared.cc:1661
LibThread::Scheduler::lock
Lock lock
Definition shared.cc:1667
LibThread::Scheduler::response
ConditionVariable response
Definition shared.cc:1664
LibThread::Scheduler::shutting_down
bool shutting_down
Definition shared.cc:1656
LibThread::Scheduler::single_threaded
bool single_threaded
Definition shared.cc:1651
LibThread::Scheduler::global_queue
priority_queue< Job *, vector< Job * >, JobCompare > global_queue
Definition shared.cc:1660
LibThread::Scheduler::cond
ConditionVariable cond
Definition shared.cc:1663
LibThread::Scheduler::threads
vector< ThreadState * > threads
Definition shared.cc:1658
LibThread::Scheduler::nthreads
int nthreads
Definition shared.cc:1653
LibThread::Scheduler::jobid
size_t jobid
Definition shared.cc:1652
LibThread::Scheduler::running
int running
Definition shared.cc:1655
LibThread::Scheduler::shutdown_counter
int shutdown_counter
Definition shared.cc:1657
LibThread::SharedObject::SharedObject
SharedObject()
Definition shared.cc:156
LibThread::JobQueue
queue< Job * > JobQueue
Definition shared.cc:1615

## ~Scheduler()

virtual LibThread::Scheduler::~Scheduler | ( |  | ) |

inline virtual

Definition at line 1691 of file shared.cc.

{
for (unsigned i = 0; i < thread_queues.size(); i++) {
JobQueue *q = thread_queues[i];
while (!q->empty()) {
Job *job = q->front();
q->pop();
releaseShared(job);
}
}
thread_queues.clear();
threads.clear();
}
i
int i
Definition cfEzgcd.cc:132
LibThread::Scheduler::Job
friend class Job
Definition shared.cc:1665
LibThread::releaseShared
void releaseShared(SharedObject *obj)
Definition shared.cc:192

## Member Function Documentation

## addThread()

void LibThread::Scheduler::addThread | ( | ThreadPool * | owner,
|  | ThreadState * | thread )

inline

Definition at line 1730 of file shared.cc.

{
lock.lock();
thread_owners.push_back(owner);
threads.push_back(thread);
thread_queues.push_back(new JobQueue());
lock.unlock();
}

## attachJob()

void LibThread::Scheduler::attachJob | ( | ThreadPool * | pool,
|  | Job * | job )

inline

Definition at line 1737 of file shared.cc.

{
lock.lock();
job->pool = pool;
job->id = jobid++;
acquireShared(job);
if (job->ready()) {
global_queue.push(job);
cond.signal();
}
else if (job->pending_index < 0) {
job->pool = pool;
job->pending_index = pending.size();
pending.push_back(job);
}
lock.unlock();
}
LibThread::Job::id
size_t id
Definition shared.cc:1548
LibThread::Job::pool
ThreadPool * pool
Definition shared.cc:1546
LibThread::Job::pending_index
long pending_index
Definition shared.cc:1549
LibThread::Job::ready
virtual bool ready()
Definition shared.cc:1600
LibThread::Scheduler::pending
vector< Job * > pending
Definition shared.cc:1662
LibThread::acquireShared
void acquireShared(SharedObject *obj)
Definition shared.cc:188

## broadcastJob()

void LibThread::Scheduler::broadcastJob | ( | ThreadPool * | pool,
|  | Job * | job )

inline

Definition at line 1771 of file shared.cc.

{
lock.lock();
for (unsigned i = 0; i <thread_queues.size(); i++) {
if (thread_owners[i] == pool) {
acquireShared(job);
thread_queues[i]->push(job);
}
}
lock.unlock();
}

## cancelDeps()

void LibThread::Scheduler::cancelDeps | ( | Job * | job | ) |

inline

Definition at line 1781 of file shared.cc.

{
vector<Job *> &notify = job->notify;
for (unsigned i = 0; i <notify.size(); i++) {
Job *next = notify[i];
if (!next->cancelled) {
cancelJob(next);
}
}
}
LibThread::Job::notify
vector< Job * > notify
Definition shared.cc:1551
LibThread::Scheduler::cancelJob
void cancelJob(Job *job)
Definition shared.cc:1790
next
ListNode * next
Definition janet.h:31

## cancelJob()

void LibThread::Scheduler::cancelJob | ( | Job * | job | ) |

inline

Definition at line 1790 of file shared.cc.

{
lock.lock();
if (!job->cancelled) {
job->cancelled = true;
if (!job->running && !job->done) {
job->done = true;
cancelDeps(job);
}
}
lock.unlock();
}
LibThread::Job::done
bool done
Definition shared.cc:1557
LibThread::Job::cancelled
bool cancelled
Definition shared.cc:1560
LibThread::Job::running
bool running
Definition shared.cc:1559
LibThread::Scheduler::cancelDeps
void cancelDeps(Job *job)
Definition shared.cc:1781

## clearThreadState()

void LibThread::Scheduler::clearThreadState | ( |  | ) |

inline

Definition at line 1821 of file shared.cc.

{
threads.clear();
}

## detachJob()

void LibThread::Scheduler::detachJob | ( | Job * | job | ) |

inline

Definition at line 1753 of file shared.cc.

{
lock.lock();
long i = job->pending_index;
job->pending_index = -1;
if (i >= 0) {
job = pending.back();
pending.resize(pending.size()-1);
pending[i] = job;
job->pending_index = i;
}
lock.unlock();
}

## get_maxconcurrency()

int LibThread::Scheduler::get_maxconcurrency | ( |  | ) |

inline

Definition at line 1680 of file shared.cc.

{
return maxconcurrency;
}

## getThread()

ThreadState * LibThread::Scheduler::getThread | ( | int | i | ) |

inline

Definition at line 1703 of file shared.cc.

{ return threads[i]; }

## main()

void * LibThread::Scheduler::main | ( | ThreadState * | ts,
|  | void * | arg )

inline static

Definition at line 1851 of file shared.cc.

{
SchedInfo *info = (SchedInfo *) arg;
Scheduler *scheduler = info->scheduler;
ThreadPool *oldThreadPool = currentThreadPoolRef;
// TODO: set current thread pool
// currentThreadPoolRef = pool;
Lock &lock = scheduler->lock;
ConditionVariable &cond = scheduler->cond;
ConditionVariable &response = scheduler->response;
JobQueue *my_queue = scheduler->thread_queues[info->num];
if (!scheduler->single_threaded)
thread_init();
lock.lock();
for (;;) {
if (info->job && info->job->done)
break;
if (scheduler->shutting_down) {
scheduler->shutdown_counter++;
scheduler->response.signal();
break;
}
if (!my_queue->empty()) {
Job *job = my_queue->front();
my_queue->pop();
if (!scheduler->global_queue.empty())
cond.signal();
currentJobRef = job;
job->run();
currentJobRef = NULL;
notifyDeps(scheduler, job);
releaseShared(job);
scheduler->response.signal();
continue;
} else if (!scheduler->global_queue.empty()) {
Job *job = scheduler->global_queue.top();
scheduler->global_queue.pop();
if (!scheduler->global_queue.empty())
cond.signal();
currentJobRef = job;
job->run();
currentJobRef = NULL;
notifyDeps(scheduler, job);
releaseShared(job);
scheduler->response.signal();
continue;
} else {
if (scheduler->single_threaded) {
break;
}
cond.wait();
}
}
// TODO: correct current thread pool
// releaseShared(currentThreadPoolRef);
currentThreadPoolRef = oldThreadPool;
scheduler->lock.unlock();
delete info;
return NULL;
}
ConditionVariable::signal
void signal()
Definition thread.h:97
LibThread::Job::run
void run()
Definition shared.cc:1977
LibThread::Scheduler::Scheduler
Scheduler(int n)
Definition shared.cc:1668
LibThread::Scheduler::notifyDeps
static void notifyDeps(Scheduler *scheduler, Job *job)
Definition shared.cc:1824
Lock::unlock
void unlock()
Definition thread.h:57
info
#define info
Definition libparse.cc:1256
LibThread::currentJobRef
STATIC_VAR Job * currentJobRef
Definition shared.cc:1626
LibThread::currentThreadPoolRef
STATIC_VAR ThreadPool * currentThreadPoolRef
Definition shared.cc:1625
LibThread::thread_init
void thread_init()
Definition shared.cc:1368
NULL
#define NULL
Definition omList.c:12

## notifyDeps()

void LibThread::Scheduler::notifyDeps | ( | Scheduler * | scheduler,
|  | Job * | job )

inline static

Definition at line 1824 of file shared.cc.

{
vector<Job *> &notify = job->notify;
job->incref(notify.size());
for (unsigned i = 0; i <notify.size(); i++) {
Job *next = notify[i];
if (!next->queued && next->ready() && !next->cancelled) {
next->queued = true;
scheduler->queueJob(next);
}
}
vector<Trigger *> &triggers = job->triggers;
leftv arg = NULL;
if (triggers.size() > 0 && job->result.size() > 0)
arg = LinTree::from_string(job->result);
for (unsigned i = 0; i < triggers.size(); i++) {
Trigger *trigger = triggers[i];
if (trigger->accept(arg)) {
trigger->activate(arg);
if (trigger->ready())
scheduler->queueJob(trigger);
}
}
if (arg) {
arg->CleanUp();
omFreeBin(arg, sleftv_bin);
}
}
LibThread::Job::result
string result
Definition shared.cc:1554
LibThread::Job::triggers
vector< Trigger * > triggers
Definition shared.cc:1552
LibThread::Scheduler::queueJob
void queueJob(Job *job)
Definition shared.cc:1765
LibThread::SharedObject::incref
void incref(int by=1)
Definition shared.cc:165
LibThread::Trigger::activate
virtual void activate(leftv arg)=0
LibThread::Trigger::accept
virtual bool accept(leftv arg)=0
sleftv::CleanUp
void CleanUp(ring r=currRing)
Definition subexpr.cc:351
sleftv_bin
EXTERN_VAR omBin sleftv_bin
Definition ipid.h:145
LinTree::from_string
leftv from_string(std::string &str)
Definition lintree.cc:854
omFreeBin
#define omFreeBin(addr, bin)
Definition omAllocDecl.h:259
leftv
sleftv * leftv
Definition structs.h:53

## queueJob()

void LibThread::Scheduler::queueJob | ( | Job * | job | ) |

inline

Definition at line 1765 of file shared.cc.

{
lock.lock();
global_queue.push(job);
cond.signal();
lock.unlock();
}

## set_maxconcurrency()

void LibThread::Scheduler::set_maxconcurrency | ( | int | n | ) |

inline

Definition at line 1677 of file shared.cc.

{
maxconcurrency = n;
}

## shutdown()

void LibThread::Scheduler::shutdown | ( | bool | wait | ) |

inline

Definition at line 1704 of file shared.cc.

{
if (single_threaded) {
SchedInfo *info = new SchedInfo();
info->num = 0;
info->scheduler = this;
acquireShared(this);
info->job = NULL;
Scheduler::main(NULL, info);
return;
}
lock.lock();
if (wait) {
while (!global_queue.empty()) {
response.wait();
}
}
shutting_down = true;
while (shutdown_counter < nthreads) {
cond.broadcast();
response.wait();
}
lock.unlock();
for (unsigned i = 0; i <threads.size(); i++) {
joinThread(threads[i]);
}
}
LibThread::Scheduler::main
static void * main(ThreadState *ts, void *arg)
Definition shared.cc:1851
LibThread::joinThread
void * joinThread(ThreadState *ts)
Definition shared.cc:1469
wait
wait
Definition si_signals.h:61

## threadpool_size()

int LibThread::Scheduler::threadpool_size | ( | ThreadPool * | pool | ) |

inline

Definition at line 1683 of file shared.cc.

{
int n;
for (unsigned i = 0; i <thread_owners.size(); i++) {
if (thread_owners[i] == pool)
n++;
}
return n;
}

## waitJob()

void LibThread::Scheduler::waitJob | ( | Job * | job | ) |

inline

Definition at line 1801 of file shared.cc.

{
if (single_threaded) {
SchedInfo *info = new SchedInfo();
info->num = 0;
info->scheduler = this;
acquireShared(this);
info->job = job;
Scheduler::main(NULL, info);
} else {
lock.lock();
for (;;) {
if (job->done || job->cancelled) {
break;
}
response.wait();
}
response.signal(); // forward signal
lock.unlock();
}
}

## Friends And Related Symbol Documentation

## Job

friend class Job

friend

Definition at line 1665 of file shared.cc.

## Field Documentation

## cond

ConditionVariable LibThread::Scheduler::cond

private

Definition at line 1663 of file shared.cc.

## global_queue

priority_queue<Job *, vector<Job *>, JobCompare> LibThread::Scheduler::global_queue

private

Definition at line 1660 of file shared.cc.

## jobid

size_t LibThread::Scheduler::jobid

private

Definition at line 1652 of file shared.cc.

## lock

Lock LibThread::Scheduler::lock

Definition at line 1667 of file shared.cc.

## maxconcurrency

int LibThread::Scheduler::maxconcurrency

private

Definition at line 1654 of file shared.cc.

## nthreads

int LibThread::Scheduler::nthreads

private

Definition at line 1653 of file shared.cc.

## pending

vector<Job *> LibThread::Scheduler::pending

private

Definition at line 1662 of file shared.cc.

## response

ConditionVariable LibThread::Scheduler::response

private

Definition at line 1664 of file shared.cc.

## running

int LibThread::Scheduler::running

private

Definition at line 1655 of file shared.cc.

## shutdown_counter

int LibThread::Scheduler::shutdown_counter

private

Definition at line 1657 of file shared.cc.

## shutting_down

bool LibThread::Scheduler::shutting_down

private

Definition at line 1656 of file shared.cc.

## single_threaded

bool LibThread::Scheduler::single_threaded

private

Definition at line 1651 of file shared.cc.

## thread_owners

vector<ThreadPool *> LibThread::Scheduler::thread_owners

private

Definition at line 1659 of file shared.cc.

## thread_queues

vector<JobQueue *> LibThread::Scheduler::thread_queues

private

Definition at line 1661 of file shared.cc.

## threads

vector<ThreadState *> LibThread::Scheduler::threads

private

Definition at line 1658 of file shared.cc.

The documentation for this class was generated from the following file:

- Singular/dyn_modules/systhreads/shared.cc