pthread.c

一个开源的网络开发库ACE

  intkey = intLock();

  if (keyList[key].valid)
    {
      intUnlock(intkey);
      return TRUE;
    }
  else
    {
      intUnlock(intkey);
      return FALSE;
    }
}
#endif /* PACE_HAS_NONUOF_FUNCS */


/*
 * PACE - POSIX Functions
 */


#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_cond_timedwait (pace_pthread_cond_t * cond,
                        pace_pthread_mutex_t * mutex,
                        const pace_timespec * abstime)
{
  /*
  int msec_timeout;
  if (abstime->sec () == 0 && abstime->usec () == 0)
    msec_timeout = 0; * Do a "poll." *
  else
    {
      // Assumes that struct timespec is same size as struct timeval,
      // which assumes that time_t is a long: it currently is.
      struct pace_timespec ts;

      pace_clock_gettime (CLOCK_REALTIME, &ts);

      // Note that we must convert between absolute time (which is
      // passed as a parameter) and relative time.
      pace_timespec relative_time = (*abstime) - ts;

      // Watchout for situations where a context switch has caused the
      // current time to be > the timeout.
      if (relative_time < 0)
        msec_timeout = 0;
      else
        msec_timeout = relative_time.msec ();
    }
  // Inline the call to ACE_OS::sema_wait () because it takes an
  // ACE_Time_Value argument.  Avoid the cost of that conversion . . .
  int ticks_per_sec = ::sysClkRateGet ();
  int ticks = msec_timeout * ticks_per_sec / ACE_ONE_SECOND_IN_MSECS;
  result = ::semTake (cv->sema_.sema_, ticks);
  */





  int timeval = 0;
  int errornumber;
  int returnval = 0;
  STATUS status;

  PACE_TRACE("pthread_cond_timedwait");

  if (pace_pthread_mutex_unlock(mutex) != OK) return ERROR;

  /* convert the abstime to timeval */
  status = semTake(*cond, timeval);

  if (status != OK)
    {
      errornumber = errnoGet();
      if (errornumber == S_objLib_OBJ_ID_ERROR)
        returnval = EINVAL;
      else if (errornumber == S_objLib_OBJ_TIMEOUT) 
        returnval = ETIMEDOUT;
      else 
        returnval = ERROR;
    }

  pace_pthread_mutex_lock(mutex);

  return returnval;
}
#endif /* PACE_HAS_NONUOF_FUNCS */


#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_cond_wait (pace_pthread_cond_t * cond,
                   pace_pthread_mutex_t * mutex)
{
  STATUS status;
  int errornumber;
  int returnval = 0;

  PACE_TRACE("pthread_cond_wait");

  if (pace_pthread_mutex_unlock(mutex) != OK)
    return ERROR;

  status = semTake(*cond, WAIT_FOREVER);

  if(status != OK)
    {
      errornumber = errnoGet();
      if (errornumber == S_objLib_OBJ_ID_ERROR) 
          returnval = EINVAL;
      else
          returnval = ERROR;
    }

  pace_pthread_mutex_lock(mutex);

  return returnval;
}
#endif /* PACE_HAS_NONUOF_FUNCS */

#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_create (pace_pthread_t * thread,
                const pace_pthread_attr_t * attr,
# if defined (PACE_HAS_CPLUSPLUS)
                pace_create_pf start_routine,
# else
                void * (*start_routine) (void*),
# endif
                void * arg)
{
  pace_pthread_attr_t pattr;
  char * pname;
  int taskid;
  pace_pthread_t pthread;
  WIND_TCB * pTcb;

  PACE_TRACE("pthread_create");

  if (attr == 0) 
    pattr = pthread_attr_default;
  else
    if (*attr == 0)
       pattr = pthread_attr_default;
    else
       pattr = *attr;

  if (pattr->name[0] != '\0')   /* name is provided */
    pname = pattr->name;
  else
    pname = (char *)0;

  taskid = taskSpawn(pname, 
                     (SCHED_FIFO_HIGH_PRI - pattr->schedule.sched_priority), 
                     VX_FP_TASK, pattr->stacksize, (FUNCPTR)start_routine, 
                     (int)arg, 0,0,0,0,0,0,0,0,0);

  if (taskid == ERROR)
     return ERROR;

  if ((pTcb = taskTcb(taskid)) == NULL)
    {
      taskDelete(taskid);
      return ERROR;
    }

  pthread = (pace_pthread_t) malloc(sizeof(struct _PTHREAD_T));

  if (pthread == NULL)
    {
      taskDelete(taskid);
      return ERROR;
    }

  /* construct pace_pthread_t structure */

  bzero((char *)pthread, sizeof(struct _PTHREAD_T));
  pthread->tid = taskid;
  pthread->stateflag = PTHREAD_CANCEL_ENABLE;
  pthread->canceltype = PTHREAD_CANCEL_ASYNCHRONOUS;
  pthread->detachflag = pattr->dstate;

  /* initialize the join semaphore also */
  if ((pthread->joinSem = semBCreate(SEM_Q_PRIORITY, SEM_EMPTY)) == NULL)
    {
      free((void *)pthread);
      taskDelete(taskid);
      return ERROR;
    }

  /* pass it to the caller */
  *thread = pthread;   

  /* save to the WIND_TCB for reference afterward */
  pTcb->_USER_SPARE4 = (int) pthread;

  if (registered_cleanup_init == 0)
    {
      taskDeleteHookAdd((FUNCPTR)pacevx_pthread_run_cleanup);
      registered_cleanup_init = 1;
    }

  pacevx_pthread_queue_add(pthread);

  return OK;
}
#endif /* PACE_HAS_NONUOF_FUNCS */

