src/ptlib/unix/tlibvx.cxx

   1 /*
   2  * tlibvx.cxx
   3  *
   4  * Thread library implementation for VxWorks
   5 */
   6
   7
   8 class PProcess;
   9 class PSemaphore;
  10
  11 #include <ptlib.h>
  12
  13 #define VX_LOWEST_PRIORITY          250
  14 #define VX_LOW_PRIORITY             200
  15 #define VX_NORMAL_PRIORITY          150
  16 #define VX_DISPLAY_PRIORITY         100
  17 #define VX_URGENT_DISPLAY_PRIORITY  50
  18
  19 ///////////////////////////////////////////////////////////////////////////////
  20 // Threads
  21 static int const priorities[] = {
  22   VX_LOWEST_PRIORITY,
  23   VX_LOW_PRIORITY,
  24   VX_NORMAL_PRIORITY,
  25   VX_DISPLAY_PRIORITY,
  26   VX_URGENT_DISPLAY_PRIORITY
  27 };
  28
  29 int PThread::ThreadFunction(void *threadPtr)
  30 {
  31         PAssertNULL(threadPtr);
  32
  33   PThread * thread = (PThread *)threadPtr;
  34
  35   PProcess & process = PProcess::Current();
  36   process.activeThreadMutex.Wait();
  37   process.activeThreads.SetAt(thread->PX_threadId, thread);
  38   process.activeThreadMutex.Signal();
  39
  40   process.SignalTimerChange();
  41
  42   thread->Main();
  43
  44   return 0;
  45 }
  46
  47
  48 PThread::PThread()
  49  : PX_threadId(ERROR),
  50    priority(VX_NORMAL_PRIORITY),
  51    originalStackSize(0)
  52 {
  53 }
  54
  55 PThread::PThread(PINDEX stackSize,
  56                  AutoDeleteFlag deletion,
  57                  Priority priorityLevel,
  58                    const PString & name
  59 )
  60 {
  61   PAssert(stackSize > 0, PInvalidParameter);
  62   autoDelete = deletion == AutoDeleteThread;
  63   originalStackSize = stackSize;
  64
  65   priority = priorities[priorityLevel];
  66
  67   PX_threadId =  ::taskSpawn(
  68                                         name,                         // Name
  69                                         priority,                     // Priority
  70                                         0,                            // options
  71                                         stackSize,                    // stacksize
  72                                         (FUNCPTR)ThreadFunction,      // entrypoint
  73                                         (int)this,0,0,0,0,0,0,0,0,0); // arg 1 --- arg 10
  74
  75
  76   PAssertOS(PX_threadId != ERROR);
  77   // threads are created suspended
  78   Suspend();
  79   if (autoDelete) {
  80     PProcess & process = PProcess::Current();
  81     process.deleteThreadMutex.Wait();
  82     process.autoDeleteThreads.Append(this);
  83     process.deleteThreadMutex.Signal();
  84   }
  85 }
  86
  87
  88 PThread::~PThread()
  89 {
  90   if (originalStackSize <= 0)
  91     return;
  92
  93   PProcess & process = PProcess::Current();
  94   process.activeThreadMutex.Wait();
  95   process.activeThreads.SetAt(PX_threadId, NULL);
  96   process.activeThreadMutex.Signal();
  97
  98   if (!IsTerminated())
  99     Terminate();
 100 }
 101
 102
 103 void PThread::Restart()
 104 {
 105   char *name = ::taskName(PX_threadId);
 106   PAssert(IsTerminated(), "Cannot restart running thread");
 107
 108   PX_threadId =  ::taskSpawn(
 109                                         name,                                                                               // Name
 110                                         priority,                                                                               // Priority
 111                                         0,                                                                                                              // options
 112                                         originalStackSize,            // stacksize
 113                                         (FUNCPTR)ThreadFunction,                        // entrypoint
 114                                         (int)this,0,0,0,0,0,0,0,0,0);   // arg 1 --- arg 10
 115
 116   PAssertOS(PX_threadId != ERROR);
 117 }
 118
 119
 120 void PThread::Terminate()
 121 {
 122         PAssert(!IsTerminated(), "Cannot terminate a thread which is already terminated");
 123         PAssert(originalStackSize > 0, PLogicError);
 124
 125         ::taskDelete(PX_threadId);
 126 }
 127
 128
 129 BOOL PThread::IsTerminated() const
 130 {
 131                 STATUS stat = taskIdVerify(PX_threadId);
 132                 return stat == ERROR;
 133 }
 134
 135
 136 void PThread::WaitForTermination() const
 137 {
 138   while (!IsTerminated()) {
 139     Current()->Sleep(100);
 140         }
 141 }
 142
 143
 144 BOOL PThread::WaitForTermination(const PTimeInterval & maxWait) const
 145 {
 146   if (PX_threadId == 0)
 147     return TRUE;
 148
 149   PTimer timeout = maxWait;
 150   while (!IsTerminated()) {
 151     if (timeout == 0)   {
 152       return FALSE;
 153                 }
 154     Current()->Sleep(100);
 155   }
 156  return TRUE;
 157 }
 158
 159
 160 void PThread::Suspend(BOOL susp)
 161 {
 162   PAssert(!IsTerminated(), "Operation on terminated thread");
 163   if (susp)
 164   {
 165     STATUS result = ::taskSuspend(PX_threadId);
 166
 167     PAssert(result != ERROR, "Thread don't want to be suspended");
 168   }
 169   else {
 170     Resume();
 171         }
 172 }
 173
 174
 175 void PThread::Resume()
 176 {
 177   PAssert(!IsTerminated(), "Operation on terminated thread");
 178         if (!IsTerminated()) {
 179           STATUS result = ::taskResume(PX_threadId);
 180           PAssert(result != ERROR, "Thread doesn't want to resume");
 181         }
 182 }
 183
 184
 185 BOOL PThread::IsSuspended() const
 186 {
 187   struct TASK_DESC info;
 188   STATUS result = ::taskInfoGet(PX_threadId, &info);
 189
 190   PAssert(result != ERROR || PX_threadId != info.td_id, "Thread info inaccessible");
 191   return taskIsSuspended(PX_threadId);
 192 }
 193
 194 void PThread::SetAutoDelete(AutoDeleteFlag deletion)
 195 {
 196   PAssert(deletion != AutoDeleteThread || this != &PProcess::Current(), PLogicError);
 197   autoDelete = deletion == AutoDeleteThread;
 198 }
 199
 200 void PThread::SetPriority(Priority priorityLevel)
 201 {
 202   PAssert(!IsTerminated(), "Operation on terminated thread");
 203
 204   priority = priorities[priorityLevel];
 205   STATUS result = ::taskPrioritySet(PX_threadId, priority );
 206
 207   PAssert(result != ERROR, "Thread priority change error");
 208 }
 209
 210
 211 PThread::Priority PThread::GetPriority() const
 212 {
 213   PAssert(!IsTerminated(), "Operation on terminated thread");
 214
 215         int prio;
 216         taskPriorityGet(PX_threadId, &prio);
 217
 218   switch (prio) {
 219     case VX_LOWEST_PRIORITY :
 220       return LowestPriority;
 221     case VX_LOW_PRIORITY :
 222       return LowPriority;
 223     case VX_NORMAL_PRIORITY :
 224       return NormalPriority;
 225     case VX_DISPLAY_PRIORITY :
 226       return HighPriority;
 227     case VX_URGENT_DISPLAY_PRIORITY :
 228       return HighestPriority;
 229   }
 230   PAssertAlways(POperatingSystemError);
 231   return LowestPriority;
 232 }
 233
 234 void PThread::Yield()
 235 {
 236   taskDelay(NO_WAIT);
 237 }
 238
 239 void PThread::Sleep( const PTimeInterval & delay ) // Time interval to sleep for in microsec.
 240 {
 241   taskDelay(delay.GetInterval()*sysClkRateGet()/1000);
 242 }
 243
 244 void PThread::InitialiseProcessThread()
 245 {
 246   originalStackSize = 0;
 247   autoDelete = FALSE;
 248
 249   PX_threadId = ::taskIdSelf();
 250   PAssertOS(PX_threadId >= OK);
 251
 252   ((PProcess *)this)->activeThreads.DisallowDeleteObjects();
 253   ((PProcess *)this)->activeThreads.SetAt(PX_threadId, this);
 254 }
 255
 256
 257 PThread * PThread::Current()
 258 {
 259   PProcess & process = PProcess::Current();
 260   process.activeThreadMutex.Wait();
 261
 262   PThread * thread = process.activeThreads.GetAt( taskIdSelf() );
 263
 264   process.activeThreadMutex.Signal();
 265   return thread;
 266 }
 267
 268 int PThread::PXBlockOnChildTerminate(int pid, const PTimeInterval & /*timeout*/) // Fix timeout
 269 {
 270
 271   while (!IsTerminated()) {
 272     Current()->Sleep(100);
 273   }
 274   return TRUE;
 275 }
 276
 277 int PThread::PXBlockOnIO(int handle, int type, const PTimeInterval & timeout)
 278 {
 279
 280   // make sure we flush the buffer before doing a write
 281   fd_set tmp_rfd, tmp_wfd, tmp_efd;
 282   fd_set * read_fds      = &tmp_rfd;
 283   fd_set * write_fds     = &tmp_wfd;
 284   fd_set * exception_fds = &tmp_efd;
 285
 286   FD_ZERO (read_fds);
 287   FD_ZERO (write_fds);
 288   FD_ZERO (exception_fds);
 289
 290   switch (type) {
 291     case PChannel::PXReadBlock:
 292     case PChannel::PXAcceptBlock:
 293       FD_SET (handle, read_fds);
 294       break;
 295     case PChannel::PXWriteBlock:
 296       FD_SET (handle, write_fds);
 297       break;
 298     case PChannel::PXConnectBlock:
 299       FD_SET (handle, write_fds);
 300       FD_SET (handle, exception_fds);
 301       break;
 302     default:
 303       PAssertAlways(PLogicError);
 304       return 0;
 305   }
 306
 307   struct timeval * tptr = NULL;
 308   struct timeval   timeout_val;
 309   if (timeout != PMaxTimeInterval) { // Clean up for infinite timeout
 310     static const PTimeInterval oneDay(0, 0, 0, 0, 1);
 311
 312     if (timeout < oneDay) {
 313
 314       timeout_val.tv_usec = (timeout.GetMilliSeconds() % 1000) * 1000;
 315       timeout_val.tv_sec  = timeout.GetSeconds();
 316       tptr                = &timeout_val;
 317     }
 318   }
 319   int retval = ::select(handle+1, read_fds, write_fds, exception_fds, tptr);
 320
 321   return retval;
 322 }
 323
 324 int PThread::PXBlockOnIO(int maxHandles,
 325            fd_set * readBits,
 326            fd_set * writeBits,
 327            fd_set * exceptionBits,
 328            const PTimeInterval & timeout,
 329            const PIntArray & /*osHandles*/)
 330 {
 331   // make sure we flush the buffer before doing a write
 332   fd_set * read_fds      = readBits;
 333   fd_set * write_fds     = writeBits;
 334   fd_set * exception_fds = exceptionBits;
 335
 336   struct timeval * tptr = NULL;
 337   struct timeval   timeout_val;
 338   if (timeout != PMaxTimeInterval) { // Clean up for infinite timeout
 339     static const PTimeInterval oneDay(0, 0, 0, 0, 1);
 340     if (timeout < oneDay) {
 341       timeout_val.tv_usec = (timeout.GetMilliSeconds() % 1000) * 1000;
 342       timeout_val.tv_sec  = timeout.GetSeconds();
 343       tptr                = &timeout_val;
 344     }
 345   }
 346   int retval = ::select(maxHandles, read_fds, write_fds, exception_fds, tptr);
 347
 348   return retval;
 349 }
 350
 351 void PThread::PXAbortBlock() const
 352 {
 353 }
 354
 355 ///////////////////////////////////////////////////////////////////////////////
 356 // PProcess::HouseKeepingThread
 357
 358 void PProcess::Construct()
 359 {
 360   // hard coded value, change this to handle more sockets at once with the select call
 361   maxHandles = 1024;
 362   houseKeeper=NULL;
 363   CommonConstruct();
 364 }
 365
 366 PProcess::HouseKeepingThread::HouseKeepingThread()
 367   : PThread(5000, NoAutoDeleteThread, LowPriority)
 368 {
 369   Resume();
 370 }
 371
 372 void PProcess::HouseKeepingThread::Main()
 373 {
 374         PProcess & process = PProcess::Current();
 375
 376         while(1) {
 377                 process.deleteThreadMutex.Wait();
 378                 for (PINDEX i = 0; i < process.autoDeleteThreads.GetSize(); i++)
 379                 {
 380                         PThread * pThread = (PThread *) process.autoDeleteThreads.GetAt(i);
 381                         if( pThread->IsTerminated() )
 382                         {
 383                                 process.autoDeleteThreads.RemoveAt(i--);
 384                         }
 385                 }
 386                 process.deleteThreadMutex.Signal();
 387                 PTimeInterval nextTimer = process.timers.Process();
 388     if (nextTimer != PMaxTimeInterval)
 389                 {
 390                         if ( nextTimer.GetInterval() > 10000 )
 391                         {
 392                                 nextTimer = 10000;
 393                         }
 394                 }
 395                 breakBlock.Wait( nextTimer );
 396         }
 397 }
 398
 399 void PProcess::SignalTimerChange()
 400 {
 401   if (houseKeeper == NULL)
 402      houseKeeper = new HouseKeepingThread;
 403   else
 404     houseKeeper->breakBlock.Signal();
 405 }
 406
 407
 408 ///////////////////////////////////////////////////////////////////////////////
 409 // PProcess
 410
 411 PProcess::~PProcess()
 412 {
 413   Sleep(100);  // Give threads time to die a natural death
 414
 415   delete houseKeeper;
 416
 417   // OK, if there are any left we get really insistent...
 418   activeThreadMutex.Wait();
 419   for (PINDEX i = 0; i < activeThreads.GetSize(); i++) {
 420     PThread & thread = activeThreads.GetDataAt(i);
 421     if (this != &thread && !thread.IsTerminated())
 422       thread.Terminate();  // With extreme prejudice
 423   }
 424   activeThreadMutex.Signal();
 425
 426   deleteThreadMutex.Wait();
 427   autoDeleteThreads.RemoveAll();
 428   deleteThreadMutex.Signal();
 429
 430   delete configFiles;
 431 }
 432
 433 ///////////////////////////////////////////////////////////////////////////////
 434 // PSemaphore
 435
 436 PSemaphore::PSemaphore(SEM_ID anId)
 437 {
 438   semId = anId;
 439   PAssertOS(semId != 0);
 440 }
 441
 442 PSemaphore::PSemaphore(unsigned initial, unsigned maxCount)
 443 {
 444   semId = semCCreate(SEM_Q_FIFO, initial);
 445   PAssertOS(semId != NULL);
 446 }
 447
 448 PSemaphore::~PSemaphore()
 449 {
 450         if ( semId != 0 )
 451         {
 452                 ::semDelete(semId);
 453         }
 454 }
 455
 456 void PSemaphore::Wait()
 457 {
 458         STATUS result = semTake(semId, WAIT_FOREVER);
 459         PAssertOS(result != ERROR);
 460 }
 461
 462
 463 BOOL PSemaphore::Wait(const PTimeInterval & timeout)
 464 {
 465   long wait;
 466         if (timeout == PMaxTimeInterval) {
 467                 wait = WAIT_FOREVER;
 468         }
 469         else {
 470          int ticks = sysClkRateGet();
 471          wait = (timeout.GetInterval() * ticks);
 472          wait = wait / 1000;
 473         }
 474
 475         STATUS result = semTake(semId, wait);
 476
 477   return result == ERROR;
 478 }
 479
 480 void PSemaphore::Signal()
 481 {
 482         semGive(semId);
 483 }
 484
 485 BOOL PSemaphore::WillBlock() const
 486 {
 487         STATUS result = semTake(semId, WAIT_FOREVER);
 488         PAssertOS(result != ERROR);
 489   return result == ERROR;
 490 }
 491
 492 ///////////////////////////////////////////////////////////////////////////////
 493 // PMutex
 494 PMutex::PMutex()
 495 : PSemaphore( ::semMCreate(SEM_Q_FIFO) )
 496 {
 497 }
 498
 499 PMutex::~PMutex()
 500 {
 501         PAssertOS(::semDelete(semId) == 0);
 502 }
 503
 504 void PMutex::Wait()
 505 {
 506   STATUS result = semTake(semId, WAIT_FOREVER); // wait forever
 507         PAssertOS(result == OK);
 508 }
 509
 510 BOOL PMutex::Wait(const PTimeInterval & timeout)
 511 {
 512   long wait;
 513         if (timeout == PMaxTimeInterval) {
 514                 wait = WAIT_FOREVER;
 515         }
 516         else {
 517          int ticks = sysClkRateGet();
 518          wait = (timeout.GetMilliSeconds() * ticks);
 519          wait = wait / 1000;
 520         }
 521         STATUS result = semTake(semId, wait);
 522   return result == ERROR;
 523 }
 524
 525 void PMutex::Signal()
 526 {
 527         ::semGive(semId);
 528 }
 529
 530 BOOL PMutex::WillBlock() const
 531 {
 532     STATUS result = semTake(semId, WAIT_FOREVER);       // wait forever
 533                 PAssertOS(result == OK);
 534                 return result == ERROR;
 535 }
 536
 537 ///////////////////////////////////////////////////////////////////////////////
 538 // PSyncPoint
 539
 540 PSyncPoint::PSyncPoint()
 541   : PSemaphore(0, 1)
 542 {
 543 }
 544
 545 // End Of File ///////////////////////////////////////////////////////////////