Changeset 431 for branches/thune

Timestamp:

07/16/07 03:30:29 (17 months ago)

Author:

krobillard

Message:

Atoms should now be thread safe. First thread experiment runs.

Location:

branches/thune/thread_safe

Files:

• config.t (modified) (1 diff)
• context.c (modified) (11 diffs)
• eval.c (modified) (1 diff)
• internal.h (modified) (2 diffs)
• project.r (modified) (1 diff)
• read_config.r (modified) (1 diff)
• series.c (modified) (1 diff)
• thread.c (modified) (10 diffs)
• thune.c (modified) (7 diffs)
• tokenize.c (modified) (1 diff)
• unix/os.h (modified) (1 diff)
• urlan.c (modified) (5 diffs)
• urlan.h (modified) (1 diff)
• win32/os.h (modified) (1 diff)

branches/thune/thread_safe/config.t

@@ -9 +9 @@
 [x] dataflow "Component! datatype for dataflow programming"
 
+[ ] threads  "CPU Threads"
+[ ] emh      "Debugger Hooks"
 [ ] dt-code  "Include 'code datatype"
 [ ] uds      "Library with datatype & gc system only - no eval"
              disable [bzip2 trig math3d dt-code]
 
-[ ] threads  "Experimental script threads"
-[ ] emh      "Debugger Hooks"
-
 ;eof

branches/thune/thread_safe/context.c

@@ -25 +25 @@
 
 #define LOWERCASE(c)    if(c >= 'A' && c <= 'Z') c -= 'A' - 'a'
+
+
+/*
+   UrlanEnv::atoms & UrlanEnv::atomNames can be made thread safe through one
+   of the following:
+
+   1. Use LOCK_ATOMS in or around these functions:
+        ur_internT()
+        ur_atomStrT()
+        ur_atomCStrT() (and all use of returned pointer)
+        ur_atomHash()
+        dumpAtoms()
+
+   2. Fix size of atom arrays and throw error/assert when full.
+      Must still lock these functions to access head/chain AtomRec members:
+        ur_internT()
+        dumpAtoms()
+
+   Option #2 is currently being used.
+*/
 
 
@@ -120 +140 @@
 void dumpAtoms( UThread* ut )
 {
-    LOCK_GLOBAL
+    LOCK_ATOMS
     {
     const char* names = ut->env->atomNames.ptr.c;
@@ -143 +163 @@
     }
     }
-    UNLOCK_GLOBAL
+    UNLOCK_ATOMS
 }
 #endif
@@ -151 +171 @@
   Add atom to environment.
 
-  If the environment has multiple threads, the caller must have called
-  LOCK_GLOBAL.
-
   \param str  Name of atom.
   \param len  Number of characters.
 
-  \returns Atom 
+  \return  Atom 
 */
 UAtom ur_internT( UThread* ut, const char* str, int len )
@@ -171 +188 @@
     AtomRec* node;
 
-
     assert( len > 0 );
 
 
     // Check if atom already exists.
+
+    hash = ur_hash( str, str + len );
+
+    LOCK_ATOMS
 
     atoms = &ut->env->atoms;
     table = (AtomRec*) atoms->ptr.v;
     names = &ut->env->atomNames;
-
-    hash = ur_hash( str, str + len );
 
     node = table + (hash % atoms->avail);
@@ -228 +246 @@
     // Nope, add new atom.
 
-    /* TODO: Make atoms & atomNames thread safe through one of the following:
-
-       1. Halt all other threads.
-       2. Fix size of atom arrays and throw error/assert when full.
-       3. Use LOCK_GLOBAL in or around these functions in addition to ur_intern:
-            ur_atomStrT()
-            ur_atomCStrT()
-            dumpAtoms()
-    */
-
     if( atoms->used == atoms->avail )
     {
+#if 1
+        // Atom table size is fixed so read only access does not need to be
+        // locked.  When the table is full, we are finished.
+        assert( 0 && "Atom table is full" );
+        return 0;       // TODO: Report fatal error
+#else
         ur_arrayReserve( atoms, sizeof(AtomRec), atoms->used + 1 );
         ur_rebuildAtomHash( atoms );
@@ -245 +259 @@
 
         HASH_INSERT( atoms, table, node, hash, atoms->used )
+#endif
     }
     node = table + atoms->used;
