LibThread Namespace Reference

Data Structures
class	AccTrigger
class	ByteBuf
class	Channel
class	Command
class	CountTrigger
class	EvalJob
class	ExecJob
class	InterpreterThread
class	Job
struct	JobCompare
class	KernelJob
class	ProcJob
class	ProcTrigger
class	RawKernelJob
class	Region
struct	SchedInfo
class	Scheduler
class	SetTrigger
class	SharedObject
class	SingularChannel
class	SingularSyncVar
class	SyncVar
class	ThreadPool
class	ThreadState
class	Transactional
class	Trigger
class	TxList
class	TxTable

Typedefs
typedef std::map< std::string, SharedObject * >	SharedObjectTable
typedef SharedObject *	SharedObjectPtr
typedef SharedObjectPtr(*	SharedConstructor) ()
typedef queue< Job * >	JobQueue
typedef void *(*	ThreadFunc) (ThreadState *, void *)

Functions
char *	allocate_space (size_t n)
void	free_space (size_t n, char *p)
ThreadPool *	createThreadPool (int threads, int prioThreads=0)
void	closeThreadPool (ThreadPool *pool, bool wait)
ThreadPool *	getCurrentThreadPool ()
Job *	createJob (void(*func)(leftv result, leftv arg))
Job *	createJob (void(*func)(long ndeps, Job **deps))
Job *	getCurrentJob ()
bool	getJobCancelled ()
bool	getJobCancelled (Job *job)
void	addJobArgs (Job *job, leftv arg)
void	setJobData (Job *job, void *data)
void *	getJobData (Job *job)
leftv	getJobResult (Job *job)
const char *	getJobName ()
void	setJobName (const char *)
Job *	startJob (ThreadPool *pool, Job *job, leftv arg)
Job *	startJob (ThreadPool *pool, Job *job)
void	cancelJob (Job *job)
void	waitJob (Job *job)
void	release (Job *job)
void	release (ThreadPool *pool)
void	retain (Job *job)
void	retain (ThreadPool *pool)
void	acquireShared (SharedObject *obj)
void	releaseShared (SharedObject *obj)
SharedObject *	makeSharedObject (SharedObjectTable &table, Lock *lock, int type, string &name, SharedConstructor scons)
SharedObject *	findSharedObject (SharedObjectTable &table, Lock *lock, string &name)
void *	shared_init (blackbox *b)
void *	new_shared (SharedObject *obj)
void	shared_destroy (blackbox *b, void *d)
void	rlock_destroy (blackbox *b, void *d)
void *	shared_copy (blackbox *b, void *d)
BOOLEAN	shared_assign (leftv l, leftv r)
BOOLEAN	rlock_assign (leftv l, leftv r)
BOOLEAN	shared_check_assign (blackbox *b, leftv l, leftv r)
BOOLEAN	shared_op2 (int op, leftv res, leftv a1, leftv a2)
BOOLEAN	shared_op3 (int op, leftv res, leftv a1, leftv a2, leftv a3)
char *	shared_string (blackbox *b, void *d)
char *	rlock_string (blackbox *b, void *d)
void	report (const char *fmt, const char *name)
int	wrong_num_args (const char *name, leftv arg, int n)
int	not_a_uri (const char *name, leftv arg)
int	not_a_region (const char *name, leftv arg)
char *	str (leftv arg)
SharedObject *	consTable ()
SharedObject *	consList ()
SharedObject *	consChannel ()
SharedObject *	consSyncVar ()
SharedObject *	consRegion ()
static void	appendArg (vector< leftv > &argv, string &s)
static void	appendArg (vector< leftv > &argv, leftv arg)
static void	appendArgCopy (vector< leftv > &argv, leftv arg)
static BOOLEAN	executeProc (sleftv &result, const char *procname, const vector< leftv > &argv)
BOOLEAN	makeAtomicTable (leftv result, leftv arg)
BOOLEAN	makeAtomicList (leftv result, leftv arg)
BOOLEAN	makeSharedTable (leftv result, leftv arg)
BOOLEAN	makeSharedList (leftv result, leftv arg)
BOOLEAN	makeChannel (leftv result, leftv arg)
BOOLEAN	makeSyncVar (leftv result, leftv arg)
BOOLEAN	makeRegion (leftv result, leftv arg)
BOOLEAN	findSharedObject (leftv result, leftv arg)
BOOLEAN	typeSharedObject (leftv result, leftv arg)
BOOLEAN	bindSharedObject (leftv result, leftv arg)
BOOLEAN	getTable (leftv result, leftv arg)
BOOLEAN	inTable (leftv result, leftv arg)
BOOLEAN	putTable (leftv result, leftv arg)
BOOLEAN	getList (leftv result, leftv arg)
BOOLEAN	putList (leftv result, leftv arg)
BOOLEAN	lockRegion (leftv result, leftv arg)
BOOLEAN	regionLock (leftv result, leftv arg)
BOOLEAN	unlockRegion (leftv result, leftv arg)
BOOLEAN	sendChannel (leftv result, leftv arg)
BOOLEAN	receiveChannel (leftv result, leftv arg)
BOOLEAN	statChannel (leftv result, leftv arg)
BOOLEAN	writeSyncVar (leftv result, leftv arg)
BOOLEAN	updateSyncVar (leftv result, leftv arg)
BOOLEAN	readSyncVar (leftv result, leftv arg)
BOOLEAN	statSyncVar (leftv result, leftv arg)
void	encode_shared (LinTree::LinTree &lintree, leftv val)
leftv	decode_shared (LinTree::LinTree &lintree)
void	ref_shared (LinTree::LinTree &lintree, int by)
void	installShared (int type)
void	makeSharedType (int &type, const char *name)
void	makeRegionlockType (int &type, const char *name)
void	setOption (int ch)
void	thread_init ()
void *	thread_main (void *arg)
void *	interpreter_thread (ThreadState *ts, void *arg)
static ThreadState *	newThread (void *(*thread_func)(ThreadState *, void *), void *arg, const char **error)
ThreadState *	createThread (void *(*thread_func)(ThreadState *, void *), void *arg)
void *	joinThread (ThreadState *ts)
static InterpreterThread *	createInterpreterThread (const char **error)
static BOOLEAN	createThread (leftv result, leftv arg)
static bool	joinInterpreterThread (InterpreterThread *thread)
static BOOLEAN	joinThread (leftv result, leftv arg)
static BOOLEAN	createThreadPool (leftv result, leftv arg)
static BOOLEAN	createThreadPoolSet (leftv result, leftv arg)
static BOOLEAN	getThreadPoolWorkers (leftv result, leftv arg)
static BOOLEAN	setThreadPoolWorkers (leftv result, leftv arg)
static BOOLEAN	getThreadPoolConcurrency (leftv result, leftv arg)
static BOOLEAN	setThreadPoolConcurrency (leftv result, leftv arg)
static BOOLEAN	closeThreadPool (leftv result, leftv arg)
BOOLEAN	currentThreadPool (leftv result, leftv arg)
BOOLEAN	setCurrentThreadPool (leftv result, leftv arg)
static BOOLEAN	createJob (leftv result, leftv arg)
static BOOLEAN	startJob (leftv result, leftv arg)
static BOOLEAN	waitJob (leftv result, leftv arg)
static BOOLEAN	cancelJob (leftv result, leftv arg)
static BOOLEAN	jobCancelled (leftv result, leftv arg)
const char *	getJobName (Job *job)
void	setJobName (Job *job, const char *name)
static BOOLEAN	createTrigger (leftv result, leftv arg)
static BOOLEAN	updateTrigger (leftv result, leftv arg)
static BOOLEAN	chainTrigger (leftv result, leftv arg)
static BOOLEAN	testTrigger (leftv result, leftv arg)
static BOOLEAN	scheduleJob (leftv result, leftv arg)
BOOLEAN	currentJob (leftv result, leftv arg)
BOOLEAN	threadID (leftv result, leftv arg)
BOOLEAN	mainThread (leftv result, leftv arg)
BOOLEAN	threadEval (leftv result, leftv arg)
BOOLEAN	threadExec (leftv result, leftv arg)
BOOLEAN	threadPoolExec (leftv result, leftv arg)
BOOLEAN	threadResult (leftv result, leftv arg)
BOOLEAN	setSharedName (leftv result, leftv arg)
BOOLEAN	getSharedName (leftv result, leftv arg)
ThreadState *	createThread (ThreadFunc body, void **arg, const char **error)
template<typename T >
T *	shared_alloc (std::size_t n)
template<typename T >
T *	shared_alloc0 (std::size_t n)
template<typename T >
T	shared_free (T *p)
template<typename T >
T *	shared_alloc (size_t n)
template<typename T >
T *	shared_alloc0 (size_t n)
template<typename T >
void	shared_free (T *p)

