regina/threader.c

   1 /* Asynchroneous thread multiplexer with a parallel use of the REXXSAA API.
   2  * This module is under construction.
   3  */
   4 #include <stdio.h>
   5 #include <string.h>
   6 #include <stdlib.h>
   7 #include <limits.h>
   8 #include <errno.h>
   9 #include <time.h>
  10
  11 #define INCL_RXSHV
  12 #define INCL_RXFUNC
  13 #define INCL_RXSYSEXIT
  14 #define INCL_RXSUBCOM
  15
  16 #ifdef POSIX_THREADS
  17 # include <sys/time.h>
  18 # include <unistd.h>
  19 # include <pthread.h>
  20 #endif
  21
  22 #ifdef _MSC_VER
  23 /* This picky compiler claims about unused formal parameters.
  24  * This is correct but hides (for human eyes) other errors since they
  25  * are many and we can't reduce them all.
  26  * Error 4100 is "unused formal parameter".
  27  * Error 4115 is generated in the rameter".
  28  */
  29 # pragma warning(disable:4100)
  30 #endif
  31
  32 #ifdef WIN32_THREADS
  33 # define WIN32_LEAN_AND_MEAN
  34 # include <process.h>
  35 # include <windows.h>
  36 # include <io.h>
  37 #endif
  38
  39 #ifdef _MSC_VER
  40 # pragma warning(default:4115)
  41 #endif
  42
  43 #include "rexxsaa.h"
  44
  45 #define MAX_THREADS 6  /* parallel starting threads, 20 is a good maximum */
  46 #define MAX_RUN (sizeof(unsigned) * CHAR_BIT)
  47 #define TOTAL_THREADS 2000  /* total count of threads */
  48
  49 #ifdef POSIX_THREADS
  50 #define ThreadIndexType pthread_t
  51 #define my_threadidx() pthread_self()
  52 #define my_threadid() pthread_self()
  53 #define THREAD_RETURN void *
  54 #define THREAD_CONVENTION
  55 static pthread_t thread[MAX_THREADS];
  56 #endif
  57
  58 #ifdef WIN32_THREADS
  59 #define ThreadIndexType DWORD
  60 #define my_threadidx() GetCurrentThreadId()
  61 #define my_threadid() GetCurrentThread()
  62 #define THREAD_RETURN unsigned
  63 #define THREAD_CONVENTION __stdcall
  64 static HANDLE thread[MAX_THREADS];
  65 #endif
  66
  67 static ThreadIndexType threadx[MAX_THREADS];
  68 static enum {Ready = 0,Running,Stopped} State[MAX_THREADS];
  69 static unsigned found[MAX_THREADS]; /* which lines we have seen (bitwise). */
  70 static int GlobalError = 0; /* int should be threadsafe by default */
  71 static int stdout_is_tty = 0;
  72 static enum {UnUsed = 0, FirstRun, DoInstore} UseInstore = UnUsed;
  73 static void *InstoreBuf = NULL;
  74 static unsigned InstoreLen;
  75
  76 static void ThreadHasStopped(unsigned position);
  77
  78 /* We redirect the output. This is the redirection handler. */
  79 LONG APIENTRY instore_exit( LONG ExNum, LONG Subfun, PEXIT PBlock )
  80 {
  81    RXSIOSAY_PARM *psiosay;
  82    char buf[256]; /* enough to hold the data */
  83    RXSTRING rx;
  84    unsigned len,tid,loop;
  85    int rc,idx;
  86
  87    if ((ExNum != RXSIO) || (Subfun != RXSIOSAY)) /* unexpected? */
  88       return RXEXIT_NOT_HANDLED;
  89
  90    if (GlobalError)
  91       return RXEXIT_HANDLED;
  92
  93    psiosay = (RXSIOSAY_PARM *)PBlock;
  94    rx = psiosay->rxsio_string;
  95    if (!RXVALIDSTRING(rx))
  96       {
  97          fprintf(stderr,"\n"
  98                         "Thread %lu gives an invalid string for a SAY output.\n",
  99                         (unsigned long) my_threadidx());
 100          GlobalError = 1;
 101          return RXEXIT_HANDLED ;
 102       }
 103
 104    len = RXSTRLEN(rx);
 105    if (len >= sizeof(buf))
 106       len = sizeof(buf) - 1;
 107    memcpy(buf,RXSTRPTR(rx),len);
 108    buf[len] = '\0'; /* We have a sscanf-able string */
 109
 110    rc = sscanf(buf,"Loop %u in thread %u",&loop,&tid);
 111    if (rc != 2)
 112       {
 113          fprintf(stderr,"\n"
 114                         "Thread %lu gives an unexpected SAY output \"%s\".\n",
 115                         (unsigned long) my_threadidx(),buf);
 116          GlobalError = 1;
 117          return RXEXIT_HANDLED ;
 118       }
 119
 120    if ((ThreadIndexType) tid != my_threadidx())
 121       {
 122          fprintf(stderr,"\n"
 123                         "Thread %lu claims an incorrect thread identifier %lu.\n",
 124                         (unsigned long) my_threadidx(),
 125                         (unsigned long) tid);
 126          GlobalError = 1;
 127          return RXEXIT_HANDLED ;
 128       }
 129
 130    for (idx = 0;idx < MAX_THREADS;idx++)
 131       if (threadx[idx] == (ThreadIndexType) tid)
 132          break;
 133
 134    if (idx >= MAX_THREADS)
 135       {
 136          fprintf(stderr,"\n"
 137                         "Thread %lu can't be found in the thread table.\n",
 138                         (unsigned long) my_threadidx());
 139          GlobalError = 1;
 140          return RXEXIT_HANDLED ;
 141       }
 142
 143    /* Check the loop number */
 144    rc = 0; /* OK */
 145    if ((loop < 1) || (loop > MAX_RUN))
 146       rc = 1;
 147    else
 148    {
 149       loop = 1 << (loop - 1); /* Bitmask for the loop */
 150       if (found[idx] & loop)
 151          rc = 1;
 152       found[idx] |= loop;
 153    }
 154    if (rc)
 155       {
 156          fprintf(stderr,"\n"
 157                         "Thread %lu's loop doesn't run continuously.\n",
 158                         (unsigned long) my_threadidx());
 159          GlobalError = 1;
 160       }
 161
 162    return RXEXIT_HANDLED ;
 163 }
 164
 165 THREAD_RETURN THREAD_CONVENTION instore( void *data )
 166 {
 167    RXSTRING Instore[2] ;
 168    RXSYSEXIT Exits[2] ;
 169    char instore_buf[256];
 170    int rc ;
 171
 172    threadx[(unsigned) data] = my_threadidx();
 173    RexxRegisterExitExe( "ExitHandler",
 174 #ifdef RX_STRONGTYPING
 175                         instore_exit,
 176 #else
 177                         (PFN) instore_exit,
 178 #endif
 179                         NULL ) ;
 180
 181    Exits[0].sysexit_name = "ExitHandler" ;
 182    Exits[0].sysexit_code = RXSIO ;
 183    Exits[1].sysexit_code = RXENDLST ;
 184
 185    sprintf(instore_buf,"Do i = 1 To %u;"
 186                        "say 'Loop' i 'in thread' gettid();"
 187                        "End;"
 188                        "Return 0",
 189                        MAX_RUN);
 190    Instore[0].strptr = instore_buf;
 191    Instore[0].strlength = strlen( Instore[0].strptr ) ;
 192    if (UseInstore == DoInstore)
 193    {
 194       Instore[1].strptr = InstoreBuf ;
 195       Instore[1].strlength = InstoreLen ;
 196    }
 197    else
 198       Instore[1].strptr = NULL ;
 199    rc = RexxStart( 0, NULL, "Testing", Instore, "Foo", RXCOMMAND,
 200                    Exits, NULL, NULL ) ;
 201    switch (UseInstore) {
 202       case UnUsed:
 203          if ( Instore[1].strptr )
 204             RexxFreeMemory( Instore[1].strptr ) ;
 205          else
 206          {
 207             GlobalError = 1;
 208             fprintf(stderr,"\n"
 209                            "Not got the instore macro.\n");
 210          }
 211          break;
 212       case FirstRun:
 213          if ( Instore[1].strptr )
 214          {
 215             InstoreBuf = Instore[1].strptr ;
 216             InstoreLen = Instore[1].strlength ;
 217          }
 218          else
 219          {
 220             GlobalError = 1;
 221             fprintf(stderr,"\n"
 222                            "Not got the instore macro.\n");
 223          }
 224          break;
 225       default:
 226          /* I don't know if the standard allows a success and a return value
 227           * of NULL in Instore[1]. Ignore it. It will be detected later.
 228           * True application should check the return code of RexxStart().
 229           */
 230          break;
 231    }
 232    RexxDeregisterExit("ExitHandler",NULL);
 233    ThreadHasStopped((unsigned) data);
 234 #ifndef THREAD_RETURN_VOID
 235    return((THREAD_RETURN) 0);
 236 #endif
 237 }
 238
 239 void reap(unsigned position)
 240 {
 241    if (found[position] != ~((unsigned) 0))
 242    {
 243       fprintf(stderr,"\n"
 244                      "Thread %lu has stopped without completing its loop.\n",
 245                      threadx[position]);
 246       GlobalError = 1;
 247    }
 248    found[position] = 0;
 249 }
 250
 251 #ifdef WIN32_THREADS
 252 int start_a_thread( unsigned position )
 253 {
 254    int rc;
 255    unsigned threadID;
 256
 257    State[position] = Running;
 258    /* Avoid some race conditions. I don't know if this is a problem of the
 259     * runtime system or the kernel. If the systems runs into severe swapping
 260     * the threads seems to run before the thread id is known which is used
 261     * in instore_exit. We suspend the thread until all details of the new
 262     * thread are known before we continue. This gives a little bit worser
 263     * performance.
 264     */
 265    thread[position] = (HANDLE) _beginthreadex( NULL,
 266                                                0,
 267                                                instore,
 268                                                (void *) position,
 269                                                CREATE_SUSPENDED,
 270                                                &threadID );
 271    rc = (long) thread[position] != 0l;
 272    if (!rc)
 273    {
 274       fprintf(stderr,"\n"
 275                      "Error starting thread, error code is %ld\n",
 276                      GetLastError() );
 277       GlobalError = 1;
 278       ThreadHasStopped(position);
 279    }
 280    ResumeThread(thread[position]);
 281    return(rc);
 282 }
 283
 284 static void ThreadHasStopped(unsigned position)
 285 {
 286    State[position] = Stopped;
 287 }
 288
 289 void wait_for_threads( void )
 290 {
 291    unsigned i,j,done,running;
 292    DWORD rc;
 293    HANDLE compressed[MAX_THREADS];
 294    unsigned BackSort[MAX_THREADS];
 295
 296    running = done = MAX_THREADS;
 297    if (stdout_is_tty)
 298       printf("%u\r",MAX_THREADS); /* already started */
 299    if (GlobalError)
 300       return;
 301
 302    for (;;)
 303    {
 304       for (i = 0,j = 0;i < MAX_THREADS;i++)
 305          if (State[i] != Ready)
 306          {
 307             compressed[j] = thread[i];
 308             BackSort[j] = i;
 309             j++;
 310          }
 311       rc = WaitForMultipleObjects(running, compressed, FALSE, 3000);
 312       if (rc == 0xFFFFFFFF)
 313       { /* Failure or dead thread, look for a stopped one */
 314          for (i = 0;i < running;i++)
 315             if (State[BackSort[i]] == Stopped)
 316                rc = WAIT_OBJECT_0 + i;
 317       }
 318       if ((rc < WAIT_OBJECT_0) || (rc >= running + WAIT_OBJECT_0))
 319       {
 320          fprintf(stderr,"\n"
 321                         "At least one thread won't finish normally within 3 seconds (rc=%u,error=%lu).\n",
 322                         rc,GetLastError());
 323          GlobalError = 1;
 324       }
 325       if (GlobalError)
 326          break;
 327       i = BackSort[rc - WAIT_OBJECT_0];
 328       if (State[i] != Stopped)
 329       {
 330          fprintf(stderr,"\n"
 331                         "Thread %u hasn't finished normally.\n");
 332          GlobalError = 1;
 333          break;
 334       }
 335       CloseHandle(thread[i]);
 336       State[i] = Ready;
 337       running--;
 338       reap(i);
 339       if (done < TOTAL_THREADS)
 340       {
 341          if (!start_a_thread(i))
 342             break;
 343          done++;
 344          running++;
 345       }
 346       if (stdout_is_tty)
 347          printf("%u(%u)\r",done,running);
 348       if (GlobalError || !running)
 349          break;
 350    }
 351 }
 352 #endif
 353
 354 #ifdef POSIX_THREADS
 355 static pthread_mutex_t thread_lock = PTHREAD_MUTEX_INITIALIZER;
 356 static pthread_cond_t something_stopped = PTHREAD_COND_INITIALIZER;
 357
 358 int start_a_thread( unsigned position )
 359 {
 360    int rc;
 361
 362    State[position] = Running;
 363    rc = pthread_create( &thread[position], NULL, instore, (void *) position );
 364    if (rc)
 365    {
 366       fprintf(stderr,"\n"
 367                      "Error starting thread, error code is %d\n",
 368                      rc );
 369       GlobalError = 1;
 370       ThreadHasStopped(position);
 371    }
 372    return(!rc);
 373 }
 374
 375 static void ThreadHasStopped(unsigned position)
 376 {
 377    pthread_mutex_lock(&thread_lock);
 378    State[position] = Stopped;
 379    pthread_cond_signal(&something_stopped);
 380    pthread_mutex_unlock(&thread_lock);
 381 }
 382
 383 void wait_for_threads( void )
 384 {
 385    unsigned i,done,running;
 386    int rc;
 387    struct timeval now;
 388    struct timespec timeout;
 389
 390    running = done = MAX_THREADS;
 391    if (stdout_is_tty)
 392       printf("%u\r",MAX_THREADS); /* already started */
 393
 394    if (GlobalError)
 395       return;
 396
 397    pthread_mutex_lock(&thread_lock);
 398    for (;;)
 399    {
 400       gettimeofday(&now,NULL);
 401       timeout.tv_sec = now.tv_sec + 3;
 402       timeout.tv_nsec = now.tv_usec * 1000;
 403       rc = pthread_cond_timedwait(&something_stopped,&thread_lock,&timeout);
 404       if (rc == ETIMEDOUT)
 405       {
 406          fprintf(stderr,"\n"
 407                         "At least one thread won't finish within 3 seconds.\n");
 408          GlobalError = 1;
 409       }
 410       if (GlobalError)
 411          break;
 412       for ( i = 0; i < MAX_THREADS; i++ )
 413       {
 414          if (State[i] != Stopped)
 415             continue;
 416          State[i] = Ready;
 417          running--;
 418          rc = pthread_join( thread[i], NULL );
 419          if (rc != 0)
 420          {
 421             fprintf(stderr,"\n"
 422                            "A thread can't be found in the internal table.\n");
 423             GlobalError = 1;
 424             break;
 425          }
 426          reap(i);
 427          if (done < TOTAL_THREADS)
 428          {
 429             if (!start_a_thread(i))
 430                break;
 431             done++;
 432             running++;
 433          }
 434       }
 435       if (stdout_is_tty)
 436          printf("%u(%u)\r",done,running);
 437       if (GlobalError || !running)
 438          break;
 439       usleep(10);
 440    }
 441    pthread_mutex_unlock(&thread_lock);
 442 }
 443 #endif
 444
 445 static void usage(void)
 446 {
 447    printf("usage: threader [-p]\n"
 448           "\n"
 449           "Options:\n"
 450           "-p\tLoad the macro only once and then use the generated instore\n"
 451           "\tmacro. Default: Always load the macro new.\n"
 452           "\n"
 453           "threader uses a macro which generated lines with numbers which can\n"
 454           "be parsed to detect problems in the multi-threading implementation.\n"
 455           "A loop counter runs until %u. The test should run within a few\n"
 456           "seconds until a few minutes. You should hit ^C to abort the program\n"
 457           "if you think your harddisk is working heavy.\n"
 458           "\n"
 459           "This program is for testing purpose only.\n"
 460           ,TOTAL_THREADS);
 461    exit(1);
 462 }
 463
 464 int main( int argc, char *argv[] )
 465 {
 466    RXSTRING version;
 467    ULONG versioncode;
 468    int i;
 469    time_t start;
 470
 471    if (isatty(fileno(stdout)))
 472       stdout_is_tty = 1;
 473
 474    if (argc > 2)
 475       usage();
 476    if (argc > 1)
 477    {
 478       if (strcmp(argv[1],"-p") == 0)
 479          UseInstore = FirstRun;
 480         else
 481          usage();
 482    }
 483
 484    memset(found,0,sizeof(found));
 485    memset(State,0,sizeof(State));
 486    /* We want to see surprisings at once: */
 487    setvbuf(stdout,NULL,_IONBF,0);
 488    setvbuf(stderr,NULL,_IONBF,0);
 489    printf( "Regina Rexx Thread Tester\n" );
 490    printf( "-------------------------\n" );
 491
 492    version.strlength = 0;
 493    version.strptr = NULL;
 494    /* This might not work if we check another Rexx: */
 495    versioncode = ReginaVersion(&version);
 496    printf("Regina's version is %lu.%02lu",
 497           versioncode >> 8,
 498           versioncode & 0xFF);
 499    if (version.strptr)
 500    {
 501       printf(" (in complete \"%s\")",version.strptr);
 502       RexxFreeMemory(version.strptr);
 503    }
 504    printf("\n");
 505
 506    if (UseInstore)
 507    {
 508       State[0] = Running;
 509       thread[0] = my_threadid();
 510       threadx[0] = my_threadidx();
 511       instore(NULL);
 512       State[0] = Stopped;
 513       found[0] = 0;
 514       UseInstore = DoInstore;
 515    }
 516
 517    if (stdout_is_tty)
 518       printf("\n"
 519              "You should see a loop counter which stops at %u.\n\n",
 520              TOTAL_THREADS);
 521
 522    start = time(NULL);
 523    for ( i = 0; i < MAX_THREADS; i++ )
 524       if (!start_a_thread(i) || GlobalError)
 525          break;
 526
 527    if (!GlobalError)
 528       wait_for_threads();
 529
 530    if (GlobalError)
 531    {
 532       fprintf(stderr,"\n"
 533                      "An error encountered. Do you use the right shared "
 534                                                             "libs or DLLs?\n");
 535       return 1;
 536    }
 537
 538    printf("\n"
 539           "Thread tester passed without an error.\n"
 540           "About %u seconds used for %u cyles.\n",
 541           (unsigned) (time(NULL) - start),TOTAL_THREADS);
 542
 543    printf("Press ENTER to continue...\n");
 544    {
 545       char buf[128];
 546       fgets(buf,sizeof(buf),stdin);
 547    }
 548    return 0 ;
 549 }