Loading...
Searching...
No Matches
shared.cc
Go to the documentation of this file.
#include "threadconf.h"
#include <iostream>
#include "kernel/mod2.h"
#include "Singular/feOpt.h"
#include "Singular/libsingular.h"
#include <cstring>
#include <string>
#include <errno.h>
#include <stdio.h>
#include <vector>
#include <map>
#include <iterator>
#include <queue>
#include <assert.h>
#include "thread.h"
#include "lintree.h"
#include "singthreads.h"
using namespace std;
#ifdef ENABLE_THREADS
extern char *global_argv0;
#endif
extern "C" void pSingular_initialize_thread();
namespace LibThread {
#ifdef ENABLE_THREADS
const int have_threads = 1;
#else
const int have_threads = 0;
#endif
class Command {
private:
const char *name;
const char *error;
leftv result;
leftv *args;
int argc;
public:
Command(const char *n, leftv r, leftv a)
{
name = n;
result = r;
error = NULL;
argc = 0;
for (leftv t = a; t != NULL; t = t->next) {
argc++;
}
args = (leftv *) omAlloc0(sizeof(leftv) * argc);
int i = 0;
for (leftv t = a; t != NULL; t = t->next) {
args[i++] = t;
}
result->rtyp = NONE;
result->data = NULL;
}
~Command() {
omFree(args);
}
void check_argc(int n) {
if (error) return;
if (argc != n) error = "wrong number of arguments";
}
void check_argc(int lo, int hi) {
if (error) return;
if (argc < lo || argc > hi) error = "wrong number of arguments";
}
void check_argc_min(int n) {
if (error) return;
if (argc < n) error = "wrong number of arguments";
}
void check_arg(int i, int type, const char *err) {
if (error) return;
if (args[i]->Typ() != type) error = err;
}
void check_init(int i, const char *err) {
if (error) return;
leftv arg = args[i];
if (arg->Data() == NULL || *(void **)(arg->Data()) == NULL)
error = err;
}
void check_arg(int i, int type, int type2, const char *err) {
if (error) return;
if (args[i]->Typ() != type && args[i]->Typ() != type2) error = err;
}
int argtype(int i) {
return args[i]->Typ();
}
int nargs() {
return argc;
}
void *arg(int i) {
return args[i]->Data();
}
template <typename T>
T *shared_arg(int i) {
return *(T **)(arg(i));
}
long int_arg(int i) {
return (long)(args[i]->Data());
}
void report(const char *err) {
error = err;
}
// intentionally not bool, so we can also do
// q = p + test_arg(p, type);
int test_arg(int i, int type) {
if (i >= argc) return 0;
return args[i]->Typ() == type;
}
void set_result(long n) {
result->rtyp = INT_CMD;
result->data = (char *)n;
}
void set_result(const char *s) {
result->rtyp = STRING_CMD;
result->data = omStrDup(s);
}
void set_result(int type, void *p) {
result->rtyp = type;
result->data = (char *) p;
}
void set_result(int type, long n) {
result->rtyp = type;
result->data = (char *) n;
}
void no_result() {
result->rtyp = NONE;
}
bool ok() {
return error == NULL;
}
BOOLEAN status() {
if (error) {
Werror("%s: %s", name, error);
}
return error != NULL;
}
BOOLEAN abort(const char *err) {
report(err);
return status();
}
};
class SharedObject {
private:
Lock lock;
long refcount;
int type;
std::string name;
public:
SharedObject(): lock(), refcount(0) { }
virtual ~SharedObject() { }
void set_type(int type_init) { type = type_init; }
int get_type() { return type; }
void set_name(std::string &name_init) { name = name_init; }
void set_name(const char *s) {
name = std::string(s);
}
std::string &get_name() { return name; }
void incref(int by = 1) {
lock.lock();
refcount += 1;
lock.unlock();
}
long decref() {
int result;
lock.lock();
result = --refcount;
lock.unlock();
return result;
}
long getref() {
return refcount;
}
virtual BOOLEAN op2(int op, leftv res, leftv a1, leftv a2) {
return TRUE;
}
virtual BOOLEAN op3(int op, leftv res, leftv a1, leftv a2, leftv a3) {
return TRUE;
}
};
void acquireShared(SharedObject *obj) {
obj->incref();
}
void releaseShared(SharedObject *obj) {
if (obj->decref() == 0) {
// delete obj;
}
}
typedef std::map<std::string, SharedObject *> SharedObjectTable;
class Region : public SharedObject {
private:
Lock region_lock;
public:
SharedObjectTable objects;
Region() : SharedObject(), region_lock(), objects() { }
virtual ~Region() { }
Lock *get_lock() { return &region_lock; }
void lock() {
if (!region_lock.is_locked())
region_lock.lock();
}
void unlock() {
if (region_lock.is_locked())
region_lock.unlock();
}
int is_locked() {
return region_lock.is_locked();
}
};
Lock global_objects_lock;
SharedObjectTable global_objects;
Lock master_lock(true);
Lock name_lock(true);
VAR long thread_id;
long thread_counter;
int type_region;
int type_regionlock;
int type_channel;
int type_syncvar;
int type_atomic_table;
int type_shared_table;
int type_atomic_list;
int type_shared_list;
int type_thread;
int type_threadpool;
int type_job;
int type_trigger;
typedef SharedObject *SharedObjectPtr;
typedef SharedObjectPtr (*SharedConstructor)();
SharedObject *makeSharedObject(SharedObjectTable &table,
Lock *lock, int type, string &name, SharedConstructor scons)
{
int was_locked = lock->is_locked();
SharedObject *result = NULL;
if (!was_locked)
lock->lock();
if (table.count(name)) {
result = table[name];
if (result->get_type() != type)
result = NULL;
} else {
result = scons();
result->set_type(type);
result->set_name(name);
table.insert(pair<string,SharedObject *>(name, result));
}
if (!was_locked)
lock->unlock();
return result;
}
SharedObject *findSharedObject(SharedObjectTable &table,
Lock *lock, string &name)
{
int was_locked = lock->is_locked();
SharedObject *result = NULL;
if (!was_locked)
lock->lock();
if (table.count(name)) {
result = table[name];
}
if (!was_locked)
lock->unlock();
return result;
}
class Transactional: public SharedObject {
private:
Region *region;
Lock *lock;
protected:
int tx_begin() {
if (!region)
lock->lock();
else {
if (!lock->is_locked()) {
return 0;
}
}
return 1;
}
void tx_end() {
if (!region)
lock->unlock();
}
public:
Transactional() :
SharedObject(), region(NULL), lock(NULL) {
}
void set_region(Region *region_init) {
region = region_init;
if (region_init) {
lock = region_init->get_lock();
} else {
lock = new Lock();
}
}
virtual ~Transactional() { if (!region && lock) delete lock; }
};
class TxTable: public Transactional {
private:
std::map<string, string> entries;
public:
TxTable() : Transactional(), entries() { }
virtual ~TxTable() { }
int put(string &key, string &value) {
int result = 0;
if (!tx_begin()) return -1;
if (entries.count(key)) {
entries[key] = value;
} else {
entries.insert(pair<string, string>(key, value));
result = 1;
}
tx_end();
return result;
}
int get(string &key, string &value) {
int result = 0;
if (!tx_begin()) return -1;
if (entries.count(key)) {
value = entries[key];
result = 1;
}
tx_end();
return result;
}
int check(string &key) {
int result;
if (!tx_begin()) return -1;
result = entries.count(key);
tx_end();
return result;
}
};
class TxList: public Transactional {
private:
vector<string> entries;
public:
TxList() : Transactional(), entries() { }
virtual ~TxList() { }
int put(size_t index, string &value) {
int result = -1;
if (!tx_begin()) return -1;
if (index >= 1 && index <= entries.size()) {
entries[index-1] = value;
result = 1;
} else {
entries.resize(index+1);
entries[index-1] = value;
result = 0;
}
tx_end();
return result;
}
int get(size_t index, string &value) {
int result = 0;
if (!tx_begin()) return -1;
if (index >= 1 && index <= entries.size()) {
result = (entries[index-1].size() != 0);
if (result)
value = entries[index-1];
}
tx_end();
return result;
}
long size() {
long result;
if (!tx_begin()) return -1;
result = (long) entries.size();
tx_end();
return result;
}
};
class SingularChannel : public SharedObject {
private:
queue<string> q;
Lock lock;
ConditionVariable cond;
public:
SingularChannel(): SharedObject(), lock(), cond(&lock) { }
virtual ~SingularChannel() { }
void send(string item) {
lock.lock();
q.push(item);
cond.signal();
lock.unlock();
}
string receive() {
lock.lock();
while (q.empty()) {
cond.wait();
}
string result = q.front();
q.pop();
if (!q.empty())
cond.signal();
lock.unlock();
return result;
}
long count() {
lock.lock();
long result = q.size();
lock.unlock();
return result;
}
};
class SingularSyncVar : public SharedObject {
private:
string value;
int init;
Lock lock;
ConditionVariable cond;
public:
SingularSyncVar(): SharedObject(), init(0), lock(), cond(&lock) { }
virtual ~SingularSyncVar() { }
void acquire() {
lock.lock();
}
void release() {
lock.unlock();
}
void wait_init() {
while (!init)
cond.wait();
}
leftv get() {
if (value.size() == 0) return NULL;
return LinTree::from_string(value);
}
void update(leftv val) {
value = LinTree::to_string(val);
init = 1;
cond.broadcast();
}
int write(string item) {
int result = 0;
lock.lock();
if (!init) {
value = item;
init = 1;
cond.broadcast();
result = 1;
}
lock.unlock();
return result;
}
string read() {
lock.lock();
while (!init)
cond.wait();
string result = value;
lock.unlock();
return result;
}
int check() {
lock.lock();
int result = init;
lock.unlock();
return result;
}
};
void *shared_init(blackbox *b) {
return omAlloc0(sizeof(SharedObject *));
}
void *new_shared(SharedObject *obj) {
acquireShared(obj);
void *result = omAlloc0(sizeof(SharedObject *));
*(SharedObject **)result = obj;
return result;
}
void shared_destroy(blackbox *b, void *d) {
SharedObject *obj = *(SharedObject **)d;
if (obj) {
releaseShared(*(SharedObject **)d);
*(SharedObject **)d = NULL;
}
}
void rlock_destroy(blackbox *b, void *d) {
SharedObject *obj = *(SharedObject **)d;
((Region *) obj)->unlock();
if (obj) {
releaseShared(*(SharedObject **)d);
*(SharedObject **)d = NULL;
}
}
void *shared_copy(blackbox *b, void *d) {
SharedObject *obj = *(SharedObject **)d;
void *result = shared_init(b);
*(SharedObject **)result = obj;
if (obj)
acquireShared(obj);
return result;
}
BOOLEAN shared_assign(leftv l, leftv r) {
if (r->Typ() == l->Typ()) {
if (l->rtyp == IDHDL) {
omFree(IDDATA((idhdl)l->data));
IDDATA((idhdl)l->data) = (char*)shared_copy(NULL,r->Data());
} else {
leftv ll=l->LData();
if (ll==NULL)
{
return TRUE; // out of array bounds or similiar
}
if (ll->data) {
shared_destroy(NULL, ll->data);
omFree(ll->data);
}
ll->data = shared_copy(NULL,r->Data());
}
} else {
Werror("assign %s(%d) = %s(%d)",
Tok2Cmdname(l->Typ()),l->Typ(),Tok2Cmdname(r->Typ()),r->Typ());
return TRUE;
}
return FALSE;
}
BOOLEAN rlock_assign(leftv l, leftv r) {
if (r->Typ() == l->Typ()) {
if (l->rtyp == IDHDL) {
omFree(IDDATA((idhdl)l->data));
IDDATA((idhdl)l->data) = (char*)shared_copy(NULL,r->Data());
} else {
leftv ll=l->LData();
if (ll==NULL)
{
return TRUE; // out of array bounds or similiar
}
rlock_destroy(NULL, ll->data);
omFree(ll->data);
ll->data = shared_copy(NULL,r->Data());
}
} else {
Werror("assign %s(%d) = %s(%d)",
Tok2Cmdname(l->Typ()),l->Typ(),Tok2Cmdname(r->Typ()),r->Typ());
return TRUE;
}
return FALSE;
}
BOOLEAN shared_check_assign(blackbox *b, leftv l, leftv r) {
int lt = l->Typ();
int rt = r->Typ();
if (lt != DEF_CMD && lt != rt) {
const char *rn=Tok2Cmdname(rt);
const char *ln=Tok2Cmdname(lt);
Werror("cannot assign %s (%d) to %s (%d)\n", rn, rt, ln, lt);
return TRUE;
}
return FALSE;
}
BOOLEAN shared_op2(int op, leftv res, leftv a1, leftv a2) {
SharedObject *obj = *(SharedObject **)a1->Data();
return obj->op2(op, res, a1, a2);
}
BOOLEAN shared_op3(int op, leftv res, leftv a1, leftv a2, leftv a3) {
SharedObject *obj = *(SharedObject **)a1->Data();
return obj->op3(op, res, a1, a2, a3);
}
char *shared_string(blackbox *b, void *d) {
char buf[80];
SharedObject *obj = *(SharedObject **)d;
if (!obj)
return omStrDup("<uninitialized shared object>");
int type = obj->get_type();
string &name = obj->get_name();
const char *type_name = "unknown";
if (type == type_channel)
type_name = "channel";
else if (type == type_atomic_table)
type_name = "atomic_table";
else if (type == type_shared_table)
type_name = "shared_table";
else if (type == type_atomic_list)
type_name = "atomic_list";
else if (type == type_shared_list)
type_name = "shared_list";
else if (type == type_syncvar)
type_name = "syncvar";
else if (type == type_region)
type_name = "region";
else if (type == type_regionlock)
type_name = "regionlock";
else if (type == type_thread) {
snprintf(buf,80, "<thread #%s>", name.c_str());
return omStrDup(buf);
}
else if (type == type_threadpool) {
if (name.size() > 0) {
name_lock.lock();
snprintf(buf,80, "<threadpool \"%.40s\"@%p>", name.c_str(), obj);
name_lock.unlock();
} else
snprintf(buf,80, "<threadpool @%p>", obj);
return omStrDup(buf);
}
else if (type == type_job) {
if (name.size() > 0) {
name_lock.lock();
snprintf(buf,80, "<job \"%.40s\"@%p>", name.c_str(), obj);
name_lock.unlock();
} else
snprintf(buf,80, "<job @%p>", obj);
return omStrDup(buf);
}
else if (type == type_trigger) {
if (name.size() > 0) {
name_lock.lock();
snprintf(buf,80, "<trigger \"%.40s\"@%p>", name.c_str(), obj);
name_lock.unlock();
} else
snprintf(buf,80, "<trigger @%p>", obj);
return omStrDup(buf);
} else {
snprintf(buf,80, "<unknown type %d>", type);
return omStrDup(buf);
}
snprintf(buf,80, "<%s \"%.40s\">", type_name, name.c_str());
return omStrDup(buf);
}
char *rlock_string(blackbox *b, void *d) {
char buf[80];
SharedObject *obj = *(SharedObject **)d;
if (!obj)
return omStrDup("<uninitialized region lock>");
snprintf(buf,80, "<region lock \"%.40s\">", obj->get_name().c_str());
return omStrDup(buf);
}
void report(const char *fmt, const char *name) {
char buf[80];
snprintf(buf,80, fmt, name);
WerrorS(buf);
}
int wrong_num_args(const char *name, leftv arg, int n) {
for (int i=1; i<=n; i++) {
if (!arg) {
report("%s: too few arguments", name);
return TRUE;
}
arg = arg->next;
}
if (arg) {
report("%s: too many arguments", name);
return TRUE;
}
return FALSE;
}
int not_a_uri(const char *name, leftv arg) {
if (arg->Typ() != STRING_CMD) {
report("%s: not a valid URI", name);
return TRUE;
}
return FALSE;
}
int not_a_region(const char *name, leftv arg) {
if (arg->Typ() != type_region || !arg->Data()) {
report("%s: not a region", name);
return TRUE;
}
return FALSE;
}
char *str(leftv arg) {
return (char *)(arg->Data());
}
SharedObject *consTable() {
return new TxTable();
}
SharedObject *consList() {
return new TxList();
}
SharedObject *consChannel() {
return new SingularChannel();
}
SharedObject *consSyncVar() {
return new SingularSyncVar();
}
SharedObject *consRegion() {
return new Region();
}
static void appendArg(vector<leftv> &argv, string &s) {
if (s.size() == 0) return;
leftv val = LinTree::from_string(s);
if (val->Typ() == NONE) {
omFreeBin(val, sleftv_bin);
return;
}
argv.push_back(val);
}
static void appendArg(vector<leftv> &argv, leftv arg) {
argv.push_back(arg);
}
static void appendArgCopy(vector<leftv> &argv, leftv arg) {
leftv val = (leftv) omAlloc0Bin(sleftv_bin);
val->Copy(arg);
argv.push_back(val);
}
static BOOLEAN executeProc(sleftv &result,
const char *procname, const vector<leftv> &argv)
{
leftv procnode = (leftv) omAlloc0Bin(sleftv_bin);
procnode->name = omStrDup(procname);
procnode->req_packhdl = basePack;
int error = procnode->Eval();
if (error) {
Werror("procedure \"%s\" not found", procname);
omFreeBin(procnode, sleftv_bin);
return TRUE;
}
memset(&result, 0, sizeof(result));
leftv *tail = &procnode->next;
for (unsigned i = 0; i < argv.size(); i++) {
*tail = argv[i];
tail = &(*tail)->next;
}
*tail = NULL;
error = iiExprArithM(&result, procnode, '(');
procnode->CleanUp();
omFreeBin(procnode, sleftv_bin);
if (error) {
Werror("procedure call of \"%s\" failed", procname);
return TRUE;
}
return FALSE;
}
BOOLEAN makeAtomicTable(leftv result, leftv arg) {
if (wrong_num_args("makeAtomicTable", arg, 1))
return TRUE;
if (not_a_uri("makeAtomicTable", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = makeSharedObject(global_objects,
&global_objects_lock, type_atomic_table, uri, consTable);
((TxTable *) obj)->set_region(NULL);
result->rtyp = type_atomic_table;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN makeAtomicList(leftv result, leftv arg) {
if (wrong_num_args("makeAtomicList", arg, 1))
return TRUE;
if (not_a_uri("makeAtomicList", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = makeSharedObject(global_objects,
&global_objects_lock, type_atomic_list, uri, consList);
((TxList *) obj)->set_region(NULL);
result->rtyp = type_atomic_list;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN makeSharedTable(leftv result, leftv arg) {
if (wrong_num_args("makeSharedTable", arg, 2))
return TRUE;
if (not_a_region("makeSharedTable", arg))
return TRUE;
if (not_a_uri("makeSharedTable", arg->next))
return TRUE;
Region *region = *(Region **) arg->Data();
fflush(stdout);
string s = str(arg->next);
SharedObject *obj = makeSharedObject(region->objects,
region->get_lock(), type_shared_table, s, consTable);
((TxTable *) obj)->set_region(region);
result->rtyp = type_shared_table;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN makeSharedList(leftv result, leftv arg) {
if (wrong_num_args("makeSharedList", arg, 2))
return TRUE;
if (not_a_region("makeSharedList", arg))
return TRUE;
if (not_a_uri("makeSharedList", arg->next))
return TRUE;
Region *region = *(Region **) arg->Data();
string s = str(arg->next);
SharedObject *obj = makeSharedObject(region->objects,
region->get_lock(), type_shared_list, s, consList);
((TxList *) obj)->set_region(region);
result->rtyp = type_shared_list;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN makeChannel(leftv result, leftv arg) {
if (wrong_num_args("makeChannel", arg, 1))
return TRUE;
if (not_a_uri("makeChannel", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = makeSharedObject(global_objects,
&global_objects_lock, type_channel, uri, consChannel);
result->rtyp = type_channel;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN makeSyncVar(leftv result, leftv arg) {
if (wrong_num_args("makeSyncVar", arg, 1))
return TRUE;
if (not_a_uri("makeSyncVar", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = makeSharedObject(global_objects,
&global_objects_lock, type_syncvar, uri, consSyncVar);
result->rtyp = type_syncvar;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN makeRegion(leftv result, leftv arg) {
if (wrong_num_args("makeRegion", arg, 1))
return TRUE;
if (not_a_uri("makeRegion", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = makeSharedObject(global_objects,
&global_objects_lock, type_region, uri, consRegion);
result->rtyp = type_region;
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN findSharedObject(leftv result, leftv arg) {
if (wrong_num_args("findSharedObject", arg, 1))
return TRUE;
if (not_a_uri("findSharedObject", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = findSharedObject(global_objects,
&global_objects_lock, uri);
result->rtyp = INT_CMD;
result->data = (char *)(long)(obj != NULL);
return FALSE;
}
BOOLEAN typeSharedObject(leftv result, leftv arg) {
if (wrong_num_args("findSharedObject", arg, 1))
return TRUE;
if (not_a_uri("findSharedObject", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = findSharedObject(global_objects,
&global_objects_lock, uri);
int type = obj ? obj->get_type() : -1;
const char *type_name = "undefined";
if (type == type_channel)
type_name = "channel";
else if (type == type_atomic_table)
type_name = "atomic_table";
else if (type == type_shared_table)
type_name = "shared_table";
else if (type == type_atomic_list)
type_name = "atomic_list";
else if (type == type_shared_list)
type_name = "shared_list";
else if (type == type_syncvar)
type_name = "syncvar";
else if (type == type_region)
type_name = "region";
else if (type == type_regionlock)
type_name = "regionlock";
result->rtyp = STRING_CMD;
result->data = (char *)(omStrDup(type_name));
return FALSE;
}
BOOLEAN bindSharedObject(leftv result, leftv arg) {
if (wrong_num_args("bindSharedObject", arg, 1))
return TRUE;
if (not_a_uri("bindSharedObject", arg))
return TRUE;
string uri = str(arg);
SharedObject *obj = findSharedObject(global_objects,
&global_objects_lock, uri);
if (!obj) {
WerrorS("bindSharedObject: cannot find object");
return TRUE;
}
result->rtyp = obj->get_type();
result->data = new_shared(obj);
return FALSE;
}
BOOLEAN getTable(leftv result, leftv arg) {
if (wrong_num_args("getTable", arg, 2))
return TRUE;
if (arg->Typ() != type_atomic_table && arg->Typ() != type_shared_table) {
WerrorS("getTable: not a valid table");
return TRUE;
}
if (arg->next->Typ() != STRING_CMD) {
WerrorS("getTable: not a valid table key");
return TRUE;
}
TxTable *table = *(TxTable **) arg->Data();
if (!table) {
WerrorS("getTable: table has not been initialized");
return TRUE;
}
string key = (char *)(arg->next->Data());
string value;
int success = table->get(key, value);
if (success < 0) {
WerrorS("getTable: region not acquired");
return TRUE;
}
if (success == 0) {
WerrorS("getTable: key not found");
return TRUE;
}
leftv tmp = LinTree::from_string(value);
result->rtyp = tmp->Typ();
result->data = tmp->Data();
return FALSE;
}
BOOLEAN inTable(leftv result, leftv arg) {
if (wrong_num_args("inTable", arg, 2))
return TRUE;
if (arg->Typ() != type_atomic_table && arg->Typ() != type_shared_table) {
WerrorS("inTable: not a valid table");
return TRUE;
}
if (arg->next->Typ() != STRING_CMD) {
WerrorS("inTable: not a valid table key");
return TRUE;
}
TxTable *table = *(TxTable **) arg->Data();
if (!table) {
WerrorS("inTable: table has not been initialized");
return TRUE;
}
string key = (char *)(arg->next->Data());
int success = table->check(key);
if (success < 0) {
WerrorS("inTable: region not acquired");
return TRUE;
}
result->rtyp = INT_CMD;
result->data = (char *)(long)(success);
return FALSE;
}
BOOLEAN putTable(leftv result, leftv arg) {
if (wrong_num_args("putTable", arg, 3))
return TRUE;
if (arg->Typ() != type_atomic_table && arg->Typ() != type_shared_table) {
WerrorS("putTable: not a valid table");
return TRUE;
}
if (arg->next->Typ() != STRING_CMD) {
WerrorS("putTable: not a valid table key");
return TRUE;
}
TxTable *table = *(TxTable **) arg->Data();
if (!table) {
WerrorS("putTable: table has not been initialized");
return TRUE;
}
string key = (char *)(arg->next->Data());
string value = LinTree::to_string(arg->next->next);
int success = table->put(key, value);
if (success < 0) {
WerrorS("putTable: region not acquired");
return TRUE;
}
result->rtyp = NONE;
return FALSE;
}
BOOLEAN getList(leftv result, leftv arg) {
if (wrong_num_args("getList", arg, 2))
return TRUE;
if (arg->Typ() != type_atomic_list && arg->Typ() != type_shared_list) {
WerrorS("getList: not a valid list (atomic or shared)");
return TRUE;
}
if (arg->next->Typ() != INT_CMD) {
WerrorS("getList: index must be an integer");
return TRUE;
}
TxList *list = *(TxList **) arg->Data();
if (!list) {
WerrorS("getList: list has not been initialized");
return TRUE;
}
long index = (long)(arg->next->Data());
string value;
int success = list->get(index, value);
if (success < 0) {
WerrorS("getList: region not acquired");
return TRUE;
}
if (success == 0) {
WerrorS("getList: no value at position");
return TRUE;
}
leftv tmp = LinTree::from_string(value);
result->rtyp = tmp->Typ();
result->data = tmp->Data();
return FALSE;
}
BOOLEAN putList(leftv result, leftv arg) {
if (wrong_num_args("putList", arg, 3))
return TRUE;
if (arg->Typ() != type_atomic_list && arg->Typ() != type_shared_list) {
WerrorS("putList: not a valid list (shared or atomic)");
return TRUE;
}
if (arg->next->Typ() != INT_CMD) {
WerrorS("putList: index must be an integer");
return TRUE;
}
TxList *list = *(TxList **) arg->Data();
if (!list) {
WerrorS("putList: list has not been initialized");
return TRUE;
}
long index = (long)(arg->next->Data());
string value = LinTree::to_string(arg->next->next);
int success = list->put(index, value);
if (success < 0) {
WerrorS("putList: region not acquired");
return TRUE;
}
result->rtyp = NONE;
return FALSE;
}
BOOLEAN lockRegion(leftv result, leftv arg) {
if (wrong_num_args("lockRegion", arg, 1))
return TRUE;
if (not_a_region("lockRegion", arg))
return TRUE;
Region *region = *(Region **)arg->Data();
if (region->is_locked()) {
WerrorS("lockRegion: region is already locked");
return TRUE;
}
region->lock();
result->rtyp = NONE;
return FALSE;
}
BOOLEAN regionLock(leftv result, leftv arg) {
if (wrong_num_args("lockRegion", arg, 1))
return TRUE;
if (not_a_region("lockRegion", arg))
return TRUE;
Region *region = *(Region **)arg->Data();
if (region->is_locked()) {
WerrorS("lockRegion: region is already locked");
return TRUE;
}
region->lock();
result->rtyp = type_regionlock;
result->data = new_shared(region);
return FALSE;
}
BOOLEAN unlockRegion(leftv result, leftv arg) {
if (wrong_num_args("unlockRegion", arg, 1))
return TRUE;
if (not_a_region("unlockRegion", arg))
return TRUE;
Region *region = *(Region **)arg->Data();
if (!region->is_locked()) {
WerrorS("unlockRegion: region is not locked");
return TRUE;
}
region->unlock();
result->rtyp = NONE;
return FALSE;
}
BOOLEAN sendChannel(leftv result, leftv arg) {
if (wrong_num_args("sendChannel", arg, 2))
return TRUE;
if (arg->Typ() != type_channel) {
WerrorS("sendChannel: argument is not a channel");
return TRUE;
}
SingularChannel *channel = *(SingularChannel **)arg->Data();
if (!channel) {
WerrorS("sendChannel: channel has not been initialized");
return TRUE;
}
channel->send(LinTree::to_string(arg->next));
result->rtyp = NONE;
return FALSE;
}
BOOLEAN receiveChannel(leftv result, leftv arg) {
if (wrong_num_args("receiveChannel", arg, 1))
return TRUE;
if (arg->Typ() != type_channel) {
WerrorS("receiveChannel: argument is not a channel");
return TRUE;
}
SingularChannel *channel = *(SingularChannel **)arg->Data();
if (!channel) {
WerrorS("receiveChannel: channel has not been initialized");
return TRUE;
}
string item = channel->receive();
leftv val = LinTree::from_string(item);
result->rtyp = val->Typ();
result->data = val->Data();
return FALSE;
}
BOOLEAN statChannel(leftv result, leftv arg) {
if (wrong_num_args("statChannel", arg, 1))
return TRUE;
if (arg->Typ() != type_channel) {
WerrorS("statChannel: argument is not a channel");
return TRUE;
}
SingularChannel *channel = *(SingularChannel **)arg->Data();
if (!channel) {
WerrorS("receiveChannel: channel has not been initialized");
return TRUE;
}
long n = channel->count();
result->rtyp = INT_CMD;
result->data = (char *)n;
return FALSE;
}
BOOLEAN writeSyncVar(leftv result, leftv arg) {
if (wrong_num_args("writeSyncVar", arg, 2))
return TRUE;
if (arg->Typ() != type_syncvar) {
WerrorS("writeSyncVar: argument is not a syncvar");
return TRUE;
}
SingularSyncVar *syncvar = *(SingularSyncVar **)arg->Data();
if (!syncvar) {
WerrorS("writeSyncVar: syncvar has not been initialized");
return TRUE;
}
if (!syncvar->write(LinTree::to_string(arg->next))) {
WerrorS("writeSyncVar: variable already has a value");
return TRUE;
}
result->rtyp = NONE;
return FALSE;
}
BOOLEAN updateSyncVar(leftv result, leftv arg) {
Command cmd("updateSyncVar", result, arg);
cmd.check_argc_min(2);
cmd.check_arg(0, type_syncvar, "first argument must be a syncvar");
cmd.check_init(0, "syncvar has not been initialized");
cmd.check_arg(1, STRING_CMD, "second argument must be a string");
if (cmd.ok()) {
SingularSyncVar *syncvar = cmd.shared_arg<SingularSyncVar>(0);
char *procname = (char *) cmd.arg(1);
arg = arg->next->next;
syncvar->acquire();
syncvar->wait_init();
vector<leftv> argv;
appendArg(argv, syncvar->get());
while (arg) {
appendArgCopy(argv, arg);
arg = arg->next;
}
int error = executeProc(*result, procname, argv);
if (!error) {
syncvar->update(result);
}
syncvar->release();
return error;
}
return cmd.status();
}
BOOLEAN readSyncVar(leftv result, leftv arg) {
if (wrong_num_args("readSyncVar", arg, 1))
return TRUE;
if (arg->Typ() != type_syncvar) {
WerrorS("readSyncVar: argument is not a syncvar");
return TRUE;
}
SingularSyncVar *syncvar = *(SingularSyncVar **)arg->Data();
if (!syncvar) {
WerrorS("readSyncVar: syncvar has not been initialized");
return TRUE;
}
string item = syncvar->read();
leftv val = LinTree::from_string(item);
result->rtyp = val->Typ();
result->data = val->Data();
return FALSE;
}
BOOLEAN statSyncVar(leftv result, leftv arg) {
if (wrong_num_args("statSyncVar", arg, 1))
return TRUE;
if (arg->Typ() != type_syncvar) {
WerrorS("statSyncVar: argument is not a syncvar");
return TRUE;
}
SingularSyncVar *syncvar = *(SingularSyncVar **)arg->Data();
if (!syncvar) {
WerrorS("statSyncVar: syncvar has not been initialized");
return TRUE;
}
int init = syncvar->check();
result->rtyp = INT_CMD;
result->data = (char *)(long) init;
return FALSE;
}
void encode_shared(LinTree::LinTree &lintree, leftv val) {
SharedObject *obj = *(SharedObject **)(val->Data());
acquireShared(obj);
lintree.put(obj);
}
leftv decode_shared(LinTree::LinTree &lintree) {
int type = lintree.get_prev<int>();
SharedObject *obj = lintree.get<SharedObject *>();
leftv result = (leftv) omAlloc0Bin(sleftv_bin);
result->rtyp = type;
result->data = (void *)new_shared(obj);
return result;
}
void ref_shared(LinTree::LinTree &lintree, int by) {
SharedObject *obj = lintree.get<SharedObject *>();
while (by > 0) {
obj->incref();
by--;
}
while (by < 0) {
obj->decref();
by++;
}
}
void installShared(int type) {
LinTree::install(type, encode_shared, decode_shared, ref_shared);
}
void makeSharedType(int &type, const char *name) {
if (type != 0) return;
blackbox *b=(blackbox*)omAlloc0(sizeof(blackbox));
b->blackbox_Init = shared_init;
b->blackbox_destroy = shared_destroy;
b->blackbox_Copy = shared_copy;
b->blackbox_String = shared_string;
b->blackbox_Assign = shared_assign;
b->blackbox_CheckAssign = shared_check_assign;
// b->blackbox_Op2 = shared_op2;
// b->blackbox_Op3 = shared_op3;
type = setBlackboxStuff(b, name);
installShared(type);
}
void makeRegionlockType(int &type, const char *name) {
if (type != 0) return;
blackbox *b=(blackbox*)omAlloc0(sizeof(blackbox));
b->blackbox_Init = shared_init;
b->blackbox_destroy = rlock_destroy;
b->blackbox_Copy = shared_copy;
b->blackbox_String = shared_string;
b->blackbox_Assign = rlock_assign;
b->blackbox_CheckAssign = shared_check_assign;
type = setBlackboxStuff(b, name);
installShared(type);
}
#define MAX_THREADS 128
class ThreadState {
public:
bool active;
bool running;
int index;
void *(*thread_func)(ThreadState *, void *);
void *arg, *result;
pthread_t id;
pthread_t parent;
Lock lock;
ConditionVariable to_cond;
ConditionVariable from_cond;
queue<string> to_thread;
queue<string> from_thread;
ThreadState() : lock(), to_cond(&lock), from_cond(&lock),
to_thread(), from_thread() {
active = false;
running = false;
index = -1;
}
~ThreadState() {
// We do nothing here. This is to prevent the condition
// variable destructor from firing upon program exit,
// which would invoke undefined behavior if the thread
// is still running.
}
};
Lock thread_lock;
ThreadState *thread_state;
void setOption(int ch) {
int index = feGetOptIndex(ch);
feSetOptValue((feOptIndex) index, (int) 1);
}
void thread_init() {
master_lock.lock();
thread_id = ++thread_counter;
master_lock.unlock();
#ifdef ENABLE_THREADS
pSingular_initialize_thread();
siInit(global_argv0);
#endif
setOption('q');
// setOption('b');
}
void *thread_main(void *arg) {
ThreadState *ts = (ThreadState *)arg;
thread_init();
return ts->thread_func(ts, ts->arg);
}
void *interpreter_thread(ThreadState *ts, void *arg) {
ts->lock.lock();
for (;;) {
bool eval = false;
while (ts->to_thread.empty())
ts->to_cond.wait();
/* TODO */
string expr = ts->to_thread.front();
switch (expr[0]) {
case '\0': case 'q':
ts->lock.unlock();
return NULL;
case 'x':
eval = false;
break;
case 'e':
eval = true;
break;
}
ts->to_thread.pop();
expr = ts->to_thread.front();
/* this will implicitly eval commands */
leftv val = LinTree::from_string(expr);
expr = LinTree::to_string(val);
ts->to_thread.pop();
if (eval)
ts->from_thread.push(expr);
ts->from_cond.signal();
}
ts->lock.unlock();
return NULL;
}
class InterpreterThread : public SharedObject {
private:
ThreadState *ts;
public:
InterpreterThread(ThreadState *ts_init) : SharedObject(), ts(ts_init) { }
virtual ~InterpreterThread() { }
ThreadState *getThreadState() { return ts; }
void clearThreadState() {
ts = NULL;
}
};
static ThreadState *newThread(void *(*thread_func)(ThreadState *, void *),
void *arg, const char **error) {
ThreadState *ts = NULL;
if (error) *error = NULL;
thread_lock.lock();
for (int i=0; i<MAX_THREADS; i++) {
if (!thread_state[i].active) {
ts = thread_state + i;
ts->index = i;
ts->parent = pthread_self();
ts->active = true;
ts->running = true;
ts->to_thread = queue<string>();
ts->from_thread = queue<string>();
ts->thread_func = thread_func;
ts->arg = arg;
ts->result = NULL;
if (pthread_create(&ts->id, NULL, thread_main, ts)<0) {
if (error)
*error = "createThread: internal error: failed to create thread";
goto fail;
}
goto exit;
}
}
if (error) *error = "createThread: too many threads";
fail:
ts = NULL;
exit:
thread_lock.unlock();
return ts;
}
ThreadState *createThread(void *(*thread_func)(ThreadState *, void *),
void *arg) {
return newThread(thread_func, arg, NULL);
}
void *joinThread(ThreadState *ts) {
void *result;
pthread_join(ts->id, NULL);
result = ts->result;
thread_lock.lock();
ts->running = false;
ts->active = false;
thread_lock.unlock();
return result;
}
static InterpreterThread *createInterpreterThread(const char **error) {
ThreadState *ts = newThread(interpreter_thread, NULL, error);
if (*error) return NULL;
InterpreterThread *thread = new InterpreterThread(ts);
char buf[10];
snprintf(buf,10, "%d", ts->index);
string name(buf);
thread->set_name(name);
thread->set_type(type_thread);
return thread;
}
static BOOLEAN createThread(leftv result, leftv arg) {
Command cmd("createThread", result, arg);
cmd.check_argc(0);
const char *error;
if (!have_threads)
cmd.report("thread support not available");
if (!cmd.ok()) return cmd.status();
InterpreterThread *thread = createInterpreterThread(&error);
if (error) {
return cmd.abort(error);
}
cmd.set_result(type_thread, new_shared(thread));
return cmd.status();
}
static bool joinInterpreterThread(InterpreterThread *thread) {
ThreadState *ts = thread->getThreadState();
if (ts && ts->parent != pthread_self()) {
return false;
}
ts->lock.lock();
string quit("q");
ts->to_thread.push(quit);
ts->to_cond.signal();
ts->lock.unlock();
pthread_join(ts->id, NULL);
thread_lock.lock();
ts->running = false;
ts->active = false;
thread->clearThreadState();
thread_lock.unlock();
return true;
}
static BOOLEAN joinThread(leftv result, leftv arg) {
if (wrong_num_args("joinThread", arg, 1))
return TRUE;
if (arg->Typ() != type_thread) {
WerrorS("joinThread: argument is not a thread");
return TRUE;
}
InterpreterThread *thread = *(InterpreterThread **)arg->Data();
if (!joinInterpreterThread(thread)) {
WerrorS("joinThread: can only be called from parent thread");
return TRUE;
}
return FALSE;
}
class ThreadPool;
class Trigger;
class Job : public SharedObject {
public:
ThreadPool *pool;
long prio;
size_t id;
long pending_index;
vector<Job *> deps;
vector<Job *> notify;
vector<Trigger *> triggers;
vector<string> args;
string result; // lintree-encoded
void *data;
bool fast;
bool done;
bool queued;
bool running;
bool cancelled;
Job() : SharedObject(), pool(NULL), deps(), pending_index(-1), fast(false),
done(false), running(false), queued(false), cancelled(false), data(NULL),
result(), args(), notify(), triggers(), prio(0)
{ set_type(type_job); }
~Job();
void addDep(Job *job) {
deps.push_back(job);
}
void addDep(vector<Job *> &jobs);
void addDep(long ndeps, Job **jobs);
void addNotify(vector<Job *> &jobs);
void addNotify(Job *job);
virtual bool ready();
virtual void execute() = 0;
void run();
};
struct JobCompare {
bool operator()(const Job* lhs, const Job* rhs) {
if (lhs->fast < rhs->fast) {
return true;
}
if (lhs->prio < rhs->prio) {
return true;
}
if (lhs->prio == rhs->prio) {
return lhs->id > rhs->id;
}
return false;
}
};
class Trigger : public Job {
public:
virtual bool accept(leftv arg) = 0;
virtual void activate(leftv arg) = 0;
Trigger() : Job() { set_type(type_trigger); fast = true; }
};
bool Job::ready() {
vector<Job *>::iterator it;
for (it = deps.begin(); it != deps.end(); it++) {
if (!(*it)->done) return false;
}
return true;
}
Job::~Job() {
vector<Job *>::iterator it;
for (it = deps.begin(); it != deps.end(); it++) {
releaseShared(*it);
}
}
typedef queue<Job *> JobQueue;
class Scheduler;
struct SchedInfo {
Scheduler *scheduler;
Job *job;
int num;
};
STATIC_VAR ThreadPool *currentThreadPoolRef;
STATIC_VAR Job *currentJobRef;
class ThreadPool : public SharedObject {
public:
Scheduler *scheduler;
int nthreads;
ThreadPool(Scheduler *sched, int n);
ThreadPool(int n);
~ThreadPool();
ThreadState *getThread(int i);
void shutdown(bool wait);
void addThread(ThreadState *thread);
void attachJob(Job *job);
void detachJob(Job *job);
void queueJob(Job *job);
void broadcastJob(Job *job);
void cancelDeps(Job * job);
void cancelJob(Job *job);
void waitJob(Job *job);
void clearThreadState();
};
class Scheduler : public SharedObject {
private:
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;
friend class Job;
public:
Lock lock;
Scheduler(int n) :
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());
}
void set_maxconcurrency(int n) {
maxconcurrency = n;
}
int get_maxconcurrency() {
return maxconcurrency;
}
int threadpool_size(ThreadPool *pool) {
int n;
for (unsigned i = 0; i <thread_owners.size(); i++) {
if (thread_owners[i] == pool)
n++;
}
return n;
}
virtual ~Scheduler() {
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();
}
ThreadState *getThread(int i) { return threads[i]; }
void shutdown(bool wait) {
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]);
}
}
void addThread(ThreadPool *owner, ThreadState *thread) {
lock.lock();
thread_owners.push_back(owner);
threads.push_back(thread);
thread_queues.push_back(new JobQueue());
lock.unlock();
}
void attachJob(ThreadPool *pool, Job *job) {
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();
}
void detachJob(Job *job) {
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();
}
void queueJob(Job *job) {
lock.lock();
global_queue.push(job);
cond.signal();
lock.unlock();
}
void broadcastJob(ThreadPool *pool, Job *job) {
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();
}
void cancelDeps(Job * job) {
vector<Job *> &notify = job->notify;
for (unsigned i = 0; i <notify.size(); i++) {
Job *next = notify[i];
if (!next->cancelled) {
cancelJob(next);
}
}
}
void cancelJob(Job *job) {
lock.lock();
if (!job->cancelled) {
job->cancelled = true;
if (!job->running && !job->done) {
job->done = true;
cancelDeps(job);
}
}
lock.unlock();
}
void waitJob(Job *job) {
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();
}
}
void clearThreadState() {
threads.clear();
}
static void notifyDeps(Scheduler *scheduler, Job *job) {
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);
}
}
static void *main(ThreadState *ts, void *arg) {
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;
}
};
ThreadPool::ThreadPool(int n) : SharedObject(), nthreads(n) {
scheduler = new Scheduler(n);
acquireShared(scheduler);
}
ThreadPool::ThreadPool(Scheduler *sched, int n) : SharedObject(), nthreads(n) {
scheduler = sched;
acquireShared(sched);
}
ThreadPool::~ThreadPool() {
releaseShared(scheduler);
}
ThreadState *ThreadPool::getThread(int i) { return scheduler->getThread(i); }
void ThreadPool::shutdown(bool wait) { scheduler->shutdown(wait); }
void ThreadPool::addThread(ThreadState *thread) {
scheduler->addThread(this, thread);
}
void ThreadPool::attachJob(Job *job) {
scheduler->attachJob(this, job);
}
void ThreadPool::detachJob(Job *job) {
scheduler->detachJob(job);
}
void ThreadPool::queueJob(Job *job) {
scheduler->queueJob(job);
}
void ThreadPool::broadcastJob(Job *job) {
scheduler->broadcastJob(this, job);
}
void ThreadPool::cancelDeps(Job * job) {
scheduler->cancelDeps(job);
}
void ThreadPool::cancelJob(Job *job) {
scheduler->cancelJob(job);
}
void ThreadPool::waitJob(Job *job) {
scheduler->waitJob(job);
}
void ThreadPool::clearThreadState() {
scheduler->clearThreadState();
}
void Job::addDep(vector<Job *> &jobs) {
deps.insert(deps.end(), jobs.begin(), jobs.end());
}
void Job::addDep(long ndeps, Job **jobs) {
for (long i = 0; i < ndeps; i++) {
deps.push_back(jobs[i]);
}
}
void Job::addNotify(vector<Job *> &jobs) {
notify.insert(notify.end(), jobs.begin(), jobs.end());
if (done) {
Scheduler::notifyDeps(pool->scheduler, this);
}
}
void Job::addNotify(Job *job) {
notify.push_back(job);
if (done) {
Scheduler::notifyDeps(pool->scheduler, this);
}
}
void Job::run() {
if (!cancelled) {
running = true;
pool->scheduler->lock.unlock();
pool->scheduler->running++;
execute();
pool->scheduler->running--;
pool->scheduler->lock.lock();
running = false;
}
done = true;
}
class AccTrigger : public Trigger {
private:
long count;
public:
AccTrigger(long count_init): Trigger(), count(count_init) {
}
virtual bool ready() {
if (!Trigger::ready()) return false;
return args.size() >= count;
}
virtual bool accept(leftv arg) {
return true;
}
virtual void activate(leftv arg) {
while (arg != NULL && !ready()) {
args.push_back(LinTree::to_string(arg));
if (ready()) {
return;
}
arg = arg->next;
}
}
virtual void execute() {
lists l = (lists) omAlloc0Bin(slists_bin);
l->Init(args.size());
for (unsigned i = 0; i < args.size(); i++) {
leftv val = LinTree::from_string(args[i]);
memcpy(&l->m[i], val, sizeof(*val));
omFreeBin(val, sleftv_bin);
}
sleftv val;
memset(&val, 0, sizeof(val));
val.rtyp = LIST_CMD;
val.data = l;
result = LinTree::to_string(&val);
// val.CleanUp();
}
};
class CountTrigger : public Trigger {
private:
long count;
public:
CountTrigger(long count_init): Trigger(), count(count_init) {
}
virtual bool ready() {
if (!Trigger::ready()) return false;
return count <= 0;
}
virtual bool accept(leftv arg) {
return arg == NULL;
}
virtual void activate(leftv arg) {
if (!ready()) {
count--;
}
}
virtual void execute() {
// do nothing
}
};
class SetTrigger : public Trigger {
private:
vector<bool> set;
long count;
public:
SetTrigger(long count_init) : Trigger(), count(0),
set(count_init) {
}
virtual bool ready() {
if (!Trigger::ready()) return false;
return count == set.size();
}
virtual bool accept(leftv arg) {
return arg->Typ() == INT_CMD;
}
virtual void activate(leftv arg) {
if (!ready()) {
long value = (long) arg->Data();
if (value < 0 || value >= count) return;
if (set[value]) return;
set[value] = true;
count++;
}
}
virtual void execute() {
// do nothing
}
};
class ProcTrigger : public Trigger {
private:
string procname;
bool success;
public:
ProcTrigger(const char *p) : Trigger(), procname(p), success(false) {
}
virtual bool ready() {
if (!Trigger::ready()) return false;
return success;
}
virtual bool accept(leftv arg) {
return TRUE;
}
virtual void activate(leftv arg) {
if (!ready()) {
pool->scheduler->lock.unlock();
vector<leftv> argv;
for (unsigned i = 0; i < args.size(); i++) {
appendArg(argv, args[i]);
}
int error = false;
while (arg) {
appendArgCopy(argv, arg);
arg = arg->next;
}
sleftv val;
if (!error)
error = executeProc(val, procname.c_str(), argv);
if (!error) {
if (val.Typ() == NONE || (val.Typ() == INT_CMD &&
(long) val.Data()))
{
success = true;
}
val.CleanUp();
}
pool->scheduler->lock.lock();
}
}
virtual void execute() {
// do nothing
}
};
static BOOLEAN createThreadPool(leftv result, leftv arg) {
long n;
Command cmd("createThreadPool", result, arg);
cmd.check_argc(1, 2);
cmd.check_arg(0, INT_CMD, "first argument must be an integer");
if (cmd.ok()) {
n = (long) cmd.arg(0);
if (n < 0) cmd.report("number of threads must be non-negative");
else if (n >= 256) cmd.report("number of threads too large");
if (!have_threads && n != 0)
cmd.report("in single-threaded mode, number of threads must be zero");
}
if (cmd.ok()) {
ThreadPool *pool = new ThreadPool((int) n);
pool->set_type(type_threadpool);
for (int i = 0; i <n; i++) {
const char *error;
SchedInfo *info = new SchedInfo();
info->scheduler = pool->scheduler;
acquireShared(pool->scheduler);
info->job = NULL;
info->num = i;
ThreadState *thread = newThread(Scheduler::main, info, &error);
if (!thread) {
// TODO: clean up bad pool
return cmd.abort(error);
}
pool->addThread(thread);
}
cmd.set_result(type_threadpool, new_shared(pool));
}
return cmd.status();
}
static BOOLEAN createThreadPoolSet(leftv result, leftv arg) {
Command cmd("createThreadPoolSet", result, arg);
cmd.check_argc(2);
cmd.check_arg(0, INT_CMD, "first argument must be an integer");
cmd.check_arg(1, LIST_CMD, "second argument must be a list of integers");
lists l;
int n;
if (cmd.ok()) {
l = (lists) (cmd.arg(1));
n = lSize(l)+1;
if (n == 0)
return cmd.abort("second argument must not be empty");
for (int i = 0; i < n; i++) {
if (l->m[i].Typ() != INT_CMD)
return cmd.abort("second argument must be a list of integers");
}
}
lists pools = (lists) omAlloc0Bin(slists_bin);
pools->Init(n);
if (cmd.ok()) {
long s = 0;
for (int i = 0; i < n; i++) {
s += (long) (l->m[i].Data());
}
Scheduler *sched = new Scheduler((int)s);
sched->set_maxconcurrency(cmd.int_arg(0));
for (int i = 0; i < n; i++) {
long m = (long) (l->m[i].Data());
ThreadPool *pool = new ThreadPool(sched, (int) m);
pool->set_type(type_threadpool);
for (int j = 0; j < m; j++) {
const char *error;
SchedInfo *info = new SchedInfo();
info->scheduler = pool->scheduler;
acquireShared(pool->scheduler);
info->job = NULL;
info->num = i;
ThreadState *thread = newThread(Scheduler::main, info, &error);
if (!thread) {
// TODO: clean up bad pool
return cmd.abort(error);
}
pool->addThread(thread);
}
pools->m[i].rtyp = type_threadpool;
pools->m[i].data = new_shared(pool);
}
cmd.set_result(LIST_CMD, pools);
}
return cmd.status();
}
ThreadPool *createThreadPool(int nthreads, int prioThreads = 0) {
ThreadPool *pool = new ThreadPool((int) nthreads);
pool->set_type(type_threadpool);
for (int i = 0; i <nthreads; i++) {
const char *error;
SchedInfo *info = new SchedInfo();
info->scheduler = pool->scheduler;
acquireShared(pool);
info->job = NULL;
info->num = i;
ThreadState *thread = newThread(Scheduler::main, info, &error);
if (!thread) {
return NULL;
}
pool->addThread(thread);
}
return pool;
}
void release(ThreadPool *pool) {
releaseShared(pool);
}
void retain(ThreadPool *pool) {
acquireShared(pool);
}
ThreadPool *getCurrentThreadPool() {
return currentThreadPoolRef;
}
static BOOLEAN getThreadPoolWorkers(leftv result, leftv arg) {
Command cmd("getThreadPoolWorkers", result, arg);
cmd.check_argc(1);
cmd.check_arg(0, type_threadpool, "argument must be a threadpool");
cmd.check_init(0, "threadpool not initialized");
int r = 0;
if (cmd.ok()) {
ThreadPool *pool = cmd.shared_arg<ThreadPool>(0);
Scheduler *sched = pool->scheduler;
sched->lock.lock();
r = sched->threadpool_size(pool);
sched->lock.unlock();
cmd.set_result(INT_CMD, r);
}
return cmd.status();
}
static BOOLEAN setThreadPoolWorkers(leftv result, leftv arg) {
Command cmd("setThreadPoolWorkers", result, arg);
cmd.check_argc(2);
cmd.check_arg(0, type_threadpool, "first argument must be a threadpool");
cmd.check_arg(1, INT_CMD, "second argument must be an integer");
cmd.check_init(0, "threadpool not initialized");
if (cmd.ok()) {
ThreadPool *pool = cmd.shared_arg<ThreadPool>(0);
Scheduler *sched = pool->scheduler;
// TODO: count/add threads
cmd.no_result();
}
return cmd.status();
}
static BOOLEAN getThreadPoolConcurrency(leftv result, leftv arg) {
Command cmd("getThreadPoolConcurrency", result, arg);
cmd.check_argc(1);
cmd.check_arg(0, type_threadpool, "argument must be a threadpool");
cmd.check_init(0, "threadpool not initialized");
if (cmd.ok()) {
ThreadPool *pool = cmd.shared_arg<ThreadPool>(0);
Scheduler *sched = pool->scheduler;
sched->lock.lock();
cmd.set_result(INT_CMD, sched->get_maxconcurrency());
sched->lock.unlock();
}
return cmd.status();
}
static BOOLEAN setThreadPoolConcurrency(leftv result, leftv arg) {
Command cmd("setThreadPoolWorkers", result, arg);
cmd.check_argc(2);
cmd.check_arg(0, type_threadpool, "first argument must be a threadpool");
cmd.check_arg(1, INT_CMD, "second argument must be an integer");
cmd.check_init(0, "threadpool not initialized");
if (cmd.ok()) {
ThreadPool *pool = cmd.shared_arg<ThreadPool>(0);
Scheduler *sched = pool->scheduler;
sched->lock.lock();
sched->set_maxconcurrency(cmd.int_arg(1));
sched->lock.unlock();
cmd.no_result();
}
return cmd.status();
}
static BOOLEAN closeThreadPool(leftv result, leftv arg) {
Command cmd("closeThreadPool", result, arg);
cmd.check_argc(1, 2);
cmd.check_arg(0, type_threadpool, "first argument must be a threadpool");
cmd.check_init(0, "threadpool not initialized");
if (cmd.nargs() > 1)
cmd.check_arg(1, INT_CMD, "optional argument must be an integer");
if (cmd.ok()) {
ThreadPool *pool = *(ThreadPool **)(cmd.arg(0));
bool wait = cmd.nargs() == 2 ? (cmd.int_arg(1) != 0) : 1;
pool->shutdown(wait);
cmd.no_result();
}
return cmd.status();
}
void closeThreadPool(ThreadPool *pool, bool wait) {
pool->shutdown(wait);
}
BOOLEAN currentThreadPool(leftv result, leftv arg) {
Command cmd("currentThreadPool", result, arg);
cmd.check_argc(0);
ThreadPool *pool = currentThreadPoolRef;
if (pool) {
cmd.set_result(type_threadpool, new_shared(pool));
} else {
cmd.report("no current threadpool");
}
return cmd.status();
}
BOOLEAN setCurrentThreadPool(leftv result, leftv arg) {
Command cmd("setCurrentThreadPool", result, arg);
cmd.check_argc(1);
cmd.check_init(0, "threadpool not initialized");
if (cmd.ok()) {
ThreadPool *pool = *(ThreadPool **)(cmd.arg(0));
acquireShared(pool);
if (currentThreadPoolRef)
releaseShared(currentThreadPoolRef);
currentThreadPoolRef = pool;
}
return cmd.status();
}
class EvalJob : public Job {
public:
EvalJob() : Job() { }
virtual void execute() {
leftv val = LinTree::from_string(args[0]);
result = (LinTree::to_string(val));
val->CleanUp();
omFreeBin(val, sleftv_bin);
}
};
class ExecJob : public Job {
public:
ExecJob() : Job() { }
virtual void execute() {
leftv val = LinTree::from_string(args[0]);
val->CleanUp();
omFreeBin(val, sleftv_bin);
}
};
class ProcJob : public Job {
string procname;
public:
ProcJob(const char *procname_init) : Job(),
procname(procname_init) {
set_name(procname_init);
}
virtual void execute() {
vector<leftv> argv;
for (unsigned i = 0; i <args.size(); i++) {
appendArg(argv, args[i]);
}
for (unsigned i = 0; i < deps.size(); i++) {
appendArg(argv, deps[i]->result);
}
sleftv val;
int error = executeProc(val, procname.c_str(), argv);
if (!error) {
result = (LinTree::to_string(&val));
val.CleanUp();
}
}
};
class KernelJob : public Job {
private:
void (*cfunc)(leftv result, leftv arg);
public:
KernelJob(void (*func)(leftv result, leftv arg)) : cfunc(func) { }
virtual void execute() {
vector<leftv> argv;
for (unsigned i = 0; i <args.size(); i++) {
appendArg(argv, args[i]);
}
for (unsigned i = 0; i < deps.size(); i++) {
appendArg(argv, deps[i]->result);
}
sleftv val;
memset(&val, 0, sizeof(val));
if (argv.size() > 0) {
leftv *tail = &argv[0]->next;
for (unsigned i = 1; i < argv.size(); i++) {
*tail = argv[i];
tail = &(*tail)->next;
}
*tail = NULL;
}
cfunc(&val, argv[0]);
result = (LinTree::to_string(&val));
val.CleanUp();
}
};
class RawKernelJob : public Job {
private:
void (*cfunc)(long ndeps, Job **deps);
public:
RawKernelJob(void (*func)(long ndeps, Job **deps)) : cfunc(func) { }
virtual void execute() {
long ndeps = deps.size();
Job **jobs = (Job **) omAlloc0(sizeof(Job *) * ndeps);
for (long i = 0; i < ndeps; i++)
jobs[i] = deps[i];
cfunc(ndeps, jobs);
omFree(jobs);
}
};
static BOOLEAN createJob(leftv result, leftv arg) {
Command cmd("createJob", result, arg);
cmd.check_argc_min(1);
cmd.check_arg(0, STRING_CMD, COMMAND,
"job name must be a string or quote expression");
if (cmd.ok()) {
if (cmd.test_arg(0, STRING_CMD)) {
ProcJob *job = new ProcJob((char *)(cmd.arg(0)));
for (leftv a = arg->next; a != NULL; a = a->next) {
job->args.push_back(LinTree::to_string(a));
}
cmd.set_result(type_job, new_shared(job));
} else {
cmd.check_argc(1);
Job *job = new EvalJob();
job->args.push_back(LinTree::to_string(arg));
cmd.set_result(type_job, new_shared(job));
}
}
return cmd.status();
}
Job *createJob(void (*func)(leftv result, leftv arg)) {
KernelJob *job = new KernelJob(func);
return job;
}
Job *createJob(void (*func)(long ndeps, Job **deps)) {
RawKernelJob *job = new RawKernelJob(func);
return job;
}
Job *startJob(ThreadPool *pool, Job *job, leftv arg) {
if (job->pool) return NULL;
while (arg) {
job->args.push_back(LinTree::to_string(arg));
arg = arg->next;
}
pool->attachJob(job);
return job;
}
Job *startJob(ThreadPool *pool, Job *job) {
return startJob(pool, job, NULL);
}
// Job *scheduleJob(ThreadPool *pool, Job *job, long ndeps, Job **deps) {
// if (job->pool) return NULL;
// pool->scheduler->lock.lock();
// bool cancelled = false;
// job->addDep(ndeps, deps);
// for (long i = 0; i < ndeps; i++) {
// deps[i]->addNotify(job);
// cancelled |= deps[i]->cancelled;
// }
// if (cancelled) {
// job->pool = pool;
// pool->cancelJob(job);
// }
// else
// pool->attachJob(job);
// pool->scheduler->lock.unlock();
// return FIXME: missing/unclear what this is supposed to be
// }
void cancelJob(Job *job) {
ThreadPool *pool = job->pool;
if (pool) pool->cancelJob(job);
}
Job *getCurrentJob() {
return currentJobRef;
}
static BOOLEAN startJob(leftv result, leftv arg) {
Command cmd("startJob", result, arg);
cmd.check_argc_min(1);
int has_pool = cmd.test_arg(0, type_threadpool);
cmd.check_argc_min(1+has_pool);
if (has_pool)
cmd.check_init(0, "threadpool not initialized");
int has_prio = cmd.test_arg(has_pool, INT_CMD);
long prio = has_prio ? (long) cmd.arg(has_pool) : 0L;
int first_arg = has_pool + has_prio;
cmd.check_arg(first_arg, type_job, STRING_CMD,
"job argument must be a job or string");
if (cmd.ok() && cmd.argtype(first_arg) == type_job)
cmd.check_init(first_arg, "job not initialized");
if (!cmd.ok()) return cmd.status();
ThreadPool *pool;
if (has_pool)
pool = cmd.shared_arg<ThreadPool>(0);
else {
if (!currentThreadPoolRef)
return cmd.abort("no current threadpool defined");
pool = currentThreadPoolRef;
}
Job *job;
if (cmd.argtype(first_arg) == type_job)
job = *(Job **)(cmd.arg(first_arg));
else
job = new ProcJob((char *)(cmd.arg(first_arg)));
leftv a = arg->next;
if (has_pool) a = a->next;
if (has_prio) a = a->next;
for (; a != NULL; a = a->next) {
job->args.push_back(LinTree::to_string(a));
}
if (job->pool)
return cmd.abort("job has already been scheduled");
job->prio = prio;
pool->attachJob(job);
cmd.set_result(type_job, new_shared(job));
return cmd.status();
}
static BOOLEAN waitJob(leftv result, leftv arg) {
Command cmd("waitJob", result, arg);
cmd.check_argc(1);
cmd.check_arg(0, type_job, "argument must be a job");
cmd.check_init(0, "job not initialized");
if (cmd.ok()) {
Job *job = *(Job **)(cmd.arg(0));
ThreadPool *pool = job->pool;
if (!pool) {
return cmd.abort("job has not yet been started or scheduled");
}
pool->waitJob(job);
if (job->cancelled) {
return cmd.abort("job has been cancelled");
}
if (job->result.size() == 0)
cmd.no_result();
else {
leftv res = LinTree::from_string(job->result);
cmd.set_result(res->Typ(), res->Data());
}
}
return cmd.status();
}
void waitJob(Job *job) {
assert(job->pool != NULL);
job->pool->waitJob(job);
}
static BOOLEAN cancelJob(leftv result, leftv arg) {
Command cmd("cancelJob", result, arg);
cmd.check_argc(1);
cmd.check_arg(0, type_job, "argument must be a job");
cmd.check_init(0, "job not initialized");
if (cmd.ok()) {
Job *job = cmd.shared_arg<Job>(0);
ThreadPool *pool = job->pool;
if (!pool) {
return cmd.abort("job has not yet been started or scheduled");
}
pool->cancelJob(job);
cmd.no_result();
}
return cmd.status();
}
static BOOLEAN jobCancelled(leftv result, leftv arg) {
Job *job;
Command cmd("jobCancelled", result, arg);
cmd.check_argc(0, 1);
if (cmd.nargs() == 1) {
cmd.check_arg(0, type_job, "argument must be a job");
cmd.check_init(0, "job not initialized");
job = cmd.shared_arg<Job>(0);
} else {
job = currentJobRef;
if (!job)
cmd.report("no current job");
}
if (cmd.ok()) {
ThreadPool *pool = job->pool;
if (!pool) {
return cmd.abort("job has not yet been started or scheduled");
}
pool->scheduler->lock.lock();
cmd.set_result((long) job->cancelled);
pool->scheduler->lock.unlock();
}
return cmd.status();
}
bool getJobCancelled(Job *job) {
ThreadPool *pool = job->pool;
if (pool) pool->scheduler->lock.lock();
bool result = job->cancelled;
if (pool) pool->scheduler->lock.unlock();
return result;
}
bool getJobCancelled() {
return getJobCancelled(currentJobRef);
}
void setJobData(Job *job, void *data) {
ThreadPool *pool = job->pool;
if (pool) pool->scheduler->lock.lock();
job->data = data;
if (pool) pool->scheduler->lock.unlock();
}
void *getJobData(Job *job) {
ThreadPool *pool = job->pool;
if (pool) pool->scheduler->lock.lock();
void *result = job->data;
if (pool) pool->scheduler->lock.unlock();
return result;
}
void addJobArgs(Job *job, leftv arg) {
ThreadPool *pool = job->pool;
if (pool) pool->scheduler->lock.lock();
while (arg) {
job->args.push_back(LinTree::to_string(arg));
arg = arg->next;
}
if (pool) pool->scheduler->lock.unlock();
}
leftv getJobResult(Job *job) {
ThreadPool *pool = job->pool;
if (pool) pool->scheduler->lock.lock();
leftv result = LinTree::from_string(job->result);
if (pool) pool->scheduler->lock.unlock();
return result;
}
const char *getJobName(Job *job) {
// TODO
return "";
}
void setJobName(Job *job, const char *name) {
// TODO
}
static BOOLEAN createTrigger(leftv result, leftv arg) {
Command cmd("createTrigger", result, arg);
cmd.check_argc_min(1);
int has_pool = cmd.test_arg(0, type_threadpool);
ThreadPool *pool;
if (has_pool) {
cmd.check_init(0, "threadpool not initialized");
pool = cmd.shared_arg<ThreadPool>(0);
} else {
pool = currentThreadPoolRef;
if (!pool)
return cmd.abort("no default threadpool");
}
cmd.check_argc(has_pool + 2);
cmd.check_arg(has_pool + 0, STRING_CMD, "trigger subtype must be a string");
const char *kind = (const char *)(cmd.arg(has_pool + 0));
if (0 == strcmp(kind, "proc")) {
cmd.check_arg(has_pool + 1, STRING_CMD, "proc trigger argument must be a string");
} else {
cmd.check_arg(has_pool + 1, INT_CMD, "trigger argument must be an integer");
}
if (cmd.ok()) {
Trigger *trigger;
long n = (long) (cmd.arg(has_pool + 1));
if (n < 0)
return cmd.abort("trigger argument must be a non-negative integer");
if (0 == strcmp(kind, "acc")) {
trigger = new AccTrigger(n);
} else if (0 == strcmp(kind, "count")) {
trigger = new CountTrigger(n);
} else if (0 == strcmp(kind, "set")) {
trigger = new SetTrigger(n);
} else if (0 == strcmp(kind, "proc")) {
trigger = new ProcTrigger((const char *) cmd.arg(has_pool + 1));
} else {
return cmd.abort("unknown trigger subtype");
}
pool->attachJob(trigger);
cmd.set_result(type_trigger, new_shared(trigger));
}
return cmd.status();
}
static BOOLEAN updateTrigger(leftv result, leftv arg) {
Command cmd("updateTrigger", result, arg);
cmd.check_argc_min(1);
cmd.check_arg(0, type_trigger, "first argument must be a trigger");
cmd.check_init(0, "trigger not initialized");
if (cmd.ok()) {
Trigger *trigger = cmd.shared_arg<Trigger>(0);
trigger->pool->scheduler->lock.lock();
if (!trigger->accept(arg->next))
cmd.report("incompatible argument type(s) for this trigger");
else {
trigger->activate(arg->next);
if (trigger->ready()) {
trigger->run();
Scheduler::notifyDeps(trigger->pool->scheduler, trigger);
}
}
trigger->pool->scheduler->lock.unlock();
}
return cmd.status();
}
static BOOLEAN chainTrigger(leftv result, leftv arg) {
Command cmd("chainTrigger", result, arg);
cmd.check_argc(2);
cmd.check_arg(0, type_trigger, "first argument must be a trigger");
cmd.check_arg(1, type_trigger, type_job,
"second argument must be a trigger or job");
cmd.check_init(0, "trigger not initialized");
cmd.check_init(1, "trigger/job not initialized");
if (cmd.ok()) {
Trigger *trigger = cmd.shared_arg<Trigger>(0);
Job *job = cmd.shared_arg<Job>(1);
if (trigger->pool != job->pool)
return cmd.abort("arguments use different threadpools");
ThreadPool *pool = trigger->pool;
pool->scheduler->lock.lock();
job->triggers.push_back(trigger);
pool->scheduler->lock.unlock();
}
return cmd.status();
}
static BOOLEAN testTrigger(leftv result, leftv arg) {
Command cmd("testTrigger", result, arg);
cmd.check_argc(1);
cmd.check_arg(0, type_trigger, "argument must be a trigger");
cmd.check_init(0, "trigger not initialized");
if (cmd.ok()) {
Trigger *trigger = cmd.shared_arg<Trigger>(0);
ThreadPool *pool = trigger->pool;
pool->scheduler->lock.lock();
cmd.set_result((long)trigger->ready());
pool->scheduler->lock.unlock();
}
return cmd.status();
}
static BOOLEAN scheduleJob(leftv result, leftv arg) {
vector<Job *> jobs;
vector<Job *> deps;
Command cmd("scheduleJob", result, arg);
cmd.check_argc_min(1);
int has_pool = cmd.test_arg(0, type_threadpool);
if (has_pool) {
cmd.check_arg(0, type_threadpool, "first argument must be a threadpool");
cmd.check_init(0, "threadpool not initialized");
}
cmd.check_argc_min(has_pool+1);
int has_prio = cmd.test_arg(has_pool, INT_CMD);
ThreadPool *pool;
if (has_pool)
pool = cmd.shared_arg<ThreadPool>(0);
else {
if (!currentThreadPoolRef)
return cmd.abort("no current threadpool defined");
pool = currentThreadPoolRef;
}
long prio = has_prio ? (long) cmd.arg(has_pool) : 0L;
int first_arg = has_pool + has_prio;
if (cmd.test_arg(first_arg, type_job)) {
jobs.push_back(*(Job **)(cmd.arg(first_arg)));
} else if (cmd.test_arg(first_arg, STRING_CMD)) {
jobs.push_back(new ProcJob((char *)(cmd.arg(first_arg))));
} else if (cmd.test_arg(first_arg, LIST_CMD)) {
lists l = (lists) (cmd.arg(first_arg));
int n = lSize(l);
for (int i = 0; i < n; i++) {
if (l->m[i].Typ() != type_job)
return cmd.abort("job argument must be a job, string, or list of jobs");
}
for (int i = 0; i < n; i++) {
Job *job = *(Job **) (l->m[i].Data());
if (!job)
return cmd.abort("job not initialized");
jobs.push_back(job);
}
} else {
return cmd.abort("job argument must be a job, string, or list of jobs");
}
bool error = false;
leftv a = arg->next;
if (has_pool) a = a->next;
if (has_prio) a = a->next;
for (; !error && a; a = a->next) {
if (a->Typ() == type_job || a->Typ() == type_trigger) {
deps.push_back(*(Job **)(a->Data()));
} else if (a->Typ() == LIST_CMD) {
lists l = (lists) a->Data();
int n = lSize(l);
for (int i = 0; i < n; i++) {
if (l->m[i].Typ() == type_job || l->m[i].Typ() == type_trigger) {
deps.push_back(*(Job **)(l->m[i].Data()));
} else {
error = true;
break;
}
}
}
}
if (error) {
return cmd.abort("illegal dependency");
}
for (unsigned i = 0; i < jobs.size(); i++) {
Job *job = jobs[i];
if (job->pool) {
return cmd.abort("job has already been scheduled");
}
job->prio = prio;
}
for (unsigned i = 0; i < deps.size(); i++) {
Job *job = deps[i];
if (!job->pool) {
return cmd.abort("dependency has not yet been scheduled");
}
if (job->pool != pool) {
return cmd.abort("dependency has been scheduled on a different threadpool");
}
}
pool->scheduler->lock.lock();
bool cancelled = false;
for (unsigned i = 0; i < jobs.size(); i++) {
jobs[i]->addDep(deps);
}
for (unsigned i = 0; i < deps.size(); i++) {
deps[i]->addNotify(jobs);
cancelled |= deps[i]->cancelled;
}
for (unsigned i = 0; i < jobs.size(); i++) {
if (cancelled) {
jobs[i]->pool = pool;
pool->cancelJob(jobs[i]);
}
else
pool->attachJob(jobs[i]);
}
pool->scheduler->lock.unlock();
if (jobs.size() > 0)
cmd.set_result(type_job, new_shared(jobs[0]));
return cmd.status();
}
BOOLEAN currentJob(leftv result, leftv arg) {
Command cmd("currentJob", result, arg);
cmd.check_argc(0);
Job *job = currentJobRef;
if (job) {
cmd.set_result(type_job, new_shared(job));
} else {
cmd.report("no current job");
}
return cmd.status();
}
BOOLEAN threadID(leftv result, leftv arg) {
if (wrong_num_args("threadID", arg, 0))
return TRUE;
result->rtyp = INT_CMD;
result->data = (char *)thread_id;
return FALSE;
}
BOOLEAN mainThread(leftv result, leftv arg) {
if (wrong_num_args("mainThread", arg, 0))
return TRUE;
result->rtyp = INT_CMD;
result->data = (char *)(long)(thread_id == 0L);
return FALSE;
}
BOOLEAN threadEval(leftv result, leftv arg) {
if (wrong_num_args("threadEval", arg, 2))
return TRUE;
if (arg->Typ() != type_thread) {
WerrorS("threadEval: argument is not a thread");
return TRUE;
}
InterpreterThread *thread = *(InterpreterThread **)arg->Data();
string expr = LinTree::to_string(arg->next);
ThreadState *ts = thread->getThreadState();
if (ts && ts->parent != pthread_self()) {
WerrorS("threadEval: can only be called from parent thread");
return TRUE;
}
if (ts) ts->lock.lock();
if (!ts || !ts->running || !ts->active) {
WerrorS("threadEval: thread is no longer running");
if (ts) ts->lock.unlock();
return TRUE;
}
ts->to_thread.push("e");
ts->to_thread.push(expr);
ts->to_cond.signal();
ts->lock.unlock();
result->rtyp = NONE;
return FALSE;
}
BOOLEAN threadExec(leftv result, leftv arg) {
if (wrong_num_args("threadExec", arg, 2))
return TRUE;
if (arg->Typ() != type_thread) {
WerrorS("threadExec: argument is not a thread");
return TRUE;
}
InterpreterThread *thread = *(InterpreterThread **)arg->Data();
string expr = LinTree::to_string(arg->next);
ThreadState *ts = thread->getThreadState();
if (ts && ts->parent != pthread_self()) {
WerrorS("threadExec: can only be called from parent thread");
return TRUE;
}
if (ts) ts->lock.lock();
if (!ts || !ts->running || !ts->active) {
WerrorS("threadExec: thread is no longer running");
if (ts) ts->lock.unlock();
return TRUE;
}
ts->to_thread.push("x");
ts->to_thread.push(expr);
ts->to_cond.signal();
ts->lock.unlock();
result->rtyp = NONE;
return FALSE;
}
BOOLEAN threadPoolExec(leftv result, leftv arg) {
Command cmd("threadPoolExec", result, arg);
ThreadPool *pool;
cmd.check_argc(1, 2);
int has_pool = cmd.nargs() == 2;
if (has_pool) {
cmd.check_arg(0, type_threadpool, "first argument must be a threadpool");
cmd.check_init(0, "threadpool not initialized");
pool = cmd.shared_arg<ThreadPool>(0);
} else {
pool = currentThreadPoolRef;
if (!pool)
return cmd.abort("no current threadpool");
}
if (cmd.ok()) {
string expr = LinTree::to_string(has_pool ? arg->next : arg);
Job* job = new ExecJob();
job->args.push_back(expr);
job->pool = pool;
pool->broadcastJob(job);
}
return cmd.status();
}
BOOLEAN threadResult(leftv result, leftv arg) {
if (wrong_num_args("threadResult", arg, 1))
return TRUE;
if (arg->Typ() != type_thread) {
WerrorS("threadResult: argument is not a thread");
return TRUE;
}
InterpreterThread *thread = *(InterpreterThread **)arg->Data();
ThreadState *ts = thread->getThreadState();
if (ts && ts->parent != pthread_self()) {
WerrorS("threadResult: can only be called from parent thread");
return TRUE;
}
if (ts) ts->lock.lock();
if (!ts || !ts->running || !ts->active) {
WerrorS("threadResult: thread is no longer running");
if (ts) ts->lock.unlock();
return TRUE;
}
while (ts->from_thread.empty()) {
ts->from_cond.wait();
}
string expr = ts->from_thread.front();
ts->from_thread.pop();
ts->lock.unlock();
leftv val = LinTree::from_string(expr);
result->rtyp = val->Typ();
result->data = val->Data();
return FALSE;
}
BOOLEAN setSharedName(leftv result, leftv arg) {
Command cmd("setSharedName", result, arg);
cmd.check_argc(2);
int type = cmd.argtype(0);
cmd.check_init(0, "first argument is not initialized");
if (type != type_job && type != type_trigger && type != type_threadpool) {
cmd.report("first argument must be a job, trigger, or threadpool");
}
cmd.check_arg(1, STRING_CMD, "second argument must be a string");
if (cmd.ok()) {
SharedObject *obj = cmd.shared_arg<SharedObject>(0);
name_lock.lock();
obj->set_name((char *) cmd.arg(1));
name_lock.unlock();
}
return cmd.status();
}
BOOLEAN getSharedName(leftv result, leftv arg) {
Command cmd("getSharedName", result, arg);
cmd.check_argc(1);
int type = cmd.argtype(0);
cmd.check_init(0, "first argument is not initialized");
if (type != type_job && type != type_trigger && type != type_threadpool) {
cmd.report("first argument must be a job, trigger, or threadpool");
}
if (cmd.ok()) {
SharedObject *obj = cmd.shared_arg<SharedObject>(0);
name_lock.lock();
cmd.set_result(obj->get_name().c_str());
name_lock.unlock();
}
return cmd.status();
}
}
using namespace LibThread;
extern "C" int SI_MOD_INIT(systhreads)(SModulFunctions *fn)
{
const char *libname = currPack->libname;
if (!libname) libname = "";
master_lock.lock();
if (!thread_state)
thread_state = new ThreadState[MAX_THREADS];
makeSharedType(type_atomic_table, "atomic_table");
makeSharedType(type_atomic_list, "atomic_list");
makeSharedType(type_shared_table, "shared_table");
makeSharedType(type_shared_list, "shared_list");
makeSharedType(type_channel, "channel");
makeSharedType(type_syncvar, "syncvar");
makeSharedType(type_region, "region");
makeSharedType(type_thread, "thread");
makeSharedType(type_threadpool, "threadpool");
makeSharedType(type_job, "job");
makeSharedType(type_trigger, "trigger");
makeRegionlockType(type_regionlock, "regionlock");
fn->iiAddCproc(libname, "putTable", FALSE, putTable);
fn->iiAddCproc(libname, "getTable", FALSE, getTable);
fn->iiAddCproc(libname, "inTable", FALSE, inTable);
fn->iiAddCproc(libname, "putList", FALSE, putList);
fn->iiAddCproc(libname, "getList", FALSE, getList);
fn->iiAddCproc(libname, "lockRegion", FALSE, lockRegion);
fn->iiAddCproc(libname, "regionLock", FALSE, regionLock);
fn->iiAddCproc(libname, "unlockRegion", FALSE, unlockRegion);
fn->iiAddCproc(libname, "sendChannel", FALSE, sendChannel);
fn->iiAddCproc(libname, "receiveChannel", FALSE, receiveChannel);
fn->iiAddCproc(libname, "statChannel", FALSE, statChannel);
fn->iiAddCproc(libname, "writeSyncVar", FALSE, writeSyncVar);
fn->iiAddCproc(libname, "updateSyncVar", FALSE, updateSyncVar);
fn->iiAddCproc(libname, "readSyncVar", FALSE, readSyncVar);
fn->iiAddCproc(libname, "statSyncVar", FALSE, statSyncVar);
fn->iiAddCproc(libname, "makeAtomicTable", FALSE, makeAtomicTable);
fn->iiAddCproc(libname, "makeAtomicList", FALSE, makeAtomicList);
fn->iiAddCproc(libname, "makeSharedTable", FALSE, makeSharedTable);
fn->iiAddCproc(libname, "makeSharedList", FALSE, makeSharedList);
fn->iiAddCproc(libname, "makeChannel", FALSE, makeChannel);
fn->iiAddCproc(libname, "makeSyncVar", FALSE, makeSyncVar);
fn->iiAddCproc(libname, "makeRegion", FALSE, makeRegion);
fn->iiAddCproc(libname, "findSharedObject", FALSE, findSharedObject);
fn->iiAddCproc(libname, "bindSharedObject", FALSE, bindSharedObject);
fn->iiAddCproc(libname, "typeSharedObject", FALSE, typeSharedObject);
fn->iiAddCproc(libname, "createThread", FALSE, createThread);
fn->iiAddCproc(libname, "joinThread", FALSE, joinThread);
fn->iiAddCproc(libname, "createThreadPool", FALSE, createThreadPool);
fn->iiAddCproc(libname, "createThreadPoolSet", FALSE, createThreadPoolSet);
#if 0
fn->iiAddCproc(libname, "adjoinThreadPool", FALSE, adjoinThreadPool);
fn->iiAddCproc(libname, "getAdjoinedThreadPools", FALSE, getAdjoinedThreadPools);
#endif
fn->iiAddCproc(libname, "closeThreadPool", FALSE, closeThreadPool);
fn->iiAddCproc(libname, "getThreadPoolWorkers", FALSE, getThreadPoolWorkers);
fn->iiAddCproc(libname, "setThreadPoolWorkers", FALSE, setThreadPoolWorkers);
fn->iiAddCproc(libname, "getThreadPoolConcurrency", FALSE, getThreadPoolConcurrency);
fn->iiAddCproc(libname, "setThreadPoolConcurrency", FALSE, setThreadPoolConcurrency);
fn->iiAddCproc(libname, "currentThreadPool", FALSE, currentThreadPool);
fn->iiAddCproc(libname, "setCurrentThreadPool", FALSE, setCurrentThreadPool);
fn->iiAddCproc(libname, "threadPoolExec", FALSE, threadPoolExec);
fn->iiAddCproc(libname, "threadID", FALSE, threadID);
fn->iiAddCproc(libname, "mainThread", FALSE, mainThread);
fn->iiAddCproc(libname, "threadEval", FALSE, threadEval);
fn->iiAddCproc(libname, "threadExec", FALSE, threadExec);
fn->iiAddCproc(libname, "threadResult", FALSE, threadResult);
fn->iiAddCproc(libname, "createJob", FALSE, createJob);
fn->iiAddCproc(libname, "currentJob", FALSE, currentJob);
fn->iiAddCproc(libname, "setSharedName", FALSE, setSharedName);
fn->iiAddCproc(libname, "getSharedName", FALSE, getSharedName);
fn->iiAddCproc(libname, "startJob", FALSE, startJob);
fn->iiAddCproc(libname, "waitJob", FALSE, waitJob);
fn->iiAddCproc(libname, "cancelJob", FALSE, cancelJob);
fn->iiAddCproc(libname, "jobCancelled", FALSE, jobCancelled);
fn->iiAddCproc(libname, "scheduleJob", FALSE, scheduleJob);
fn->iiAddCproc(libname, "scheduleJobs", FALSE, scheduleJob);
fn->iiAddCproc(libname, "createTrigger", FALSE, createTrigger);
fn->iiAddCproc(libname, "updateTrigger", FALSE, updateTrigger);
fn->iiAddCproc(libname, "testTrigger", FALSE, testTrigger);
fn->iiAddCproc(libname, "chainTrigger", FALSE, chainTrigger);
LinTree::init();
master_lock.unlock();
return MAX_TOK;
}
BOOLEAN
int BOOLEAN
Definition auxiliary.h:88
TRUE
#define TRUE
Definition auxiliary.h:101
FALSE
#define FALSE
Definition auxiliary.h:97
setBlackboxStuff
int setBlackboxStuff(blackbox *bb, const char *n)
define a new type
Definition blackbox.cc:143
l
int l
Definition cfEzgcd.cc:100
m
int m
Definition cfEzgcd.cc:128
i
int i
Definition cfEzgcd.cc:132
p
int p
Definition cfModGcd.cc:4086
false
return false
Definition cfModGcd.cc:85
b
CanonicalForm b
Definition cfModGcd.cc:4111
ConditionVariable
Definition thread.h:73
ConditionVariable::wait
void wait()
Definition thread.h:88
ConditionVariable::signal
void signal()
Definition thread.h:97
LibThread::AccTrigger
Definition shared.cc:1990
LibThread::AccTrigger::AccTrigger
AccTrigger(long count_init)
Definition shared.cc:1994
LibThread::AccTrigger::execute
virtual void execute()
Definition shared.cc:2012
LibThread::AccTrigger::count
long count
Definition shared.cc:1992
LibThread::AccTrigger::activate
virtual void activate(leftv arg)
Definition shared.cc:2003
LibThread::AccTrigger::accept
virtual bool accept(leftv arg)
Definition shared.cc:2000
LibThread::AccTrigger::ready
virtual bool ready()
Definition shared.cc:1996
LibThread::Command
Definition shared.cc:36
LibThread::Command::set_result
void set_result(int type, long n)
Definition shared.cc:127
LibThread::Command::test_arg
int test_arg(int i, int type)
Definition shared.cc:111
LibThread::Command::check_init
void check_init(int i, const char *err)
Definition shared.cc:80
LibThread::Command::~Command
~Command()
Definition shared.cc:61
LibThread::Command::argc
int argc
Definition shared.cc:42
LibThread::Command::report
void report(const char *err)
Definition shared.cc:106
LibThread::Command::abort
BOOLEAN abort(const char *err)
Definition shared.cc:143
LibThread::Command::no_result
void no_result()
Definition shared.cc:131
LibThread::Command::int_arg
long int_arg(int i)
Definition shared.cc:103
LibThread::Command::status
BOOLEAN status()
Definition shared.cc:137
LibThread::Command::check_arg
void check_arg(int i, int type, int type2, const char *err)
Definition shared.cc:86
LibThread::Command::ok
bool ok()
Definition shared.cc:134
LibThread::Command::args
leftv * args
Definition shared.cc:41
LibThread::Command::check_arg
void check_arg(int i, int type, const char *err)
Definition shared.cc:76
LibThread::Command::error
const char * error
Definition shared.cc:39
LibThread::Command::arg
void * arg(int i)
Definition shared.cc:96
LibThread::Command::result
leftv result
Definition shared.cc:40
LibThread::Command::set_result
void set_result(long n)
Definition shared.cc:115
LibThread::Command::name
const char * name
Definition shared.cc:38
LibThread::Command::check_argc_min
void check_argc_min(int n)
Definition shared.cc:72
LibThread::Command::nargs
int nargs()
Definition shared.cc:93
LibThread::Command::set_result
void set_result(int type, void *p)
Definition shared.cc:123
LibThread::Command::argtype
int argtype(int i)
Definition shared.cc:90
LibThread::Command::set_result
void set_result(const char *s)
Definition shared.cc:119
LibThread::Command::shared_arg
T * shared_arg(int i)
Definition shared.cc:100
LibThread::Command::check_argc
void check_argc(int n)
Definition shared.cc:64
LibThread::Command::check_argc
void check_argc(int lo, int hi)
Definition shared.cc:68
LibThread::Command::Command
Command(const char *n, leftv r, leftv a)
Definition shared.cc:44
LibThread::CountTrigger
Definition shared.cc:2029
LibThread::CountTrigger::count
long count
Definition shared.cc:2031
LibThread::CountTrigger::CountTrigger
CountTrigger(long count_init)
Definition shared.cc:2033
LibThread::CountTrigger::execute
virtual void execute()
Definition shared.cc:2047
LibThread::CountTrigger::activate
virtual void activate(leftv arg)
Definition shared.cc:2042
LibThread::CountTrigger::accept
virtual bool accept(leftv arg)
Definition shared.cc:2039
LibThread::CountTrigger::ready
virtual bool ready()
Definition shared.cc:2035
LibThread::EvalJob
Definition shared.cc:2355
LibThread::EvalJob::execute
virtual void execute()
Definition shared.cc:2358
LibThread::EvalJob::EvalJob
EvalJob()
Definition shared.cc:2357
LibThread::ExecJob
Definition shared.cc:2366
LibThread::ExecJob::ExecJob
ExecJob()
Definition shared.cc:2368
LibThread::ExecJob::execute
virtual void execute()
Definition shared.cc:2369
LibThread::InterpreterThread
Definition shared.cc:1419
LibThread::InterpreterThread::clearThreadState
void clearThreadState()
Definition shared.cc:1426
LibThread::InterpreterThread::getThreadState
ThreadState * getThreadState()
Definition shared.cc:1425
LibThread::InterpreterThread::~InterpreterThread
virtual ~InterpreterThread()
Definition shared.cc:1424
LibThread::InterpreterThread::ts
ThreadState * ts
Definition shared.cc:1421
LibThread::InterpreterThread::InterpreterThread
InterpreterThread(ThreadState *ts_init)
Definition shared.cc:1423
LibThread::Job
Definition shared.cc:1544
LibThread::Job::args
vector< string > args
Definition shared.cc:1553
LibThread::Job::deps
vector< Job * > deps
Definition shared.cc:1550
LibThread::Job::id
size_t id
Definition shared.cc:1548
LibThread::Job::done
bool done
Definition shared.cc:1557
LibThread::Job::cancelled
bool cancelled
Definition shared.cc:1560
LibThread::Job::run
void run()
Definition shared.cc:1977
LibThread::Job::pool
ThreadPool * pool
Definition shared.cc:1546
LibThread::Job::data
void * data
Definition shared.cc:1555
LibThread::Job::result
string result
Definition shared.cc:1554
LibThread::Job::running
bool running
Definition shared.cc:1559
LibThread::Job::fast
bool fast
Definition shared.cc:1556
LibThread::Job::Job
Job()
Definition shared.cc:1561
LibThread::Job::pending_index
long pending_index
Definition shared.cc:1549
LibThread::Job::~Job
~Job()
Definition shared.cc:1608
LibThread::Job::ready
virtual bool ready()
Definition shared.cc:1600
LibThread::Job::addDep
void addDep(Job *job)
Definition shared.cc:1566
LibThread::Job::execute
virtual void execute()=0
LibThread::Job::queued
bool queued
Definition shared.cc:1558
LibThread::Job::addNotify
void addNotify(vector< Job * > &jobs)
Definition shared.cc:1963
LibThread::Job::notify
vector< Job * > notify
Definition shared.cc:1551
LibThread::Job::triggers
vector< Trigger * > triggers
Definition shared.cc:1552
LibThread::Job::prio
long prio
Definition shared.cc:1547
LibThread::KernelJob
Definition shared.cc:2400
LibThread::KernelJob::KernelJob
KernelJob(void(*func)(leftv result, leftv arg))
Definition shared.cc:2404
LibThread::KernelJob::cfunc
void(* cfunc)(leftv result, leftv arg)
Definition shared.cc:2402
LibThread::KernelJob::execute
virtual void execute()
Definition shared.cc:2405
LibThread::ProcJob
Definition shared.cc:2376
LibThread::ProcJob::ProcJob
ProcJob(const char *procname_init)
Definition shared.cc:2379
LibThread::ProcJob::execute
virtual void execute()
Definition shared.cc:2383
LibThread::ProcJob::procname
string procname
Definition shared.cc:2377
LibThread::ProcTrigger
Definition shared.cc:2082
LibThread::ProcTrigger::accept
virtual bool accept(leftv arg)
Definition shared.cc:2093
LibThread::ProcTrigger::execute
virtual void execute()
Definition shared.cc:2122
LibThread::ProcTrigger::ProcTrigger
ProcTrigger(const char *p)
Definition shared.cc:2087
LibThread::ProcTrigger::success
bool success
Definition shared.cc:2085
LibThread::ProcTrigger::procname
string procname
Definition shared.cc:2084
LibThread::ProcTrigger::activate
virtual void activate(leftv arg)
Definition shared.cc:2096
LibThread::ProcTrigger::ready
virtual bool ready()
Definition shared.cc:2089
LibThread::RawKernelJob
Definition shared.cc:2429
LibThread::RawKernelJob::execute
virtual void execute()
Definition shared.cc:2434
LibThread::RawKernelJob::cfunc
void(* cfunc)(long ndeps, Job **deps)
Definition shared.cc:2431
LibThread::RawKernelJob::RawKernelJob
RawKernelJob(void(*func)(long ndeps, Job **deps))
Definition shared.cc:2433
LibThread::Region
Definition shared.cc:200
LibThread::Region::region_lock
Lock region_lock
Definition shared.cc:202
LibThread::Region::Region
Region()
Definition shared.cc:205
LibThread::Region::unlock
void unlock()
Definition shared.cc:212
LibThread::Region::objects
SharedObjectTable objects
Definition shared.cc:204
LibThread::Region::get_lock
Lock * get_lock()
Definition shared.cc:207
LibThread::Region::~Region
virtual ~Region()
Definition shared.cc:206
LibThread::Region::lock
void lock()
Definition shared.cc:208
LibThread::Region::is_locked
int is_locked()
Definition shared.cc:216
LibThread::Scheduler
Definition shared.cc:1649
LibThread::Scheduler::maxconcurrency
int maxconcurrency
Definition shared.cc:1654
LibThread::Scheduler::~Scheduler
virtual ~Scheduler()
Definition shared.cc:1691
LibThread::Scheduler::Scheduler
Scheduler(int n)
Definition shared.cc:1668
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::addThread
void addThread(ThreadPool *owner, ThreadState *thread)
Definition shared.cc:1730
LibThread::Scheduler::response
ConditionVariable response
Definition shared.cc:1664
LibThread::Scheduler::shutting_down
bool shutting_down
Definition shared.cc:1656
LibThread::Scheduler::cancelDeps
void cancelDeps(Job *job)
Definition shared.cc:1781
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::getThread
ThreadState * getThread(int i)
Definition shared.cc:1703
LibThread::Scheduler::set_maxconcurrency
void set_maxconcurrency(int n)
Definition shared.cc:1677
LibThread::Scheduler::detachJob
void detachJob(Job *job)
Definition shared.cc:1753
LibThread::Scheduler::broadcastJob
void broadcastJob(ThreadPool *pool, Job *job)
Definition shared.cc:1771
LibThread::Scheduler::threads
vector< ThreadState * > threads
Definition shared.cc:1658
LibThread::Scheduler::nthreads
int nthreads
Definition shared.cc:1653
LibThread::Scheduler::cancelJob
void cancelJob(Job *job)
Definition shared.cc:1790
LibThread::Scheduler::waitJob
void waitJob(Job *job)
Definition shared.cc:1801
LibThread::Scheduler::get_maxconcurrency
int get_maxconcurrency()
Definition shared.cc:1680
LibThread::Scheduler::Job
friend class Job
Definition shared.cc:1665
LibThread::Scheduler::notifyDeps
static void notifyDeps(Scheduler *scheduler, Job *job)
Definition shared.cc:1824
LibThread::Scheduler::jobid
size_t jobid
Definition shared.cc:1652
LibThread::Scheduler::clearThreadState
void clearThreadState()
Definition shared.cc:1821
LibThread::Scheduler::queueJob
void queueJob(Job *job)
Definition shared.cc:1765
LibThread::Scheduler::main
static void * main(ThreadState *ts, void *arg)
Definition shared.cc:1851
LibThread::Scheduler::shutdown
void shutdown(bool wait)
Definition shared.cc:1704
LibThread::Scheduler::running
int running
Definition shared.cc:1655
LibThread::Scheduler::shutdown_counter
int shutdown_counter
Definition shared.cc:1657
LibThread::Scheduler::attachJob
void attachJob(ThreadPool *pool, Job *job)
Definition shared.cc:1737
LibThread::Scheduler::pending
vector< Job * > pending
Definition shared.cc:1662
LibThread::Scheduler::threadpool_size
int threadpool_size(ThreadPool *pool)
Definition shared.cc:1683
LibThread::SetTrigger
Definition shared.cc:2052
LibThread::SetTrigger::execute
virtual void execute()
Definition shared.cc:2076
LibThread::SetTrigger::count
long count
Definition shared.cc:2055
LibThread::SetTrigger::activate
virtual void activate(leftv arg)
Definition shared.cc:2067
LibThread::SetTrigger::SetTrigger
SetTrigger(long count_init)
Definition shared.cc:2057
LibThread::SetTrigger::ready
virtual bool ready()
Definition shared.cc:2060
LibThread::SetTrigger::set
vector< bool > set
Definition shared.cc:2054
LibThread::SetTrigger::accept
virtual bool accept(leftv arg)
Definition shared.cc:2064
LibThread::SharedObject
Definition shared.cc:149
LibThread::SharedObject::refcount
long refcount
Definition shared.cc:152
LibThread::SharedObject::name
std::string name
Definition shared.cc:154
LibThread::SharedObject::decref
long decref()
Definition shared.cc:170
LibThread::SharedObject::get_type
int get_type()
Definition shared.cc:159
LibThread::SharedObject::~SharedObject
virtual ~SharedObject()
Definition shared.cc:157
LibThread::SharedObject::set_name
void set_name(std::string &name_init)
Definition shared.cc:160
LibThread::SharedObject::getref
long getref()
Definition shared.cc:177
LibThread::SharedObject::type
int type
Definition shared.cc:153
LibThread::SharedObject::op3
virtual BOOLEAN op3(int op, leftv res, leftv a1, leftv a2, leftv a3)
Definition shared.cc:183
LibThread::SharedObject::set_name
void set_name(const char *s)
Definition shared.cc:161
LibThread::SharedObject::incref
void incref(int by=1)
Definition shared.cc:165
LibThread::SharedObject::lock
Lock lock
Definition shared.cc:151
LibThread::SharedObject::op2
virtual BOOLEAN op2(int op, leftv res, leftv a1, leftv a2)
Definition shared.cc:180
LibThread::SharedObject::SharedObject
SharedObject()
Definition shared.cc:156
LibThread::SharedObject::set_type
void set_type(int type_init)
Definition shared.cc:158
LibThread::SharedObject::get_name
std::string & get_name()
Definition shared.cc:164
LibThread::SingularChannel
Definition shared.cc:392
LibThread::SingularChannel::cond
ConditionVariable cond
Definition shared.cc:396
LibThread::SingularChannel::lock
Lock lock
Definition shared.cc:395
LibThread::SingularChannel::q
queue< string > q
Definition shared.cc:394
LibThread::SingularChannel::send
void send(string item)
Definition shared.cc:400
LibThread::SingularChannel::~SingularChannel
virtual ~SingularChannel()
Definition shared.cc:399
LibThread::SingularChannel::count
long count()
Definition shared.cc:418
LibThread::SingularChannel::SingularChannel
SingularChannel()
Definition shared.cc:398
LibThread::SingularChannel::receive
string receive()
Definition shared.cc:406
LibThread::SingularSyncVar
Definition shared.cc:426
LibThread::SingularSyncVar::acquire
void acquire()
Definition shared.cc:435
LibThread::SingularSyncVar::write
int write(string item)
Definition shared.cc:454
LibThread::SingularSyncVar::~SingularSyncVar
virtual ~SingularSyncVar()
Definition shared.cc:434
LibThread::SingularSyncVar::cond
ConditionVariable cond
Definition shared.cc:431
LibThread::SingularSyncVar::SingularSyncVar
SingularSyncVar()
Definition shared.cc:433
LibThread::SingularSyncVar::wait_init
void wait_init()
Definition shared.cc:441
LibThread::SingularSyncVar::get
leftv get()
Definition shared.cc:445
LibThread::SingularSyncVar::value
string value
Definition shared.cc:428
LibThread::SingularSyncVar::init
int init
Definition shared.cc:429
LibThread::SingularSyncVar::check
int check()
Definition shared.cc:474
LibThread::SingularSyncVar::update
void update(leftv val)
Definition shared.cc:449
LibThread::SingularSyncVar::release
void release()
Definition shared.cc:438
LibThread::SingularSyncVar::lock
Lock lock
Definition shared.cc:430
LibThread::SingularSyncVar::read
string read()
Definition shared.cc:466
LibThread::ThreadPool
Definition shared.cc:1628
LibThread::ThreadPool::attachJob
void attachJob(Job *job)
Definition shared.cc:1928
LibThread::ThreadPool::~ThreadPool
~ThreadPool()
Definition shared.cc:1920
LibThread::ThreadPool::waitJob
void waitJob(Job *job)
Definition shared.cc:1946
LibThread::ThreadPool::queueJob
void queueJob(Job *job)
Definition shared.cc:1934
LibThread::ThreadPool::broadcastJob
void broadcastJob(Job *job)
Definition shared.cc:1937
LibThread::ThreadPool::getThread
ThreadState * getThread(int i)
Definition shared.cc:1923
LibThread::ThreadPool::cancelJob
void cancelJob(Job *job)
Definition shared.cc:1943
LibThread::ThreadPool::clearThreadState
void clearThreadState()
Definition shared.cc:1949
LibThread::ThreadPool::nthreads
int nthreads
Definition shared.cc:1631
LibThread::ThreadPool::scheduler
Scheduler * scheduler
Definition shared.cc:1630
LibThread::ThreadPool::ThreadPool
ThreadPool(Scheduler *sched, int n)
Definition shared.cc:1916
LibThread::ThreadPool::detachJob
void detachJob(Job *job)
Definition shared.cc:1931
LibThread::ThreadPool::cancelDeps
void cancelDeps(Job *job)
Definition shared.cc:1940
LibThread::ThreadPool::shutdown
void shutdown(bool wait)
Definition shared.cc:1924
LibThread::ThreadPool::addThread
void addThread(ThreadState *thread)
Definition shared.cc:1925
LibThread::ThreadState
Definition shared.cc:1331
LibThread::ThreadState::id
pthread_t id
Definition shared.cc:1338
LibThread::ThreadState::thread_func
void *(* thread_func)(ThreadState *, void *)
Definition shared.cc:1336
LibThread::ThreadState::running
bool running
Definition shared.cc:1334
LibThread::ThreadState::lock
Lock lock
Definition shared.cc:1340
LibThread::ThreadState::arg
void * arg
Definition shared.cc:1337
LibThread::ThreadState::~ThreadState
~ThreadState()
Definition shared.cc:1351
LibThread::ThreadState::ThreadState
ThreadState()
Definition shared.cc:1345
LibThread::ThreadState::result
void * result
Definition shared.cc:1337
LibThread::ThreadState::parent
pthread_t parent
Definition shared.cc:1339
LibThread::ThreadState::index
int index
Definition shared.cc:1335
LibThread::ThreadState::to_cond
ConditionVariable to_cond
Definition shared.cc:1341
LibThread::ThreadState::from_thread
queue< string > from_thread
Definition shared.cc:1344
LibThread::ThreadState::from_cond
ConditionVariable from_cond
Definition shared.cc:1342
LibThread::ThreadState::to_thread
queue< string > to_thread
Definition shared.cc:1343
LibThread::ThreadState::active
bool active
Definition shared.cc:1333
LibThread::Transactional::tx_end
void tx_end()
Definition shared.cc:296
LibThread::Transactional::Transactional
Transactional()
Definition shared.cc:301
LibThread::Transactional::~Transactional
virtual ~Transactional()
Definition shared.cc:312
LibThread::Transactional::lock
Lock * lock
Definition shared.cc:284
LibThread::Transactional::set_region
void set_region(Region *region_init)
Definition shared.cc:304
LibThread::Transactional::region
Region * region
Definition shared.cc:283
LibThread::Transactional::tx_begin
int tx_begin()
Definition shared.cc:286
LibThread::Trigger
Definition shared.cc:1593
LibThread::Trigger::activate
virtual void activate(leftv arg)=0
LibThread::Trigger::accept
virtual bool accept(leftv arg)=0
LibThread::Trigger::Trigger
Trigger()
Definition shared.cc:1597
LibThread::TxList
Definition shared.cc:352
LibThread::TxList::~TxList
virtual ~TxList()
Definition shared.cc:357
LibThread::TxList::size
long size()
Definition shared.cc:383
LibThread::TxList::put
int put(size_t index, string &value)
Definition shared.cc:358
LibThread::TxList::TxList
TxList()
Definition shared.cc:356
LibThread::TxList::entries
vector< string > entries
Definition shared.cc:354
LibThread::TxList::get
int get(size_t index, string &value)
Definition shared.cc:372
LibThread::TxTable
Definition shared.cc:315
LibThread::TxTable::put
int put(string &key, string &value)
Definition shared.cc:321
LibThread::TxTable::entries
std::map< string, string > entries
Definition shared.cc:317
LibThread::TxTable::TxTable
TxTable()
Definition shared.cc:319
LibThread::TxTable::get
int get(string &key, string &value)
Definition shared.cc:333
LibThread::TxTable::check
int check(string &key)
Definition shared.cc:343
LibThread::TxTable::~TxTable
virtual ~TxTable()
Definition shared.cc:320
LinTree::LinTree
Definition lintree.h:24
LinTree::LinTree::get
T get()
Definition lintree.h:48
LinTree::LinTree::put
void put(T data)
Definition lintree.h:61
LinTree::LinTree::get_prev
T get_prev()
Definition lintree.h:55
Lock
Definition thread.h:17
Lock::is_locked
bool is_locked()
Definition thread.h:68
Lock::lock
void lock()
Definition thread.h:46
Lock::unlock
void unlock()
Definition thread.h:57
sleftv
Class used for (list of) interpreter objects.
Definition subexpr.h:83
sleftv::Typ
int Typ()
Definition subexpr.cc:1048
sleftv::name
const char * name
Definition subexpr.h:87
sleftv::req_packhdl
package req_packhdl
Definition subexpr.h:106
sleftv::rtyp
int rtyp
Definition subexpr.h:91
sleftv::Data
void * Data()
Definition subexpr.cc:1192
sleftv::next
leftv next
Definition subexpr.h:86
sleftv::Eval
int Eval()
Definition subexpr.cc:2000
sleftv::Copy
void Copy(leftv e)
Definition subexpr.cc:689
sleftv::data
void * data
Definition subexpr.h:88
sleftv::CleanUp
void CleanUp(ring r=currRing)
Definition subexpr.cc:351
slists::m
sleftv * m
Definition lists.h:46
slists::Init
INLINE_THIS void Init(int l=0)
result
return result
Definition facAbsBiFact.cc:76
s
const CanonicalForm int s
Definition facAbsFact.cc:51
res
CanonicalForm res
Definition facAbsFact.cc:60
eval
CFList & eval
Definition facFactorize.cc:47
j
int j
Definition facHensel.cc:110
name
char name(const Variable &v)
Definition factory.h:189
WerrorS
void WerrorS(const char *s)
Definition feFopen.cc:24
feOptIndex
feOptIndex
Definition feOptGen.h:15
feGetOptIndex
feOptIndex feGetOptIndex(const char *name)
Definition feOpt.cc:104
feSetOptValue
const char * feSetOptValue(feOptIndex opt, char *optarg)
Definition feOpt.cc:154
feOpt.h
Tok2Cmdname
const char * Tok2Cmdname(int tok)
Definition gentable.cc:137
STATIC_VAR
#define STATIC_VAR
Definition globaldefs.h:7
VAR
#define VAR
Definition globaldefs.h:5
iiExprArithM
BOOLEAN iiExprArithM(leftv res, leftv a, int op)
Definition iparith.cc:9666
basePack
VAR package basePack
Definition ipid.cc:56
currPack
VAR package currPack
Definition ipid.cc:55
sleftv_bin
EXTERN_VAR omBin sleftv_bin
Definition ipid.h:145
IDDATA
#define IDDATA(a)
Definition ipid.h:126
SModulFunctions
Definition ipid.h:69
T
STATIC_VAR jList * T
Definition janet.cc:30
next
ListNode * next
Definition janet.h:31
info
#define info
Definition libparse.cc:1256
libsingular.h
siInit
void siInit(char *)
Definition misc_ip.cc:1364
lintree.h
slists_bin
VAR omBin slists_bin
Definition lists.cc:23
lSize
int lSize(lists L)
Definition lists.cc:25
mod2.h
error
#define error(a)
Definition mpr_numeric.cc:966
lists
slists * lists
Definition mpr_numeric.h:146
LibThread
Definition bytebuf.h:8
LibThread::readSyncVar
BOOLEAN readSyncVar(leftv result, leftv arg)
Definition shared.cc:1233
LibThread::str
char * str(leftv arg)
Definition shared.cc:699
LibThread::getThreadPoolWorkers
static BOOLEAN getThreadPoolWorkers(leftv result, leftv arg)
Definition shared.cc:2244
LibThread::retain
void retain(Job *job)
LibThread::getTable
BOOLEAN getTable(leftv result, leftv arg)
Definition shared.cc:936
LibThread::type_thread
int type_thread
Definition shared.cc:236
LibThread::getThreadPoolConcurrency
static BOOLEAN getThreadPoolConcurrency(leftv result, leftv arg)
Definition shared.cc:2276
LibThread::type_region
int type_region
Definition shared.cc:228
LibThread::createThread
ThreadState * createThread(void *(*thread_func)(ThreadState *, void *), void *arg)
Definition shared.cc:1464
LibThread::getCurrentJob
Job * getCurrentJob()
Definition shared.cc:2514
LibThread::consSyncVar
SharedObject * consSyncVar()
Definition shared.cc:715
LibThread::not_a_uri
int not_a_uri(const char *name, leftv arg)
Definition shared.cc:682
LibThread::setThreadPoolConcurrency
static BOOLEAN setThreadPoolConcurrency(leftv result, leftv arg)
Definition shared.cc:2291
LibThread::scheduleJob
static BOOLEAN scheduleJob(leftv result, leftv arg)
Definition shared.cc:2789
LibThread::makeSyncVar
BOOLEAN makeSyncVar(leftv result, leftv arg)
Definition shared.cc:849
LibThread::shared_check_assign
BOOLEAN shared_check_assign(blackbox *b, leftv l, leftv r)
Definition shared.cc:568
LibThread::makeSharedList
BOOLEAN makeSharedList(leftv result, leftv arg)
Definition shared.cc:819
LibThread::shared_assign
BOOLEAN shared_assign(leftv l, leftv r)
Definition shared.cc:519
LibThread::new_shared
void * new_shared(SharedObject *obj)
Definition shared.cc:486
LibThread::ref_shared
void ref_shared(LinTree::LinTree &lintree, int by)
Definition shared.cc:1285
LibThread::getList
BOOLEAN getList(leftv result, leftv arg)
Definition shared.cc:1023
LibThread::global_objects_lock
Lock global_objects_lock
Definition shared.cc:221
LibThread::type_atomic_list
int type_atomic_list
Definition shared.cc:234
LibThread::appendArgCopy
static void appendArgCopy(vector< leftv > &argv, leftv arg)
Definition shared.cc:737
LibThread::shared_init
void * shared_init(blackbox *b)
Definition shared.cc:482
LibThread::statChannel
BOOLEAN statChannel(leftv result, leftv arg)
Definition shared.cc:1166
LibThread::threadEval
BOOLEAN threadEval(leftv result, leftv arg)
Definition shared.cc:2922
LibThread::shared_copy
void * shared_copy(blackbox *b, void *d)
Definition shared.cc:510
LibThread::SharedObjectPtr
SharedObject * SharedObjectPtr
Definition shared.cc:241
LibThread::getJobCancelled
bool getJobCancelled()
Definition shared.cc:2640
LibThread::unlockRegion
BOOLEAN unlockRegion(leftv result, leftv arg)
Definition shared.cc:1115
LibThread::sendChannel
BOOLEAN sendChannel(leftv result, leftv arg)
Definition shared.cc:1130
LibThread::joinThread
void * joinThread(ThreadState *ts)
Definition shared.cc:1469
LibThread::startJob
Job * startJob(ThreadPool *pool, Job *job, leftv arg)
Definition shared.cc:2476
LibThread::newThread
static ThreadState * newThread(void *(*thread_func)(ThreadState *, void *), void *arg, const char **error)
Definition shared.cc:1431
LibThread::setSharedName
BOOLEAN setSharedName(leftv result, leftv arg)
Definition shared.cc:3033
LibThread::appendArg
static void appendArg(vector< leftv > &argv, string &s)
Definition shared.cc:723
LibThread::makeSharedObject
SharedObject * makeSharedObject(SharedObjectTable &table, Lock *lock, int type, string &name, SharedConstructor scons)
Definition shared.cc:244
LibThread::writeSyncVar
BOOLEAN writeSyncVar(leftv result, leftv arg)
Definition shared.cc:1184
LibThread::SharedObjectTable
std::map< std::string, SharedObject * > SharedObjectTable
Definition shared.cc:198
LibThread::makeAtomicTable
BOOLEAN makeAtomicTable(leftv result, leftv arg)
Definition shared.cc:773
LibThread::bindSharedObject
BOOLEAN bindSharedObject(leftv result, leftv arg)
Definition shared.cc:919
LibThread::report
void report(const char *fmt, const char *name)
Definition shared.cc:661
LibThread::shared_string
char * shared_string(blackbox *b, void *d)
Definition shared.cc:590
LibThread::lockRegion
BOOLEAN lockRegion(leftv result, leftv arg)
Definition shared.cc:1083
LibThread::currentJob
BOOLEAN currentJob(leftv result, leftv arg)
Definition shared.cc:2893
LibThread::thread_counter
long thread_counter
Definition shared.cc:226
LibThread::wrong_num_args
int wrong_num_args(const char *name, leftv arg, int n)
Definition shared.cc:667
LibThread::thread_state
ThreadState * thread_state
Definition shared.cc:1361
LibThread::createInterpreterThread
static InterpreterThread * createInterpreterThread(const char **error)
Definition shared.cc:1480
LibThread::type_threadpool
int type_threadpool
Definition shared.cc:237
LibThread::name_lock
Lock name_lock(true)
LibThread::type_job
int type_job
Definition shared.cc:238
LibThread::testTrigger
static BOOLEAN testTrigger(leftv result, leftv arg)
Definition shared.cc:2773
LibThread::createThreadPool
ThreadPool * createThreadPool(int threads, int prioThreads=0)
Definition shared.cc:2213
LibThread::threadExec
BOOLEAN threadExec(leftv result, leftv arg)
Definition shared.cc:2950
LibThread::rlock_assign
BOOLEAN rlock_assign(leftv l, leftv r)
Definition shared.cc:544
LibThread::have_threads
const int have_threads
Definition shared.cc:33
LibThread::setJobData
void setJobData(Job *job, void *data)
Definition shared.cc:2644
LibThread::putTable
BOOLEAN putTable(leftv result, leftv arg)
Definition shared.cc:996
LibThread::makeAtomicList
BOOLEAN makeAtomicList(leftv result, leftv arg)
Definition shared.cc:787
LibThread::getCurrentThreadPool
ThreadPool * getCurrentThreadPool()
Definition shared.cc:2240
LibThread::currentJobRef
STATIC_VAR Job * currentJobRef
Definition shared.cc:1626
LibThread::createTrigger
static BOOLEAN createTrigger(leftv result, leftv arg)
Definition shared.cc:2687
LibThread::not_a_region
int not_a_region(const char *name, leftv arg)
Definition shared.cc:690
LibThread::currentThreadPool
BOOLEAN currentThreadPool(leftv result, leftv arg)
Definition shared.cc:2329
LibThread::addJobArgs
void addJobArgs(Job *job, leftv arg)
Definition shared.cc:2660
LibThread::rlock_destroy
void rlock_destroy(blackbox *b, void *d)
Definition shared.cc:501
LibThread::type_atomic_table
int type_atomic_table
Definition shared.cc:232
LibThread::interpreter_thread
void * interpreter_thread(ThreadState *ts, void *arg)
Definition shared.cc:1386
LibThread::setThreadPoolWorkers
static BOOLEAN setThreadPoolWorkers(leftv result, leftv arg)
Definition shared.cc:2261
LibThread::type_syncvar
int type_syncvar
Definition shared.cc:231
LibThread::updateTrigger
static BOOLEAN updateTrigger(leftv result, leftv arg)
Definition shared.cc:2730
LibThread::getSharedName
BOOLEAN getSharedName(leftv result, leftv arg)
Definition shared.cc:3051
LibThread::global_objects
SharedObjectTable global_objects
Definition shared.cc:222
LibThread::consList
SharedObject * consList()
Definition shared.cc:707
LibThread::receiveChannel
BOOLEAN receiveChannel(leftv result, leftv arg)
Definition shared.cc:1147
LibThread::mainThread
BOOLEAN mainThread(leftv result, leftv arg)
Definition shared.cc:2914
LibThread::releaseShared
void releaseShared(SharedObject *obj)
Definition shared.cc:192
LibThread::getJobData
void * getJobData(Job *job)
Definition shared.cc:2652
LibThread::closeThreadPool
void closeThreadPool(ThreadPool *pool, bool wait)
Definition shared.cc:2324
LibThread::SchedInfo::job
Job * job
Definition shared.cc:1621
LibThread::type_shared_table
int type_shared_table
Definition shared.cc:233
LibThread::consChannel
SharedObject * consChannel()
Definition shared.cc:711
LibThread::type_channel
int type_channel
Definition shared.cc:230
LibThread::threadPoolExec
BOOLEAN threadPoolExec(leftv result, leftv arg)
Definition shared.cc:2978
LibThread::type_shared_list
int type_shared_list
Definition shared.cc:235
LibThread::setCurrentThreadPool
BOOLEAN setCurrentThreadPool(leftv result, leftv arg)
Definition shared.cc:2341
LibThread::regionLock
BOOLEAN regionLock(leftv result, leftv arg)
Definition shared.cc:1098
LibThread::threadID
BOOLEAN threadID(leftv result, leftv arg)
Definition shared.cc:2906
LibThread::makeRegionlockType
void makeRegionlockType(int &type, const char *name)
Definition shared.cc:1316
LibThread::rlock_string
char * rlock_string(blackbox *b, void *d)
Definition shared.cc:652
LibThread::encode_shared
void encode_shared(LinTree::LinTree &lintree, leftv val)
Definition shared.cc:1270
LibThread::makeSharedTable
BOOLEAN makeSharedTable(leftv result, leftv arg)
Definition shared.cc:801
LibThread::typeSharedObject
BOOLEAN typeSharedObject(leftv result, leftv arg)
Definition shared.cc:888
LibThread::createJob
Job * createJob(void(*func)(leftv result, leftv arg))
Definition shared.cc:2466
LibThread::release
void release(Job *job)
LibThread::SchedInfo::num
int num
Definition shared.cc:1622
LibThread::thread_id
VAR long thread_id
Definition shared.cc:225
LibThread::currentThreadPoolRef
STATIC_VAR ThreadPool * currentThreadPoolRef
Definition shared.cc:1625
LibThread::SchedInfo::scheduler
Scheduler * scheduler
Definition shared.cc:1620
LibThread::statSyncVar
BOOLEAN statSyncVar(leftv result, leftv arg)
Definition shared.cc:1252
LibThread::setOption
void setOption(int ch)
Definition shared.cc:1363
LibThread::JobQueue
queue< Job * > JobQueue
Definition shared.cc:1615
LibThread::shared_op2
BOOLEAN shared_op2(int op, leftv res, leftv a1, leftv a2)
Definition shared.cc:580
LibThread::master_lock
Lock master_lock(true)
LibThread::findSharedObject
SharedObject * findSharedObject(SharedObjectTable &table, Lock *lock, string &name)
Definition shared.cc:266
LibThread::SharedConstructor
SharedObjectPtr(* SharedConstructor)()
Definition shared.cc:242
LibThread::type_regionlock
int type_regionlock
Definition shared.cc:229
LibThread::type_trigger
int type_trigger
Definition shared.cc:239
LibThread::makeChannel
BOOLEAN makeChannel(leftv result, leftv arg)
Definition shared.cc:836
LibThread::shared_op3
BOOLEAN shared_op3(int op, leftv res, leftv a1, leftv a2, leftv a3)
Definition shared.cc:585
LibThread::thread_main
void * thread_main(void *arg)
Definition shared.cc:1380
LibThread::getJobName
const char * getJobName()
LibThread::acquireShared
void acquireShared(SharedObject *obj)
Definition shared.cc:188
LibThread::thread_init
void thread_init()
Definition shared.cc:1368
LibThread::jobCancelled
static BOOLEAN jobCancelled(leftv result, leftv arg)
Definition shared.cc:2607
LibThread::consTable
SharedObject * consTable()
Definition shared.cc:703
LibThread::chainTrigger
static BOOLEAN chainTrigger(leftv result, leftv arg)
Definition shared.cc:2752
LibThread::makeRegion
BOOLEAN makeRegion(leftv result, leftv arg)
Definition shared.cc:862
LibThread::putList
BOOLEAN putList(leftv result, leftv arg)
Definition shared.cc:1056
LibThread::joinInterpreterThread
static bool joinInterpreterThread(InterpreterThread *thread)
Definition shared.cc:1507
LibThread::setJobName
void setJobName(const char *)
LibThread::createThreadPoolSet
static BOOLEAN createThreadPoolSet(leftv result, leftv arg)
Definition shared.cc:2161
LibThread::makeSharedType
void makeSharedType(int &type, const char *name)
Definition shared.cc:1301
LibThread::shared_destroy
void shared_destroy(blackbox *b, void *d)
Definition shared.cc:493
LibThread::executeProc
static BOOLEAN executeProc(sleftv &result, const char *procname, const vector< leftv > &argv)
Definition shared.cc:744
LibThread::consRegion
SharedObject * consRegion()
Definition shared.cc:719
LibThread::updateSyncVar
BOOLEAN updateSyncVar(leftv result, leftv arg)
Definition shared.cc:1204
LibThread::inTable
BOOLEAN inTable(leftv result, leftv arg)
Definition shared.cc:969
LibThread::getJobResult
leftv getJobResult(Job *job)
Definition shared.cc:2670
LibThread::thread_lock
Lock thread_lock
Definition shared.cc:1359
LibThread::installShared
void installShared(int type)
Definition shared.cc:1297
LibThread::threadResult
BOOLEAN threadResult(leftv result, leftv arg)
Definition shared.cc:3002
LibThread::cancelJob
void cancelJob(Job *job)
Definition shared.cc:2509
LibThread::waitJob
void waitJob(Job *job)
Definition shared.cc:2585
LibThread::decode_shared
leftv decode_shared(LinTree::LinTree &lintree)
Definition shared.cc:1276
LibThread::SchedInfo
Definition shared.cc:1619
LinTree::init
void init()
Definition lintree.cc:864
LinTree::to_string
std::string to_string(leftv val)
Definition lintree.cc:843
LinTree::install
void install(int typ, LinTreeEncodeFunc enc, LinTreeDecodeFunc dec, LinTreeRefFunc ref)
Definition lintree.cc:51
LinTree::from_string
leftv from_string(std::string &str)
Definition lintree.cc:854
omStrDup
#define omStrDup(s)
Definition omAllocDecl.h:263
omAlloc0Bin
#define omAlloc0Bin(bin)
Definition omAllocDecl.h:206
omFree
#define omFree(addr)
Definition omAllocDecl.h:261
omAlloc0
#define omAlloc0(size)
Definition omAllocDecl.h:211
omFreeBin
#define omFreeBin(addr, bin)
Definition omAllocDecl.h:259
NULL
#define NULL
Definition omList.c:12
index
static int index(p_Length length, p_Ord ord)
Definition p_Procs_Impl.h:592
Werror
void Werror(const char *fmt,...)
Definition reporter.cc:189
idhdl
idrec * idhdl
Definition ring.h:22
systhreads
int SI_MOD_INIT systhreads(SModulFunctions *fn)
Definition shared.cc:3073
MAX_THREADS
#define MAX_THREADS
Definition shared.cc:1329
pSingular_initialize_thread
void pSingular_initialize_thread()
buf
int status int void * buf
Definition si_signals.h:69
wait
wait
Definition si_signals.h:61
singthreads.h
LibThread::JobCompare
Definition shared.cc:1578
LibThread::JobCompare::operator()
bool operator()(const Job *lhs, const Job *rhs)
Definition shared.cc:1579
leftv
sleftv * leftv
Definition structs.h:53
assert
#define assert(A)
Definition svd_si.h:3
thread.h
threadconf.h
IDHDL
#define IDHDL
Definition tok.h:31
LIST_CMD
@ LIST_CMD
Definition tok.h:118
DEF_CMD
@ DEF_CMD
Definition tok.h:58
STRING_CMD
@ STRING_CMD
Definition tok.h:187
INT_CMD
@ INT_CMD
Definition tok.h:96
MAX_TOK
@ MAX_TOK
Definition tok.h:220
NONE
#define NONE
Definition tok.h:223
COMMAND
#define COMMAND
Definition tok.h:29