Variables
const int	have_threads = 0
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
Lock	thread_lock
ThreadState *	thread_state
STATIC_VAR ThreadPool *	currentThreadPoolRef
STATIC_VAR Job *	currentJobRef

---

Data Structure Documentation

LibThread::SchedInfo

struct LibThread::SchedInfo

Definition at line 1624 of file shared.cc.

Data Fields
Job *	job
int	num
Scheduler *	scheduler

Typedef Documentation

JobQueue

typedef queue<Job *> LibThread::JobQueue

Definition at line 1620 of file shared.cc.

SharedConstructor

typedef SharedObjectPtr(* LibThread::SharedConstructor) ()

Definition at line 247 of file shared.cc.

SharedObjectPtr

typedef SharedObject* LibThread::SharedObjectPtr

Definition at line 246 of file shared.cc.

SharedObjectTable

typedef std::map<std::string, SharedObject *> LibThread::SharedObjectTable

Definition at line 203 of file shared.cc.

ThreadFunc

typedef void *(* LibThread::ThreadFunc) (ThreadState *, void *)

Definition at line 12 of file singthreads.h.

Function Documentation

acquireShared()

void LibThread::acquireShared

(

SharedObject *

obj

)

Definition at line 193 of file shared.cc.

                                      {
  obj->incref();
}

addJobArgs()

void LibThread::addJobArgs	(	Job *	job,
		leftv	arg
	)

Definition at line 2665 of file shared.cc.

                                     {
  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();
}

allocate_space()

char * LibThread::allocate_space

(

size_t

n

)