@@ -253 +268 @@
     node->nameLen   = len;
 
+#if 1
+    if( (names->used + len + 1) > names->avail )
+    {
+        assert( 0 && "Atom name buffer is full" );
+        return 0;       // TODO: Report fatal error
+    }
+#else
     ur_arrayReserve( names, sizeof(char), names->used + len + 1 );
+#endif
+
     cp = names->ptr.c + names->used;
     ep = cp + len;
@@ -267 +291 @@
 done:
 
-    return node - table;
+    c = node - table;
+
+    UNLOCK_ATOMS
+
+    return c;
 }
 
@@ -278 +306 @@
   If added, the word is initialied as unset.
 
-  \returns  Index of word in context.
+  \return  Index of word in context.
 */
 int ur_internWordT( UThread* ut, const UContext* ctx, UAtom atom )
@@ -424 +452 @@
 /**
   Find word in context by atom.
-  \returns  Word index or -1 if not found.
+  \return  Word index or -1 if not found.
 */
 int ur_lookupT( UThread* ut, const UContext* ctx, UAtom atom )

branches/thune/thread_safe/eval.c

@@ -1765 +1765 @@
     { uc_thread,       "thread" },
 #endif
+#ifdef UR_CONFIG_TASKS
+    { uc_task,         "task" },
+#endif
     { uc_showTOS,      "." },
     { uc_lift_local,   "lift-local" },

branches/thune/thread_safe/internal.h

@@ -31 +31 @@
 #define UNLOCK_GLOBAL   mutexUnlock( ut->env->mutex );
 
+#define LOCK_ATOMS      LOCK_GLOBAL
+#define UNLOCK_ATOMS    UNLOCK_GLOBAL
+
 
 typedef struct
@@ -42 +45 @@
 }
 AtomRec;
+
+
+/*
+typedef struct
+{
+    uint8_t  type;
+    uint8_t  flags;
+    uint16_t _pad0;
+    uint32_t _pad1;
+    OSThread thread;
+}
+UCellThread;
+*/
 
 

branches/thune/thread_safe/project.r

@@ -93 +93 @@
     ]
     unix  [
-        libs %m
-        libs %bz2
+        libs {m bz2 pthread}
     ]
     win32 [

branches/thune/thread_safe/read_config.r

@@ -17 +17 @@
        uds:
        threads:
+       tasks:
        macros:
        emh:

branches/thune/thread_safe/series.c

@@ -99 +99 @@
   Get pointers to the start and end of a series or slice.
 
-  \returns
+  \return
     Pointer to UBinary if cell is a series or slice.
     Otherwise, zero is returned.

branches/thune/thread_safe/thread.c

@@ -38 +38 @@
 
 
+extern UCellContext ur_thrGlobal;
+
 UThread* ur_threadMake( UrlanEnv* env, int binCount, int blkCount )
 {
@@ -53 +55 @@
         if( env->threads )
         {
+#if 0
             UThread* link = env->threads;
             ut->nextThread = link->nextThread;
             link->nextThread = ut;
+#endif
         }
         else
@@ -71 +75 @@
         //ur_hold( ut, UT_BINARY, binN );       // Hardcoded in recycle
         }
+
+        // Make hold & global context blocks.
+        ur_makeBlock( 8 );        // 0 - BLK_THREAD_HOLD
+        ur_makeBlock( 1024 );     // 1 - BLK_GLOBAL_WORD
+        ur_makeBlock( 1024 );     // 2 - BLK_GLOBAL_VAL
+        ur_makeBlock( 1 );        // 3 - BLK_CTX_STACK
+        //ur_makeBlock( 0 );        // 4 - BLK_DSTACK
+
+        ur_pushContext( ut, (UCell*) &ur_thrGlobal );
     }
     return ut;
