"Fossies" - the Fresh Open Source Software Archive

Member "postal-0.76/thread.cpp" (10 Apr 2008, 3662 Bytes) of package /linux/privat/postal-0.76.tgz:

    1 #include <stdlib.h>
    2 #include "thread.h"
    3 #include <stdio.h>
    4 #include <errno.h>
    5 
    6 #include <unistd.h>
    7 #include <time.h>
    8 #include <sys/types.h>
    9 #include <sys/wait.h>
   10 #include <pthread.h>
   11 
   12 Thread::Thread()
   13  : m_read(-1)
   14  , m_write(-1)
   15  , m_threadNum(-1)
   16 
   17  , m_parentRead(-1)
   18  , m_parentWrite(-1)
   19  , m_childRead(-1)
   20  , m_childWrite(-1)
   21  , m_numThreads(0)
   22  , m_retVal(NULL)
   23  , m_thread_info(NULL)
   24 {
   25 }
   26 
   27 Thread::Thread(int threadNum, const Thread *parent)
   28  : m_read(parent->m_childRead)
   29  , m_write(parent->m_childWrite)
   30  , m_threadNum(threadNum)
   31 
   32  , m_parentRead(-1)
   33  , m_parentWrite(-1)
   34  , m_childRead(-1)
   35  , m_childWrite(-1)
   36  , m_numThreads(parent->m_numThreads)
   37  , m_retVal(&parent->m_retVal[threadNum])
   38  , m_thread_info(NULL)
   39 {
   40 }
   41 
   42 Thread::~Thread()
   43 {
   44   if(m_threadNum == -1)
   45   {
   46     for(int i = 0; i < m_numThreads; i++)
   47     {
   48       pthread_join(m_thread_info[i], NULL);
   49     }
   50     close(m_parentRead);
   51     close(m_parentWrite);
   52     close(m_childRead);
   53     close(m_childWrite);
   54     delete m_thread_info;
   55     delete m_retVal;
   56   }
   57 }
   58 
   59 // for the benefit of this function and the new Thread class it may create
   60 // the Thread class must do nothing of note in it's constructor or it's
   61 // go() member function.
   62 PVOID thread_func(PVOID param)
   63 {
   64   THREAD_DATA *td = (THREAD_DATA *)param;
   65   Thread *thread = td->f->newThread(td->threadNum);
   66   thread->setRetVal(thread->action(td->param));
   67   delete thread;
   68   delete td;
   69   return NULL;
   70 }
   71 
   72 void Thread::go(PVOID param, int num)
   73 {
   74   m_numThreads += num;
   75   int control[2];
   76   int feedback[2];
   77   if (pipe(feedback) || pipe(control))
   78   {
   79     fprintf(stderr, "Can't open pipes.\n");
   80     exit(1);
   81   }
   82   m_parentRead = feedback[0];
   83   m_parentWrite = control[1];
   84   m_childRead = control[0];
   85   m_childWrite = feedback[1];
   86   m_read = m_parentRead;
   87   m_write = m_parentWrite;
   88   m_readPoll.events = POLLIN | POLLERR | POLLHUP | POLLNVAL;
   89   m_writePoll.events = POLLOUT | POLLERR | POLLHUP | POLLNVAL;
   90   m_readPoll.fd = m_parentRead;
   91   m_writePoll.fd = m_parentWrite;
   92   pthread_attr_t attr;
   93   if(pthread_attr_init(&attr)
   94    || pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE)
   95    || pthread_attr_setstacksize(&attr, 32*1024))
   96     fprintf(stderr, "Can't set thread attributes.\n");
   97 
   98   m_retVal = new int[num + 1];
   99   m_thread_info = new pthread_t[num];
  100   for(int i = 1; i <= num; i++)
  101   {
  102     m_retVal[i] = -1;
  103     THREAD_DATA *td = new THREAD_DATA;
  104     td->f = this;
  105     td->param = param;
  106     td->threadNum = i;
  107     int p = pthread_create(&m_thread_info[i - 1], &attr, thread_func, PVOID(td));
  108     if(p)
  109     {
  110       fprintf(stderr, "Can't create a thread.\n");
  111       exit(1);
  112     }
  113   }
  114   if(pthread_attr_destroy(&attr))
  115     fprintf(stderr, "Can't destroy thread attributes.\n");
  116   m_readPoll.fd = m_read;
  117   m_writePoll.fd = m_write;
  118 }
  119 
  120 void Thread::setRetVal(int rc)
  121 {
  122   *m_retVal = rc;
  123 }
  124 
  125 int Thread::Read(PVOID buf, int size, int timeout)
  126 {
  127   if(timeout)
  128   {
  129     int rc = poll(&m_readPoll, 1, timeout * 1000);
  130     if(rc < 0)
  131     {
  132       fprintf(stderr, "Can't poll read ITC.\n");
  133       return -1;
  134     }
  135     if(!rc)
  136       return 0;
  137   }
  138   if(size != read(m_read, buf, size) )
  139   {
  140     fprintf(stderr, "Can't read data from ITC pipe.\n");
  141     return -1;
  142   }
  143   return size;
  144 }
  145 
  146 int Thread::Write(PVOID buf, int size, int timeout)
  147 {
  148   if(timeout)
  149   {
  150     int rc;
  151     do
  152     {
  153       rc = poll(&m_writePoll, 1, timeout * 1000);
  154       if(rc < 0 && errno != EINTR)
  155       {
  156         fprintf(stderr, "Can't poll write ITC.\n");
  157         return -1;
  158       }
  159     } while(rc == -1);
  160     if(!rc)
  161       return 0;
  162   }
  163   if(size != write(m_write, buf, size))
  164   {
  165     fprintf(stderr, "Can't write data to ITC pipe.\n");
  166     return -1;
  167   }
  168   return size;
  169 }
  170