mythread_unix.cpp

linux下线程的创建、管理

#include <pthread.h>
#include <sys/time.h>
#include <signal.h>
#include "mythread.h"

#include "mymutex.h"
#include "mywaitcondition.h"
#include "mymutexpool_p.h"

#include <errno.h>


static MyMutexPool *my_thread_mutexpool = 0;

extern "C"
{
  static void *start_thread (void *_arg);
}
extern "C"
{
  static void finish_thread (void *arg);
}
extern "C"
{
  void suspend_signal_handler(int sig);
}
extern "C"
{
  void resume_signal_handler(int sig);
}
extern "C"
{
  void suspend_init_routine(void);
}

class MyThreadPrivate
{
public:
  MyWaitCondition thread_done;
  pthread_t thread_id;
  size_t stacksize;
  void *args[3];
  bool finished;
  bool running;
  bool orphan;
  bool suspended;

  static volatile sig_atomic_t sentinel;
  static pthread_once_t once; 

       MyThreadPrivate (size_t ss = 0);

  static inline void* start (MyThread * thread, void *para)
  {
    thread->run (para);
    return (void*)NULL;
  }

};
volatile sig_atomic_t MyThreadPrivate::sentinel=0;
pthread_once_t MyThreadPrivate::once=PTHREAD_ONCE_INIT;

extern "C"
{
  void suspend_signal_handler(int sig)
  {
    sigset_t signal_set;
    sigfillset(&signal_set);
    sigdelset(&signal_set,SIGUSR2);
    MyThreadPrivate::sentinel = 1;
    sigsuspend(&signal_set);

    return;
  }

  void resume_signal_handler(int sig)
  {
    MyThreadPrivate::sentinel = 2;
    return;
  }

  void suspend_init_routine(void)
  {
    int status;
    struct sigaction sigusr1,sigusr2;
    sigusr1.sa_flags=0;
    sigusr1.sa_handler=suspend_signal_handler;
    sigemptyset(&sigusr1.sa_mask);
    sigaddset(&sigusr1.sa_mask,SIGUSR2);

    sigusr2.sa_flags=0;
    sigusr2.sa_handler=resume_signal_handler;
    sigemptyset(&sigusr2.sa_mask);

    status=sigaction(SIGUSR1,&sigusr1,NULL);
    status=sigaction(SIGUSR2,&sigusr2,NULL);
  }
}


extern "C"
{
  static void *start_thread (void *_arg)
  {
    void **arg = (void **) _arg;

    pthread_cleanup_push (finish_thread, arg[1]);
    pthread_testcancel ();

    MyThreadPrivate::start ((MyThread *) arg[0], arg[2]);

    pthread_cleanup_pop (true);

    return 0;
  }

  static void finish_thread (void *arg)
  {
    MyThreadPrivate *d = (MyThreadPrivate *) arg;

    if (!d)
    {
      return;
    }

    MyMutexLocker locker (my_thread_mutexpool->get (d));

    d->running = false;
    d->finished = true;
    d->thread_id = 0;

    d->thread_done.wakeAll ();

    d->args[0] = d->args[1] = 0;

    if (d->orphan)
      delete d;
  }
}
inline MyThreadPrivate::MyThreadPrivate (size_t ss):
thread_id (0),
stacksize (ss), finished (false), running (false), orphan (false),suspended(false)
{
  if (!my_thread_mutexpool)
    MyThread::initialize ();
}



HANDLE MyThread::currentThread ()
{
  return (HANDLE) pthread_self ();
}

void MyThread::initialize ()
{
  if (!my_global_mutexpool)
    my_global_mutexpool = new MyMutexPool ();
  if (!my_thread_mutexpool)
    my_thread_mutexpool = new MyMutexPool ();
}

void MyThread::cleanup ()
{
  delete my_global_mutexpool;
  delete my_thread_mutexpool;

  my_global_mutexpool = 0;
  my_thread_mutexpool = 0;
}

static void thread_sleep (struct timespec *ti)
{
  pthread_mutex_t mtx = PTHREAD_MUTEX_INITIALIZER;
  pthread_cond_t cnd = PTHREAD_COND_INITIALIZER;

  pthread_mutex_lock (&mtx);
  (void) pthread_cond_timedwait (&cnd, &mtx, ti);
  pthread_mutex_unlock (&mtx);

  pthread_cond_destroy (&cnd);
  pthread_mutex_destroy (&mtx);
}

void MyThread::sleep (unsigned long secs)
{
  struct timeval tv;

  gettimeofday (&tv, 0);
  struct timespec ti;

  ti.tv_sec = tv.tv_sec + secs;
  ti.tv_nsec = (tv.tv_usec * 1000);
  thread_sleep (&ti);
}