appendArg() [1/2]

static void LibThread::appendArg ( vector< leftv > &  argv, leftv  arg  )

static

Definition at line 738 of file shared.cc.

                                                      {
  argv.push_back(arg);
}

appendArg() [2/2]

static void LibThread::appendArg ( vector< leftv > &  argv, string &  s  )

static

Definition at line 728 of file shared.cc.

                                                      {
  if (s.size() == 0) return;
  leftv val = LinTree::from_string(s);
  if (val->Typ() == NONE) {
    omFreeBin(val, sleftv_bin);
    return;
  }
  argv.push_back(val);
}

appendArgCopy()

static void LibThread::appendArgCopy ( vector< leftv > &  argv, leftv  arg  )

static

Definition at line 742 of file shared.cc.

                                                          {
  leftv val = (leftv) omAlloc0Bin(sleftv_bin);
  val->Copy(arg);
  argv.push_back(val);
}

bindSharedObject()

BOOLEAN LibThread::bindSharedObject	(	leftv	result,
		leftv	arg
	)

Definition at line 924 of file shared.cc.

                                                  {
  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;
}

cancelJob() [1/2]

void LibThread::cancelJob

(

Job *

job

)

Definition at line 2514 of file shared.cc.

                         {
  ThreadPool *pool = job->pool;
  if (pool) pool->cancelJob(job);
}

cancelJob() [2/2]

static BOOLEAN LibThread::cancelJob ( leftv  result, leftv  arg  )

static

Definition at line 2595 of file shared.cc.

                                                  {
  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();
}

chainTrigger()

static BOOLEAN LibThread::chainTrigger ( leftv  result, leftv  arg  )

static

Definition at line 2757 of file shared.cc.

                                                     {
  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();
}

closeThreadPool() [1/2]

static BOOLEAN LibThread::closeThreadPool ( leftv  result, leftv  arg  )

static

Definition at line 2313 of file shared.cc.

                                                        {
  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();
}

closeThreadPool() [2/2]

void LibThread::closeThreadPool	(	ThreadPool *	pool,
		bool	wait
	)

Definition at line 2329 of file shared.cc.

                                                  {
  pool->shutdown(wait);
}

consChannel()

SharedObject * LibThread::consChannel

(

)

Definition at line 716 of file shared.cc.

                            {
  return new SingularChannel();
}

consList()

SharedObject * LibThread::consList

(

)

Definition at line 712 of file shared.cc.

                         {
  return new TxList();
}

consRegion()

SharedObject * LibThread::consRegion

(

)

Definition at line 724 of file shared.cc.

                           {
  return new Region();
}

consSyncVar()

SharedObject * LibThread::consSyncVar

(

)

Definition at line 720 of file shared.cc.

                            {
  return new SingularSyncVar();
}

consTable()

SharedObject * LibThread::consTable

(

)

Definition at line 708 of file shared.cc.

                          {
  return new TxTable();
}

createInterpreterThread()

static InterpreterThread * LibThread::createInterpreterThread ( const char **  error )

static

Definition at line 1485 of file shared.cc.

                                                                      {
  ThreadState *ts = newThread(interpreter_thread, NULL, error);
  if (*error) return NULL;
  InterpreterThread *thread = new InterpreterThread(ts);
  char buf[10];
  sprintf(buf, "%d", ts->index);
  string name(buf);
  thread->set_name(name);
  thread->set_type(type_thread);
  return thread;
}

createJob() [1/3]

static BOOLEAN LibThread::createJob ( leftv  result, leftv  arg  )

static

Definition at line 2449 of file shared.cc.

                                                  {
  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();
}

createJob() [2/3]

Job * LibThread::createJob

(

void(*)(leftv result, leftv arg)

func

)

Definition at line 2471 of file shared.cc.

                                                      {
  KernelJob *job = new KernelJob(func);
  return job;
}

createJob() [3/3]

Job * LibThread::createJob

(

void(*)(long ndeps, Job **deps)

func

)

Definition at line 2476 of file shared.cc.

                                                     {
  RawKernelJob *job = new RawKernelJob(func);
  return job;
}

createThread() [1/3]

static BOOLEAN LibThread::createThread ( leftv  result, leftv  arg  )

static

Definition at line 1497 of file shared.cc.

                                                     {
  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();
}

createThread() [2/3]

ThreadState * LibThread::createThread	(	ThreadFunc	body,
		void **	arg,
		const char **	error
	)

createThread() [3/3]

ThreadState * LibThread::createThread	(	void *(*)(ThreadState *, void *)	thread_func,
		void *	arg
	)

Definition at line 1469 of file shared.cc.

               {
  return newThread(thread_func, arg, NULL);
}

createThreadPool() [1/2]

ThreadPool * LibThread::createThreadPool	(	int	threads,
		int	prioThreads = 0
	)

Definition at line 2218 of file shared.cc.

                                                                {
  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;
}

createThreadPool() [2/2]

static BOOLEAN LibThread::createThreadPool ( leftv  result, leftv  arg  )

static

Definition at line 2132 of file shared.cc.

                                                         {
  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();
}

createThreadPoolSet()

static BOOLEAN LibThread::createThreadPoolSet ( leftv  result, leftv  arg  )

static

