updates of XThead

source/tensor/XPRunner.cpp

@@ -146,7 +146,7 @@ run a set of jobs in parallel
 >> jobArgs - the list of arguments for each job
 >> sleepTime - time to sleep (in ms) for each round
 */
-void XPRunner::Run(TensorList * jobFunctions, TensorList * jobArgs, float sleepTime)
+void XPRunner::Run(XList * jobFunctions, XList * jobArgs, float sleepTime)
 {
     if(threadNum <= 0){
         XPRINT(1, stderr, "Error! No threads were created!\n");
@@ -195,13 +195,12 @@ void XPRunner::Run(TensorList * jobFunctions, TensorList * jobArgs, float sleepT
             TFunction function = (TFunction)jobFunctions->GetItem(jobArgs->count - c);
 
             /* the arguments that are passed to the function */
-            volatile TensorList * args = (TensorList*)jobArgs->GetItem(jobArgs->count - c);
+            XList * args = (XList*)jobArgs->GetItem(jobArgs->count - c);
 
             /* thread */
             XThread * thread  = threads + availableThreads[i];
 
-            thread->argv = args;
-            thread->function = function;
+            thread->SetFunc(function, args);
 
             MUTEX_LOCK(thread->workingMutex);
             thread->working = 1;

source/tensor/XPRunner.h

@@ -106,7 +106,7 @@ public:
     void KillThreads();
 
     /* run a set of jobs in parallel */
-    void Run(TensorList * jobFunctions, TensorList * jobArgs, float sleepTime = 0);
+    void Run(XList * jobFunctions, XList * jobArgs, float sleepTime = 0);
 
     /* get the number of parallel jobs to run */
     int GetJobNum(int size);

source/tensor/XQueue.cpp

@@ -42,7 +42,7 @@ job item used in queues
 JobQueueNode::JobQueueNode()
 {
     job  = NULL;
-    args = new TensorList(1);
+    args = new XList(1);
 }
 
 /* de-constructor */
@@ -67,7 +67,7 @@ XQueue::XQueue(int mySize)
     head = 0;
     tail = 0;
     isJobQueue = false;
-    jobDequeuerArgs = new TensorList(1);
+    jobDequeuerArgs = new XList(1);
     jobDequeuerBreak = false;
     runningJobCount = 0;
     
@@ -171,11 +171,10 @@ void XQueue::RunJobConsumer(int jobDevID)
     jobDequeuerArgs->Clear();
 
     // warning: this may cause unknown error
-    jobDequeuerArgs->Add((XTensor*)this);
-    jobDequeuerArgs->Add(jobDevID >= 0 ? (XTensor*)(devids + jobDevID) : (XTensor*)&cpuid);
+    jobDequeuerArgs->Add(this);
+    jobDequeuerArgs->Add(jobDevID >= 0 ? (devids + jobDevID) : &cpuid);
 
-    jobDequeuer.function = (TFunction)DequeueJobs;
-    jobDequeuer.argv = jobDequeuerArgs;
+    jobDequeuer.SetFunc((TFunction)DequeueJobs, jobDequeuerArgs);
 
     jobDequeuer.Start();
     jobDequeuer.LetItGo();
@@ -194,7 +193,7 @@ void XQueue::StopJobConsumer()
 }
 
 /* add a job item to process */
-void XQueue::EnqueueJob(void * job, TensorList * jobArgs)
+void XQueue::EnqueueJob(void * job, XList * jobArgs)
 {
     MUTEX_LOCK(jobQueueMutex);
     runningJobCount++;
@@ -208,7 +207,7 @@ void XQueue::EnqueueJob(void * job, TensorList * jobArgs)
 }
 
 /* job item consumer */
-void XQueue::DequeueJobs(TensorList * args)
+void XQueue::DequeueJobs(XList * args)
 {
     CheckNTErrors((args->count == 2), "Illegal arguments!");
 

source/tensor/XQueue.h

@@ -51,7 +51,7 @@ public:
     void * job;
 
     /* arguments of the job */
-    TensorList * args;
+    XList * args;
 
 public:
     /* constructor */
@@ -101,7 +101,7 @@ private:
     XThread jobDequeuer;
 
     /* argument list of jobDequeuer */
-    TensorList * jobDequeuerArgs;
+    XList * jobDequeuerArgs;
 
     /* indicates whether jobDequeuer stops */
     bool jobDequeuerBreak;
@@ -135,11 +135,11 @@ public:
     void StopJobConsumer();
 
     /* add a job item to process */
-    void EnqueueJob(void * job, TensorList * jobArgs);
+    void EnqueueJob(void * job, XList * jobArgs);
 
     /* job item consumer */
     static
-    void DequeueJobs(TensorList * args);
+    void DequeueJobs(XList * args);
 
     /* get the break flag */
     bool GetJobBreak();

source/tensor/XThread.cpp

@@ -38,7 +38,7 @@ XThread::XThread()
 #endif
     MUTEX_INIT(gMutex);
     function = NULL;
-    argv = NULL;
+    argv.Clear();
     toBreak = false;
     jobCount = 0;
     working = 0;
@@ -69,6 +69,18 @@ void * XThread::Wrapper(void * ptr)
     return 0;
 }
 
+/* 
+initialize the thread with the function and its parameters 
+>> myFunc - the function to run
+>> myArgv - arguments of the function
+*/
+void XThread::SetFunc(TFunction myFunc, XList * myArgv)
+{
+    function = myFunc;
+    argv.Clear();
+    argv.AddList(myArgv);
+}
+
 
 /* 
 Tunning for this thread. It is very very native implementation.
@@ -77,6 +89,10 @@ After that, we wait again if there is no new job.
 */
 void XThread::Run()
 {
+    if (function == NULL) {
+        ShowNTErrors("You are running a thread with no function specified!");
+    }
+
 #ifdef _WIN32
     //COND_RESET(gCond);
 #endif    
@@ -104,7 +120,7 @@ void XThread::Run()
         }
 
         /* do what you want to do*/
-        function(argv);
+        function(&argv);
 
 #ifdef USE_PTHREAD
         jobCount--;

source/tensor/XThread.h

@@ -85,7 +85,7 @@ namespace nts{
 
 #endif
 
-typedef void (*TFunction) (volatile TensorList*);
+typedef void (*TFunction) (volatile XList*);
 
 /*
 This is a class that wraps the standard implementation of threading
@@ -128,12 +128,10 @@ public:
 
 public:
     /* function to run */
-    volatile
     TFunction function;
 
     /* arguments (for the function to run) */
-    volatile
-    TensorList * argv;
+    XList argv;
 
     /* a flag to break */
     volatile
@@ -154,6 +152,9 @@ public:
     /* a wrapper for the start-routine parameter in pthread_create */
     static void * Wrapper(void * ptr);
 
+    /* initialize the thread with the function and its parameters */
+    void SetFunc(TFunction myFunc, XList * myArgv);
+
     /* 
     Core of the thread. It is very very native impelementation.
     We loop and wait for a singnal to activate the job processing.

source/tensor/core/utilities/XMatrixSegment.cpp

@@ -51,7 +51,7 @@ void RunParallel2D(XPRunner * parallelRunner, void * job,
     CheckNTErrors(jobNum != 0, "TODO!");
 
     /* argument list of the jobs */
-    TensorList * jobArgList = new TensorList(argNum);
+    XList * jobArgList = new XList(argNum);
 
     va_list ap;
     va_start(ap, argNum);
@@ -62,8 +62,8 @@ void RunParallel2D(XPRunner * parallelRunner, void * job,
     va_end(ap);
 
     /* prepare the neccesary argument list for parallel processing */
-    TensorList * jobs = new TensorList(jobNum);
-    TensorList * args = new TensorList(jobNum);
+    XList * jobs = new XList(jobNum);
+    XList * args = new XList(jobNum);
 
     int * indexList = new int[jobNum * 4 * 4];
 
@@ -78,7 +78,7 @@ void RunParallel2D(XPRunner * parallelRunner, void * job,
     */
     for (int i = 0; i < jobNum; i++) {
         IntList* indexArgs = new IntList(4);
-        TensorList * blockArgs = new TensorList(argNum);
+        XList * blockArgs = new XList(argNum);
         int * blockIndex = indexList + i * 4;
 
         indexArgs->Add(blockIndex[0]);
@@ -89,10 +89,10 @@ void RunParallel2D(XPRunner * parallelRunner, void * job,
         for (int j = 0; j < argNum; j++)
             blockArgs->Add(jobArgList->GetItem(j));
 
-        args->Add((XTensor*)indexArgs);
-        args->Add((XTensor*)blockArgs);
+        args->Add((void*)indexArgs);
+        args->Add((void*)blockArgs);
 
-        jobs->Add((XTensor*)job);
+        jobs->Add((void*)job);
     }
 
     args->count = jobNum * 2;