void MyThread::msleep (unsigned long msecs)
{
  struct timeval tv;

  gettimeofday (&tv, 0);
  struct timespec ti;

  ti.tv_nsec = (tv.tv_usec * 1000) + (msecs % 1000) * 1000000;
  ti.tv_sec = tv.tv_sec + (msecs / 1000) + (ti.tv_nsec / 1000000000);
  ti.tv_nsec %= 1000000000;
  thread_sleep (&ti);
}

void MyThread::usleep (unsigned long usecs)
{
  struct timeval tv;

  gettimeofday (&tv, 0);
  struct timespec ti;

  ti.tv_nsec = (tv.tv_usec * 1000) + (usecs % 1000000) * 1000;
  ti.tv_sec = tv.tv_sec + (usecs / 1000000) + (ti.tv_nsec / 1000000000);
  ti.tv_nsec %= 1000000000;
  thread_sleep (&ti);
}

MyThread::MyThread ()
{
  d = new MyThreadPrivate;
}

MyThread::MyThread (unsigned int stackSize)
{
  d = new MyThreadPrivate (stackSize);
}

MyThread::~MyThread ()
{
  MyMutexLocker locker (my_thread_mutexpool->get (d));

  if (d->running && !d->finished)
  {
    d->orphan = true;
    return;
  }

  delete d;
}

void MyThread::exit ()
{
  pthread_exit (0);
}

void MyThread::start (void *para)
{
  MyMutexLocker locker (my_thread_mutexpool->get (d));

  if (d->running)
  {
    d->thread_done.wait (locker.mutex ());
  }

  d->running = true;
  d->finished = false;

  int ret;
  pthread_attr_t attr;

  pthread_attr_init (&attr);
  pthread_attr_setinheritsched (&attr, PTHREAD_INHERIT_SCHED);
  pthread_attr_setdetachstate (&attr, PTHREAD_CREATE_DETACHED);
  if (d->stacksize > 0)
  {
    ret = pthread_attr_setstacksize (&attr, d->stacksize);
    if (ret)
    {
      d->running = false;
      d->finished = false;
      return;
    }
  }
  d->args[0] = this;
  d->args[1] = d;
  d->args[2] = para;
  ret = pthread_create (&d->thread_id, &attr, start_thread, d->args);
  pthread_attr_destroy (&attr);

  if (ret)
  {
    d->running = false;
    d->finished = false;
  }
}

void MyThread::terminate ()
{
  MyMutexLocker private_locker (my_thread_mutexpool->get (d));

  if (d->finished || !d->running)
    return;
  if (!d->thread_id)
    return;
  pthread_cancel (d->thread_id);
}

bool MyThread::wait (unsigned long time)
{
  MyMutexLocker locker (my_thread_mutexpool->get (d));

  if (d->finished || !d->running)
    return true;
  return d->thread_done.wait (locker.mutex (), time);
}

bool MyThread::finished () const
{
  MyMutexLocker locker (my_thread_mutexpool->get (d));

  return d->finished;
}

bool MyThread::running () const
{
  MyMutexLocker locker (my_thread_mutexpool->get (d));

  return d->running;
}

int MyThread::suspend()
{
  int status;
  status=pthread_once(&MyThreadPrivate::once,suspend_init_routine);
  if(status!=0)
    return status;

  MyMutexLocker locker (my_thread_mutexpool->get (d));
  if(d->suspended==true)
  {
    return 0;
  }

  MyMutexLocker sentinel_locker (my_global_mutexpool->get ((void*)(&MyThreadPrivate::sentinel)));

  MyThreadPrivate::sentinel=0;
  status=pthread_kill(d->thread_id,SIGUSR1);
  while(MyThreadPrivate::sentinel!=1)
   sched_yield();

  d->suspended=true;

  return status;

}
int MyThread::resume()
{
  int status;
  status=pthread_once(&MyThreadPrivate::once,suspend_init_routine);
  if(status!=0)
    return status;

  MyMutexLocker locker (my_thread_mutexpool->get (d));
  if(d->suspended==false)
  {
    return 0;
  }

  MyMutexLocker sentinel_locker (my_global_mutexpool->get ((void*)(&MyThreadPrivate::sentinel)));

  MyThreadPrivate::sentinel=0;
  status=pthread_kill(d->thread_id,SIGUSR2);
  while(MyThreadPrivate::sentinel!=2)
    sched_yield();
  d->suspended=false;
  return status;

}