Definition at line 2166 of file shared.cc.

                                                            {
  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();
}

createTrigger()

static BOOLEAN LibThread::createTrigger ( leftv  result, leftv  arg  )

static

Definition at line 2692 of file shared.cc.

                                                      {
  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();
}

currentJob()

BOOLEAN LibThread::currentJob	(	leftv	result,
		leftv	arg
	)

Definition at line 2898 of file shared.cc.

                                            {
  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();
}

currentThreadPool()

BOOLEAN LibThread::currentThreadPool	(	leftv	result,
		leftv	arg
	)

Definition at line 2334 of file shared.cc.

                                                   {
  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();
}

decode_shared()

leftv LibThread::decode_shared

(

LinTree::LinTree &

lintree

)

Definition at line 1281 of file shared.cc.

                                             {
  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;
}

encode_shared()

void LibThread::encode_shared	(	LinTree::LinTree &	lintree,
		leftv	val
	)

Definition at line 1275 of file shared.cc.

                                                       {
  SharedObject *obj = *(SharedObject **)(val->Data());
  acquireShared(obj);
  lintree.put(obj);
}

executeProc()

static BOOLEAN LibThread::executeProc ( sleftv &  result, const char *  procname, const vector< leftv > &  argv  )

static

Definition at line 749 of file shared.cc.

{
  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;
}

findSharedObject() [1/2]

BOOLEAN LibThread::findSharedObject	(	leftv	result,
		leftv	arg
	)

Definition at line 880 of file shared.cc.

                                                  {
  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;
}

findSharedObject() [2/2]

SharedObject * LibThread::findSharedObject	(	SharedObjectTable &	table,
		Lock *	lock,
		string &	name
	)

Definition at line 271 of file shared.cc.

{
  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;
}

free_space()

void LibThread::free_space	(	size_t	n,
		char *	p
	)

getCurrentJob()

Job * LibThread::getCurrentJob

(

)

Definition at line 2519 of file shared.cc.

                     {
  return currentJobRef;
}

getCurrentThreadPool()

ThreadPool * LibThread::getCurrentThreadPool

(

)

Definition at line 2245 of file shared.cc.

                                   {
  return currentThreadPoolRef;
}

getJobCancelled() [1/2]

bool LibThread::getJobCancelled

(

)

Definition at line 2645 of file shared.cc.

                       {
  return getJobCancelled(currentJobRef);
}

getJobCancelled() [2/2]

bool LibThread::getJobCancelled

(

Job *

job

)

Definition at line 2637 of file shared.cc.

                               {
  ThreadPool *pool = job->pool;
  if (pool) pool->scheduler->lock.lock();
  bool result = job->cancelled;
  if (pool) pool->scheduler->lock.unlock();
  return result;
}

getJobData()

void * LibThread::getJobData

(

Job *

job

)

Definition at line 2657 of file shared.cc.

                           {
  ThreadPool *pool = job->pool;
  if (pool) pool->scheduler->lock.lock();
  void *result = job->data;
  if (pool) pool->scheduler->lock.unlock();
  return result;
}

getJobName() [1/2]

const char * LibThread::getJobName

(

)

getJobName() [2/2]

const char * LibThread::getJobName

(

Job *

job

)

Definition at line 2683 of file shared.cc.

                                 {
  // TODO
  return "";
}

getJobResult()

leftv LibThread::getJobResult

(

Job *

job

)

Definition at line 2675 of file shared.cc.

                             {
  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;
}

getList()

BOOLEAN LibThread::getList	(	leftv	result,
		leftv	arg
	)

Definition at line 1028 of file shared.cc.

                                         {
  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;
}

getSharedName()

BOOLEAN LibThread::getSharedName	(	leftv	result,
		leftv	arg
	)

Definition at line 3056 of file shared.cc.

                                               {
  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();
}

getTable()

BOOLEAN LibThread::getTable	(	leftv	result,
		leftv	arg
	)

Definition at line 941 of file shared.cc.

                                          {
  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;
}

getThreadPoolConcurrency()

static BOOLEAN LibThread::getThreadPoolConcurrency ( leftv  result, leftv  arg  )

static

Definition at line 2281 of file shared.cc.

                                                                 {
  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();
}

getThreadPoolWorkers()

static BOOLEAN LibThread::getThreadPoolWorkers ( leftv  result, leftv  arg  )

static

Definition at line 2249 of file shared.cc.

                                                             {
  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();
}

installShared()

void LibThread::installShared

(

int

type

)

Definition at line 1302 of file shared.cc.

                             {
  LinTree::install(type, encode_shared, decode_shared, ref_shared);
}

inTable()

BOOLEAN LibThread::inTable	(	leftv	result,
		leftv	arg
	)

Definition at line 974 of file shared.cc.

                                         {
  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;
}

interpreter_thread()

void * LibThread::interpreter_thread	(	ThreadState *	ts,
		void *	arg
	)

Definition at line 1391 of file shared.cc.

                                                     {
  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;
}

jobCancelled()

static BOOLEAN LibThread::jobCancelled ( leftv  result, leftv  arg  )

static

Definition at line 2612 of file shared.cc.

                                                     {
  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();
}

joinInterpreterThread()