#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_detach (pace_pthread_t thread)
{
  int key;
  int needfree;
  needfree = 0;

  PACE_TRACE("pthread_detach");

  if (!pacevx_pthread_verify(thread))
    return EINVAL;

  key = intLock();

  switch (thread->joinstate)
    {
      /* task is joined, or detached, but still running, do nothing */
      case JOIN_PENDING:
      case JOIN_DETATCHED:
        intUnlock(key);
        return OK;
        break;
      case JOIN_NORMAL:      /* task is running */
        thread->joinstate = JOIN_DETATCHED;
        break;
      case JOIN_TERMINATED:
        needfree = 1;
        break;
      default:
        break;
    }

  intUnlock(key);

  if (needfree) 
    {
      pacevx_pthread_queue_remove(thread);
      free(thread);
    }
  return OK;
}
#endif /* PACE_HAS_NONUOF_FUNCS */

#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_join (pace_pthread_t thread, void ** value_ptr)
{
  /*
   * The pthread_join() function suspends execution of the calling
   * thread until the target thread terminates, unless the target 
   * thread has already terminated.
   * The terminating thread can pass value to the caller by
   * pthread_exit() and the calling thread gets it from "value_ptr"
   * The application must verify the "value_ptr" value before using it.
   */
  pace_pthread_t pthread;
  int needfree;
  int key;

  PACE_TRACE("pthread_join");

  if (!pacevx_pthread_verify(thread))
    return ERROR;

  if (thread->detachflag != PTHREAD_CREATE_JOINABLE)
     return ERROR;

  if ((pthread = pace_pthread_self()) == NULL)
      return ERROR;

  needfree = 0;

  key = intLock();
  switch (thread->joinstate)
    {
      case JOIN_NORMAL:
        thread->jointhread = pthread;
        thread->joinstate = JOIN_PENDING;
        break;
      case JOIN_TERMINATED:
        needfree = 1;
        break;
      case JOIN_PENDING:
      default:
        intUnlock(key);
        return ERROR;
    }
  intUnlock(key);

  if (needfree)
  {
     *value_ptr = thread->joinvalue;
     pacevx_pthread_queue_remove(thread);
     free(thread);
     return OK;
  }

  /* if we are here, thread is not terminated yet */
  semTake(pthread->joinSem, WAIT_FOREVER);

  /* cleanup */
  *value_ptr = thread->joinvalue;
  pacevx_pthread_queue_remove(thread);
  free(thread);
  return OK;
}
#endif /* PACE_HAS_NONUOF_FUNCS */

#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_key_create (pace_pthread_key_t * key,
# if defined (PACE_HAS_CPLUSPLUS)
                    pace_keycreate_pf destructor)
# else
                    void (*destructor)(void*))
# endif
{
  /*
   * Create a thread-specific data key. Also initialize the 
   * data structure if it is called first time.
   */
  int i;
  int intkey;

  PACE_TRACE("pthread_key_create");

  /* do the initialization if it is the first time */
  if (initialized == 0)
    {
      initialized = 1;

      /* initialize the data structure */
      for (i = 0; i < PTHREAD_KEYS_MAX; i++)
        {
          keyList[i].index = i;
          keyList[i].valid = FALSE;
          keyList[i].destructor = NULL;
        } 
    }

  /* find first available position */
  intkey = intLock();
  for (i = 0; i < PTHREAD_KEYS_MAX; i++)
    {
      if (keyList[i].valid == FALSE)
        {
          *key = (pace_pthread_key_t)keyList[i].index;
          keyList[i].valid = TRUE;
          keyList[i].destructor = destructor;
          intUnlock(intkey);

          return OK;
        }
    }

  intUnlock(intkey);
  return ERROR;
}
#endif /* PACE_HAS_NONUOF_FUNCS */

#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_key_delete (pace_pthread_key_t key)
{
  int intkey;

  PACE_TRACE("pthread_key_delete");

  if ((key < 0) || (key >= PTHREAD_KEYS_MAX))
    return ERROR;

  intkey = intLock();

  keyList[key].valid = FALSE;
  keyList[key].destructor = NULL;

  intUnlock(intkey);

  return OK;
}
#endif /* PACE_HAS_NONUOF_FUNCS */

#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_once (pace_pthread_once_t * once_control,
# if defined (PACE_HAS_CPLUSPLUS)
              pace_once_pf init_routine)
# else
              void (*init_routine) (void))
# endif 
{
  /*
   * Once function allows the function to be executed exact only once
   * Subsequent calls of pthread_once() with the same once_control will 
   * not call the void_routine(). 
   */
  int i;
  int key;
  pace_pthread_t pthread;

  PACE_TRACE("pthread_once");

  if ((pthread = pace_pthread_self()) == NULL)
    return ERROR;

  /* make it atomic */
  key = intLock();

  for (i = 0; i < pthread->onceCount; i++)
    {
      if (*once_control == pthread->onceList[i].once_ctl)
        {
          /* do nothing, already called */
          intUnlock(key);
          return OK;
        }
    }

  /* if we are here, no match is found */
  pthread->onceList[pthread->onceCount].once_ctl = *once_control;
  pthread->onceCount++;
  intUnlock(key);

  /* run the init routine */
  (*init_routine)();
  return OK;
}
#endif /* PACE_HAS_NONUOF_FUNCS */


#if (PACE_HAS_POSIX_NONUOF_FUNCS)
int
pthread_attr_init (pace_pthread_attr_t * attr)
{
  PACE_TRACE("pthread_attr_init");

  /*
   * Attempt to allocate memory for the attributes object.
   */

  if ((*attr = (pace_pthread_attr_t) malloc(sizeof(struct _PTHREAD_ATTR)))
      == NULL)
    {
      return ERROR;
    }

  /*
   * Set the default attributes.