src/thread_pool_observer.cpp

   1
   2 /* $Id: thread_pool_observer.cpp,v 1.8 2005-11-04 10:27:51 adam Exp $
   3    Copyright (c) 2005, Index Data.
   4
   5 %LICENSE%
   6  */
   7 #include "config.hpp"
   8
   9 #if HAVE_UNISTD_H
  10 #include <unistd.h>
  11 #endif
  12 #ifdef WIN32
  13 #include <winsock.h>
  14 #endif
  15
  16 #include <ctype.h>
  17 #include <stdio.h>
  18
  19 #include <yaz++/socket-observer.h>
  20 #include <yaz/log.h>
  21
  22 #include "thread_pool_observer.hpp"
  23
  24 using namespace yazpp_1;
  25 using namespace yp2;
  26
  27 IThreadPoolMsg::~IThreadPoolMsg()
  28 {
  29
  30 }
  31
  32 ThreadPoolSocketObserver::ThreadPoolSocketObserver(ISocketObservable *obs, int no_threads)
  33     : m_SocketObservable(obs)
  34 {
  35     pipe(m_fd);
  36     obs->addObserver(m_fd[0], this);
  37     obs->maskObserver(this, SOCKET_OBSERVE_READ);
  38
  39     m_stop_flag = false;
  40     m_no_threads = no_threads;
  41     int i;
  42     for (i = 0; i<no_threads; i++)
  43     {
  44         Worker w(this);
  45         m_thrds.add_thread(new boost::thread(w));
  46     }
  47 }
  48
  49 ThreadPoolSocketObserver::~ThreadPoolSocketObserver()
  50 {
  51     {
  52         boost::mutex::scoped_lock input_lock(m_mutex_input_data);
  53         m_stop_flag = true;
  54         m_cond_input_data.notify_all();
  55     }
  56     m_thrds.join_all();
  57
  58     m_SocketObservable->deleteObserver(this);
  59
  60     close(m_fd[0]);
  61     close(m_fd[1]);
  62 }
  63
  64 void ThreadPoolSocketObserver::socketNotify(int event)
  65 {
  66     if (event & SOCKET_OBSERVE_READ)
  67     {
  68         char buf[2];
  69         read(m_fd[0], buf, 1);
  70         IThreadPoolMsg *out;
  71         {
  72             boost::mutex::scoped_lock output_lock(m_mutex_output_data);
  73             out = m_output.front();
  74             m_output.pop_front();
  75         }
  76         if (out)
  77             out->result();
  78     }
  79 }
  80
  81 void ThreadPoolSocketObserver::run(void *p)
  82 {
  83     while(1)
  84     {
  85         IThreadPoolMsg *in = 0;
  86         {
  87             boost::mutex::scoped_lock input_lock(m_mutex_input_data);
  88             while (!m_stop_flag && m_input.size() == 0)
  89                 m_cond_input_data.wait(input_lock);
  90             if (m_stop_flag)
  91                 break;
  92
  93             in = m_input.front();
  94             m_input.pop_front();
  95         }
  96         IThreadPoolMsg *out = in->handle();
  97         {
  98             boost::mutex::scoped_lock output_lock(m_mutex_output_data);
  99             m_output.push_back(out);
 100             write(m_fd[1], "", 1);
 101         }
 102     }
 103 }
 104
 105 void ThreadPoolSocketObserver::put(IThreadPoolMsg *m)
 106 {
 107     boost::mutex::scoped_lock input_lock(m_mutex_input_data);
 108     m_input.push_back(m);
 109     m_cond_input_data.notify_one();
 110 }
 111 /*
 112  * Local variables:
 113  * c-basic-offset: 4
 114  * indent-tabs-mode: nil
 115  * c-file-style: "stroustrup"
 116  * End:
 117  * vim: shiftwidth=4 tabstop=8 expandtab
 118  */
 119