static bool LibThread::joinInterpreterThread ( InterpreterThread *  thread )

static

Definition at line 1512 of file shared.cc.

                                                             {
  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;
}

joinThread() [1/2]

static BOOLEAN LibThread::joinThread ( leftv  result, leftv  arg  )

static

Definition at line 1531 of file shared.cc.

                                                   {
  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;
}

joinThread() [2/2]

void * LibThread::joinThread

(

ThreadState *

ts

)

Definition at line 1474 of file shared.cc.

                                  {
  void *result;
  pthread_join(ts->id, NULL);
  result = ts->result;
  thread_lock.lock();
  ts->running = false;
  ts->active = false;
  thread_lock.unlock();
  return result;
}

lockRegion()

BOOLEAN LibThread::lockRegion	(	leftv	result,
		leftv	arg
	)

Definition at line 1088 of file shared.cc.

                                            {
  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;
}

mainThread()

BOOLEAN LibThread::mainThread	(	leftv	result,
		leftv	arg
	)

Definition at line 2919 of file shared.cc.

                                            {
  if (wrong_num_args("mainThread", arg, 0))
    return TRUE;
  result->rtyp = INT_CMD;
  result->data = (char *)(long)(thread_id == 0L);
  return FALSE;
}

makeAtomicList()

BOOLEAN LibThread::makeAtomicList	(	leftv	result,
		leftv	arg
	)

Definition at line 792 of file shared.cc.

                                                {
  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;
}

makeAtomicTable()

BOOLEAN LibThread::makeAtomicTable	(	leftv	result,
		leftv	arg
	)

Definition at line 778 of file shared.cc.

                                                 {
  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;
}

makeChannel()

BOOLEAN LibThread::makeChannel	(	leftv	result,
		leftv	arg
	)

Definition at line 841 of file shared.cc.

                                             {
  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;
}

makeRegion()

BOOLEAN LibThread::makeRegion	(	leftv	result,
		leftv	arg
	)

Definition at line 867 of file shared.cc.

                                            {
  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;
}

makeRegionlockType()

void LibThread::makeRegionlockType	(	int &	type,
		const char *	name
	)

Definition at line 1321 of file shared.cc.

                                                     {
  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);
}

makeSharedList()

BOOLEAN LibThread::makeSharedList	(	leftv	result,
		leftv	arg
	)

Definition at line 824 of file shared.cc.

                                                {
  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;
}

makeSharedObject()

SharedObject * LibThread::makeSharedObject	(	SharedObjectTable &	table,
		Lock *	lock,
		int	type,
		string &	name,
		SharedConstructor	scons
	)

Definition at line 249 of file shared.cc.

{
  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;
}

makeSharedTable()

BOOLEAN LibThread::makeSharedTable	(	leftv	result,
		leftv	arg
	)

Definition at line 806 of file shared.cc.

                                                 {
  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;
}

makeSharedType()

void LibThread::makeSharedType	(	int &	type,
		const char *	name
	)

Definition at line 1306 of file shared.cc.

                                                 {
  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);
}

makeSyncVar()

BOOLEAN LibThread::makeSyncVar	(	leftv	result,
		leftv	arg
	)

Definition at line 854 of file shared.cc.

                                             {
  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;
}

new_shared()

void * LibThread::new_shared

(

SharedObject *

obj

)

Definition at line 491 of file shared.cc.

                                    {
  acquireShared(obj);
  void *result = omAlloc0(sizeof(SharedObject *));
  *(SharedObject **)result = obj;
  return result;
}

newThread()

static ThreadState * LibThread::newThread ( void *(*)(ThreadState *, void *)  thread_func, void *  arg, const char **  error  )

static

Definition at line 1436 of file shared.cc.

                                   {
  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;
}

not_a_region()

int LibThread::not_a_region	(	const char *	name,
		leftv	arg
	)

Definition at line 695 of file shared.cc.

                                              {
  if (arg->Typ() != type_region || !arg->Data()) {
    report("%s: not a region", name);
    return TRUE;
  }
  return FALSE;
}

not_a_uri()

int LibThread::not_a_uri	(	const char *	name,
		leftv	arg
	)

Definition at line 687 of file shared.cc.

                                           {
  if (arg->Typ() != STRING_CMD) {
    report("%s: not a valid URI", name);
    return TRUE;
  }
  return FALSE;
}

putList()

BOOLEAN LibThread::putList	(	leftv	result,
		leftv	arg
	)

Definition at line 1061 of file shared.cc.

                                         {
  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;
}

putTable()

BOOLEAN LibThread::putTable	(	leftv	result,
		leftv	arg
	)

Definition at line 1001 of file shared.cc.

                                          {
  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;
}

readSyncVar()

BOOLEAN LibThread::readSyncVar	(	leftv	result,
		leftv	arg
	)

Definition at line 1238 of file shared.cc.

                                             {
  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;
}

receiveChannel()

BOOLEAN LibThread::receiveChannel	(	leftv	result,
		leftv	arg
	)

Definition at line 1152 of file shared.cc.

                                                {
  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;
}

ref_shared()

void LibThread::ref_shared	(	LinTree::LinTree &	lintree,
		int	by
	)