@@ -76 +89 @@
 
 
-#ifdef UR_CONFIG_THREADS
+/*
 void ur_threadUnlink( UThread* th )
 {
@@ -82 +95 @@
     it = ((UrlanEnv*) th->env)->threads;
     if( it == th )
-        ((UrlanEnv*) th->env)->threads = (UThread*) th->nextThread;
-    while( it->nextThread != th )
-        it = it->nextThread;
-    it->nextThread = th->nextThread;
+        ((UrlanEnv*) th->env)->threads = (UThread*) th->nextTask;
+    while( it->nextTask != th )
+        it = it->nextTask;
+    it->nextTask = th->nextTask;
+}
+*/
+
+
+#ifdef UR_CONFIG_TASKS
+void ur_taskUnlink( UThread* th )
+{
+    UThread* it;
+    it = ((UrlanEnv*) th->env)->threads;
+    if( it == th )
+        ((UrlanEnv*) th->env)->threads = (UThread*) th->nextTask;
+    while( it->nextTask != th )
+        it = it->nextTask;
+    it->nextTask = th->nextTask;
 }
 
 
 // Returns zero and sets error if no threads are ready.
-UThread* ur_threadNextReady( UThread* cur )
+UThread* ur_taskNextReady( UThread* cur )
 {
     UThread* next = cur;
@@ -96 +123 @@
     do
     {
-        next = next->nextThread;
-        if( next->state == UR_THREAD_READY )
-        {
-            next->state = UR_THREAD_RUNNING;
+        next = next->nextTask;
+        if( next->state == UR_TASK_READY )
+        {
+            next->state = UR_TASK_RUNNING;
             return next;
         }
-        else if( next->state == UR_THREAD_BLOCKED )
+        else if( next->state == UR_TASK_BLOCKED )
         {
             UCell* pc;
@@ -113 +140 @@
                 if( pc && ! ur_is(pc, UT_UNSET) )
                 {
-                    next->state = UR_THREAD_RUNNING;
+                    next->state = UR_TASK_RUNNING;
                     return next;
                 }
@@ -121 +148 @@
     while( next != cur );
 
-    ur_throwErr( cur, UR_EX_SCRIPT, "All threads are blocked" ); 
+    ur_throwErr( cur, UR_EX_SCRIPT, "All tasks are blocked" ); 
     return 0;
 }
@@ -145 +172 @@
     */
 
-    th->state  = UR_THREAD_READY;
+    th->state  = UR_TASK_READY;
     th->callOp = 0;
     th->tos = th->dstack;
@@ -183 +210 @@
 
 /**
-  \returns cell pointer to TOS.
+  \return  cell pointer to TOS.
   Pops TOS if pop is non-zero.
 */

branches/thune/thread_safe/thune.c

@@ -11 +11 @@
 extern int ur_lessOrEqual( const UCell*, const UCell* );
 
-#ifdef UR_CONFIG_THREADS
-extern void ur_threadUnlink( UThread* );
-extern UThread* ur_threadNextReady( UThread* );
+#ifdef UR_CONFIG_TASKS
+extern void ur_taskUnlink( UThread* );
+extern UThread* ur_taskNextReady( UThread* );
 #endif
 
@@ -164 +164 @@
 
 #ifdef UR_CONFIG_THREADS
-// (block -- )
+#ifdef _WIN32
+static DWORD WINAPI threadRoutine( LPVOID arg )
+#else
+static void* threadRoutine( void* arg )
+#endif
+{
+    UThread* ut = (UThread*) arg;
+    ur_eval( ut, ut->tos[1].series.n, 0 );
+    ur_threadFree( ut );
+    return 0;
+}
+
+/*
+static void ur_threadCloneCell( UThread* ut, UCell* cellA,
+                                UThread* to, UCell* cellB )
+{
+    UIndex newBlkN;
+
+    switch( ur_type(cell) )
+    {
+        case UT_BLOCK:
+        {
+            UBlock* blkA;
+            UBlock* blkB;
+            blkA = ur_blockPtr( blkN );
+            newBlkN = ur_makeBlockT( to, blkA->used, &blkB ); 
+        }
+            break;
+
+        case UT_STRING:
+            break;
+    }
+}
+*/
+
+
+// (code -- )
 UR_CALL( uc_thread )
 {
+    if( ur_is(tos, UT_BLOCK) || ur_is(tos, UT_STRING) )
+    {
+        UThread* thr = ur_threadMake( ut->env, 64, 128 );
+        if( thr )
+        {
+            OSThread pth;
+#ifdef _WIN32
+            DWORD winId;
+#endif
+
+            if( ur_is(tos, UT_STRING) )
+            {
+                char* cpA;
+                char* cpB;
+                ur_seriesMem( ut, tos, &cpA, &cpB );
+                thr->tos[1].series.n = ur_tokenize( thr, cpA, cpB );
+            }
+
+#ifdef _WIN32
+            pth = CreateThread( NULL, 0, threadRoutine, thr, 0, &winId );
+            if( pth == NULL )
+#else
+            if( pthread_create( &pth, 0, threadRoutine, thr ) != 0 )
+#endif
+            {
+                ur_throwErr( UR_ERR_INTERNAL, "Could not create thread" );
+                return;
+            }
+
+#if 1
+            {
+            void* status;
+            pthread_join( pth, &status ); //KR
+            }
+#endif
+        }
+        UR_S_DROP;
+    }
+    else
+    {
+        ur_throwErr( UR_ERR_DATATYPE, "thread expected block!/string!" );
+    }
+}
+#endif
+
+
+#ifdef UR_CONFIG_TASKS
+// (block -- )
+UR_CALL( uc_task )
+{
     if( ur_is(tos, UT_BLOCK) )
     {
         UThread* thr;
         UThread* save;
-        thr = ur_threadMake( ut->env );
+        thr = ur_threadMake( ut->env, 32, 32 );
         if( thr )
         {
@@ -307 +393 @@
     UCell* pc;
     UCell* end;
-#ifdef UR_CONFIG_THREADS
+#ifdef UR_CONFIG_TASKS
     int insCycle = 0;
 #endif
@@ -323 +409 @@
 execute:
 
-#ifdef UR_CONFIG_THREADS
+#ifdef UR_CONFIG_TASKS
     if( ++insCycle == 100 )
     {
-        if( ut->nextThread != ut )
-        {
-            ut->state = UR_THREAD_READY;
+        if( ut->nextTask != ut )
+        {
+            ut->state = UR_TASK_READY;
 switch_thread:
             PUSHC_EVAL( blkN, start, pc );
-            ut = ur_threadNextReady( ut );
+            ut = ur_taskNextReady( ut );
             if( ! ut )
                 goto error;
@@ -363 +449 @@
         case UT_UNSET:
             --pc;
-#ifdef UR_CONFIG_THREADS
-            if( ut->nextThread != ut )
-            {
-                ut->state = UR_THREAD_BLOCKED;
+#ifdef UR_CONFIG_TASKS
+            if( ut->nextTask != ut )
+            {
+                ut->state = UR_TASK_BLOCKED;
                 goto switch_thread;
             }
@@ -956 +1042 @@
             goto execute;
 
-#ifdef UR_CONFIG_THREADS
+#ifdef UR_CONFIG_TASKS
         case CC_TERM:
-            if( ut->nextThread == ut )
+            if( ut->nextTask == ut )
                 goto finish;
-term_thread:
+term_task:
             {
             UThread* thr = ut;
-            ut->state = UR_THREAD_TERM;
-            ut = ur_threadNextReady( ut );
+            ut->state = UR_TASK_TERM;
+            ut = ur_taskNextReady( ut );
             if( ! ut )
                 goto error;
-            ur_threadUnlink( thr );
+            ur_taskUnlink( thr );
             ur_threadFree( thr );
             }
@@ -974 +1060 @@
 
         case CC_END:
-#ifdef UR_CONFIG_THREADS
-            if( ut->nextThread != ut )
-                goto term_thread;
+#ifdef UR_CONFIG_TASKS
+            if( ut->nextTask != ut )
+                goto term_task;
 #endif
             goto finish;

branches/thune/thread_safe/tokenize.c

@@ -299 +299 @@
 
 /**
- \return  Returns block index or zero if an error is thrown.
+ \return  Block index or zero if an error is thrown.
 */
 UIndex ur_tokenize( UThread* ut, const char* it, const char* end )

branches/thune/thread_safe/unix/os.h

@@ -59 +59 @@
 
 
+typedef pthread_t           OSThread;
 typedef pthread_mutex_t     OSMutex;
 
-#define mutexInit(mh)        pthread_mutex_init(&mh,0)
-#define mutexFree(mh)        pthread_mutex_destroy(&mh)
-#define mutexLock(mh)        pthread_mutex_lock(&mh)
-#define mutexUnlock(mh)      pthread_mutex_unlock(&mh)
+#define mutexInitF(mh)      (pthread_mutex_init(&mh,0) == -1)
+#define mutexFree(mh)       pthread_mutex_destroy(&mh)
+#define mutexLock(mh)       pthread_mutex_lock(&mh)
+#define mutexUnlock(mh)     pthread_mutex_unlock(&mh)
 
 

branches/thune/thread_safe/urlan.c

@@ -411 +411 @@
     int i;
 
+    //printf( "KR sizeof(pthread_t) = %lu\n", sizeof(pthread_t) );
+    //assert( sizeof(UCellThread) == 16 );
+
     assert( sizeof(UCell) == 16 );
     //assert( sizeof(LocalFrame) == 16 );
@@ -451 +454 @@
   \param customCount  Number of UDatatype in custom array.
 
-  \returns UR_EVAL_ code.
+  \return  UR_EVAL_ code.
 */
 int ur_startup( UrlanEnv* env, UDatatype* custom, int customCount )
@@ -467 +470 @@
     env->dtCount  = UT_BI_COUNT + customCount;
 
-    ur_arrayInit( &env->atoms,     sizeof(AtomRec),      4096 );
-    ur_arrayInit( &env->atomNames, sizeof(char),         2048 );
+    ur_arrayInit( &env->atoms,     sizeof(AtomRec),      2048 );
+    ur_arrayInit( &env->atomNames, sizeof(char),         2048 * 6 );
     ur_arrayInit( &env->devices,   sizeof(UPortDevice*),    8 );
     ur_rebuildAtomHash( &env->atoms );
@@ -485 +488 @@
     // Init main thread.
 
-    if( mutexInit( env->mutex ) == -1 )
-    {
-#ifndef _WIN32
+    if( mutexInitF( env->mutex ) )
+    {
+#ifdef _WIN32
+        //GetLastError();
+#else
         perror( "mutexInit" );
 #endif
@@ -495 +500 @@
     ut = ur_threadMake( env, 512, 1024 );
 
-    // Make hold & global context blocks.
-    ur_makeBlock( 8 );        // 0 - BLK_THREAD_HOLD
-    ur_makeBlock( 1024 );     // 1 - BLK_GLOBAL_WORD
-    ur_makeBlock( 1024 );     // 2 - BLK_GLOBAL_VAL
-    ur_makeBlock( 1 );        // 3 - BLK_CTX_STACK
-    //ur_makeBlock( 0 );        // 4 - BLK_DSTACK
-
-    ur_pushContext( ut, (UCell*) &ur_thrGlobal );
-
-    {
-    //UBlock* blk = ur_tblockPtr( BLK_DSTACK );
-    }
 
     // Add datatypes first so UT_ defines match the index & atom.