Definition at line 1290 of file shared.cc.

                                                 {
  SharedObject *obj = lintree.get<SharedObject *>();
  while (by > 0) {
    obj->incref();
    by--;
  }
  while (by < 0) {
    obj->decref();
    by++;
  }
}

regionLock()

BOOLEAN LibThread::regionLock	(	leftv	result,
		leftv	arg
	)

Definition at line 1103 of file shared.cc.

                                            {
  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;
}

release() [1/2]

void LibThread::release

(

Job *

job

)

release() [2/2]

void LibThread::release

(

ThreadPool *

pool

)

Definition at line 2237 of file shared.cc.

                               {
  releaseShared(pool);
}

releaseShared()

void LibThread::releaseShared

(

SharedObject *

obj

)

Definition at line 197 of file shared.cc.

                                      {
  if (obj->decref() == 0) {
    // delete obj;
  }
}

report()

void LibThread::report	(	const char *	fmt,
		const char *	name
	)

Definition at line 666 of file shared.cc.

                                               {
  char buf[80];
  sprintf(buf, fmt, name);
  WerrorS(buf);
}

retain() [1/2]

void LibThread::retain

(

Job *

job

)

retain() [2/2]

void LibThread::retain

(

ThreadPool *

pool

)

Definition at line 2241 of file shared.cc.

                              {
  acquireShared(pool);
}

rlock_assign()

BOOLEAN LibThread::rlock_assign	(	leftv	l,
		leftv	r
	)

Definition at line 549 of file shared.cc.

                                       {
  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;
}

rlock_destroy()

void LibThread::rlock_destroy	(	blackbox *	b,
		void *	d
	)

Definition at line 506 of file shared.cc.

                                         {
  SharedObject *obj = *(SharedObject **)d;
  ((Region *) obj)->unlock();
  if (obj) {
    releaseShared(*(SharedObject **)d);
    *(SharedObject **)d = NULL;
  }
}

rlock_string()

char * LibThread::rlock_string	(	blackbox *	b,
		void *	d
	)

Definition at line 657 of file shared.cc.

                                         {
  char buf[80];
  SharedObject *obj = *(SharedObject **)d;
  if (!obj)
    return omStrDup("<uninitialized region lock>");
  sprintf(buf, "<region lock \"%.40s\">", obj->get_name().c_str());
  return omStrDup(buf);
}

scheduleJob()

static BOOLEAN LibThread::scheduleJob ( leftv  result, leftv  arg  )

static

Definition at line 2794 of file shared.cc.

                                                    {
  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();
}

sendChannel()

BOOLEAN LibThread::sendChannel	(	leftv	result,
		leftv	arg
	)

Definition at line 1135 of file shared.cc.

                                             {
  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;
}

setCurrentThreadPool()

BOOLEAN LibThread::setCurrentThreadPool	(	leftv	result,
		leftv	arg
	)

Definition at line 2346 of file shared.cc.

                                                      {
  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();
}

setJobData()

void LibThread::setJobData	(	Job *	job,
		void *	data
	)

Definition at line 2649 of file shared.cc.

                                      {
  ThreadPool *pool = job->pool;
  if (pool) pool->scheduler->lock.lock();
  job->data = data;
  if (pool) pool->scheduler->lock.unlock();
}

setJobName() [1/2]

void LibThread::setJobName

(

const char *

)

setJobName() [2/2]

void LibThread::setJobName	(	Job *	job,
		const char *	name
	)

Definition at line 2688 of file shared.cc.

                                            {
  // TODO
}

setOption()

void LibThread::setOption

(

int

ch

)

Definition at line 1368 of file shared.cc.

                       {
  int index = feGetOptIndex(ch);
  feSetOptValue((feOptIndex) index, (int) 1);
}

setSharedName()

BOOLEAN LibThread::setSharedName	(	leftv	result,
		leftv	arg
	)

Definition at line 3038 of file shared.cc.

                                               {
  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();
}

setThreadPoolConcurrency()

static BOOLEAN LibThread::setThreadPoolConcurrency ( leftv  result, leftv  arg  )

static

Definition at line 2296 of file shared.cc.

                                                                 {
  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();
}

setThreadPoolWorkers()

static BOOLEAN LibThread::setThreadPoolWorkers ( leftv  result, leftv  arg  )

static

Definition at line 2266 of file shared.cc.

                                                             {
  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();
}

shared_alloc() [1/2]

template<typename T >

T * LibThread::shared_alloc

(

size_t

n

)

Definition at line 42 of file thread.cc.

                            {
    T *p = (T *) malloc(n * sizeof(T));
    return p;
  }

shared_alloc() [2/2]

template<typename T >

T * LibThread::shared_alloc

(

std::size_t

n

)

shared_alloc0() [1/2]

template<typename T >

T * LibThread::shared_alloc0

(

size_t

n

)

Definition at line 47 of file thread.cc.

                             {
    T *p = (T *) calloc(n, sizeof(T));
    return p;
  }

shared_alloc0() [2/2]

template<typename T >

T * LibThread::shared_alloc0

(

std::size_t

n

)

shared_assign()

BOOLEAN LibThread::shared_assign	(	leftv	l,
		leftv	r
	)

Definition at line 524 of file shared.cc.

                                        {
  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;
}

shared_check_assign()

BOOLEAN LibThread::shared_check_assign	(	blackbox *	b,
		leftv	l,
		leftv	r
	)

Definition at line 573 of file shared.cc.

                                                           {
  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;
}

shared_copy()

void * LibThread::shared_copy	(	blackbox *	b,
		void *	d
	)

Definition at line 515 of file shared.cc.

                                        {
  SharedObject *obj = *(SharedObject **)d;
  void *result = shared_init(b);
  *(SharedObject **)result = obj;
  if (obj)
    acquireShared(obj);
  return result;
}

shared_destroy()

void LibThread::shared_destroy	(	blackbox *	b,
		void *	d
	)

Definition at line 498 of file shared.cc.

                                          {
  SharedObject *obj = *(SharedObject **)d;
  if (obj) {
    releaseShared(*(SharedObject **)d);
    *(SharedObject **)d = NULL;
  }
}

shared_free() [1/2]

template<typename T >

T LibThread::shared_free

(

T *

p

)

Definition at line 52 of file thread.cc.

                         {
    free(p);
  }

shared_free() [2/2]

template<typename T >

void LibThread::shared_free

(

T *

p

)

Definition at line 52 of file thread.cc.

                         {
    free(p);
  }

shared_init()

void * LibThread::shared_init

(

blackbox *

b

)

Definition at line 487 of file shared.cc.

                               {
  return omAlloc0(sizeof(SharedObject *));
}

shared_op2()

BOOLEAN LibThread::shared_op2	(	int	op,
		leftv	res,
		leftv	a1,
		leftv	a2
	)

Definition at line 585 of file shared.cc.

                                                          {
  SharedObject *obj = *(SharedObject **)a1->Data();
  return obj->op2(op, res, a1, a2);
}

shared_op3()

BOOLEAN LibThread::shared_op3	(	int	op,
		leftv	res,
		leftv	a1,
		leftv	a2,
		leftv	a3
	)

Definition at line 590 of file shared.cc.

                                                                    {
  SharedObject *obj = *(SharedObject **)a1->Data();
  return obj->op3(op, res, a1, a2, a3);
}

shared_string()

char * LibThread::shared_string	(	blackbox *	b,
		void *	d
	)

Definition at line 595 of file shared.cc.

                                          {
  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) {
    sprintf(buf, "<thread #%s>", name.c_str());
    return omStrDup(buf);
  }
  else if (type == type_threadpool) {
    if (name.size() > 0) {
      name_lock.lock();
      sprintf(buf, "<threadpool \"%.40s\"@%p>", name.c_str(), obj);
      name_lock.unlock();
    } else
      sprintf(buf, "<threadpool @%p>", obj);
    return omStrDup(buf);
  }
  else if (type == type_job) {
    if (name.size() > 0) {
      name_lock.lock();
      sprintf(buf, "<job \"%.40s\"@%p>", name.c_str(), obj);
      name_lock.unlock();
    } else
      sprintf(buf, "<job @%p>", obj);
    return omStrDup(buf);
  }
  else if (type == type_trigger) {
    if (name.size() > 0) {
      name_lock.lock();
      sprintf(buf, "<trigger \"%.40s\"@%p>", name.c_str(), obj);
      name_lock.unlock();
    } else
      sprintf(buf, "<trigger @%p>", obj);
    return omStrDup(buf);
  } else {
    sprintf(buf, "<unknown type %d>", type);
    return omStrDup(buf);
  }
  sprintf(buf, "<%s \"%.40s\">", type_name, name.c_str());
  return omStrDup(buf);
}

startJob() [1/3]

static BOOLEAN LibThread::startJob ( leftv  result, leftv  arg  )

static

Definition at line 2523 of file shared.cc.

                                                 {
  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();
}

startJob() [2/3]

Job * LibThread::startJob	(	ThreadPool *	pool,
		Job *	job
	)

Definition at line 2491 of file shared.cc.

                                          {
  return startJob(pool, job, NULL);
}

startJob() [3/3]

Job * LibThread::startJob	(	ThreadPool *	pool,
		Job *	job,
		leftv	arg
	)

Definition at line 2481 of file shared.cc.

                                                     {
  if (job->pool) return NULL;
  while (arg) {
    job->args.push_back(LinTree::to_string(arg));
    arg = arg->next;
  }
  pool->attachJob(job);
  return job;
}

statChannel()

BOOLEAN LibThread::statChannel	(	leftv	result,
		leftv	arg
	)

Definition at line 1171 of file shared.cc.

                                             {
  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;
}

statSyncVar()

BOOLEAN LibThread::statSyncVar	(	leftv	result,
		leftv	arg
	)

Definition at line 1257 of file shared.cc.

                                             {
  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;
}

str()

char * LibThread::str

(

leftv

arg

)

Definition at line 704 of file shared.cc.

                     {
  return (char *)(arg->Data());
}

testTrigger()

static BOOLEAN LibThread::testTrigger ( leftv  result, leftv  arg  )

static

Definition at line 2778 of file shared.cc.

                                                    {
  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();
}

thread_init()

void LibThread::thread_init

(

)

Definition at line 1373 of file shared.cc.

                   {
  master_lock.lock();
  thread_id = ++thread_counter;
  master_lock.unlock();
#ifdef ENABLE_THREADS
  pSingular_initialize_thread();
  siInit(global_argv0);
#endif
  setOption('q');
  // setOption('b');
}

thread_main()

void * LibThread::thread_main

(

void *

arg

)

Definition at line 1385 of file shared.cc.

                             {
  ThreadState *ts = (ThreadState *)arg;
  thread_init();
  return ts->thread_func(ts, ts->arg);
}

threadEval()

BOOLEAN LibThread::threadEval	(	leftv	result,
		leftv	arg
	)

Definition at line 2927 of file shared.cc.

                                            {
  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;
}

threadExec()

BOOLEAN LibThread::threadExec	(	leftv	result,
		leftv	arg
	)

Definition at line 2955 of file shared.cc.

                                            {
  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;
}

threadID()

BOOLEAN LibThread::threadID	(	leftv	result,
		leftv	arg
	)

Definition at line 2911 of file shared.cc.

                                          {
  if (wrong_num_args("threadID", arg, 0))
    return TRUE;
  result->rtyp = INT_CMD;
  result->data = (char *)thread_id;
  return FALSE;
}

threadPoolExec()

BOOLEAN LibThread::threadPoolExec	(	leftv	result,
		leftv	arg
	)

Definition at line 2983 of file shared.cc.

                                                {
  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();
}

threadResult()

BOOLEAN LibThread::threadResult	(	leftv	result,
		leftv	arg
	)

Definition at line 3007 of file shared.cc.

                                              {
  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;
}

typeSharedObject()

BOOLEAN LibThread::typeSharedObject	(	leftv	result,
		leftv	arg
	)

Definition at line 893 of file shared.cc.

                                                  {
  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;
}

unlockRegion()

BOOLEAN LibThread::unlockRegion	(	leftv	result,
		leftv	arg
	)

Definition at line 1120 of file shared.cc.

                                              {
  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;
}

updateSyncVar()

BOOLEAN LibThread::updateSyncVar	(	leftv	result,
		leftv	arg
	)

Definition at line 1209 of file shared.cc.

                                               {
  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();
}

updateTrigger()

static BOOLEAN LibThread::updateTrigger ( leftv  result, leftv  arg  )

static

Definition at line 2735 of file shared.cc.

                                                      {
  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();
}

waitJob() [1/2]

void LibThread::waitJob

(

Job *

job

)

Definition at line 2590 of file shared.cc.

                       {
  assert(job->pool != NULL);
  job->pool->waitJob(job);
}

waitJob() [2/2]

static BOOLEAN LibThread::waitJob ( leftv  result, leftv  arg  )

static

Definition at line 2565 of file shared.cc.

                                                {
  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();
}

writeSyncVar()

BOOLEAN LibThread::writeSyncVar	(	leftv	result,
		leftv	arg
	)

Definition at line 1189 of file shared.cc.

                                              {
  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;
}

wrong_num_args()

int LibThread::wrong_num_args	(	const char *	name,
		leftv	arg,
		int	n
	)

Definition at line 672 of file shared.cc.

                                                       {
  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;
}

Variable Documentation

currentJobRef

STATIC_VAR Job* LibThread::currentJobRef

Definition at line 1631 of file shared.cc.

currentThreadPoolRef

STATIC_VAR ThreadPool* LibThread::currentThreadPoolRef

Definition at line 1630 of file shared.cc.

global_objects

SharedObjectTable LibThread::global_objects

Definition at line 227 of file shared.cc.

global_objects_lock

Lock LibThread::global_objects_lock

Definition at line 226 of file shared.cc.

have_threads

const int LibThread::have_threads = 0

Definition at line 38 of file shared.cc.

master_lock

Lock LibThread::master_lock(true)

(

true

)

name_lock

Lock LibThread::name_lock(true)

(

true

)

thread_counter

long LibThread::thread_counter

Definition at line 231 of file shared.cc.

thread_id

VAR long LibThread::thread_id

Definition at line 230 of file shared.cc.

thread_lock

Lock LibThread::thread_lock

Definition at line 1364 of file shared.cc.

thread_state

ThreadState* LibThread::thread_state

Definition at line 1366 of file shared.cc.

type_atomic_list

int LibThread::type_atomic_list

Definition at line 239 of file shared.cc.

type_atomic_table

int LibThread::type_atomic_table

Definition at line 237 of file shared.cc.

type_channel

int LibThread::type_channel

Definition at line 235 of file shared.cc.

type_job

int LibThread::type_job

Definition at line 243 of file shared.cc.

type_region

int LibThread::type_region

Definition at line 233 of file shared.cc.

type_regionlock

int LibThread::type_regionlock

Definition at line 234 of file shared.cc.

type_shared_list

int LibThread::type_shared_list

Definition at line 240 of file shared.cc.

type_shared_table

int LibThread::type_shared_table

Definition at line 238 of file shared.cc.

type_syncvar

int LibThread::type_syncvar

Definition at line 236 of file shared.cc.

type_thread

int LibThread::type_thread

Definition at line 241 of file shared.cc.

type_threadpool

int LibThread::type_threadpool

Definition at line 242 of file shared.cc.

type_trigger

int LibThread::type_trigger

Definition at line 244 of file shared.cc.