src/thread_pool_observer.cpp

   1
   2 /* $Id: thread_pool_observer.cpp,v 1.15 2006-03-29 13:44:45 adam Exp $
   3    Copyright (c) 2005-2006, 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 #if HAVE_SYS_SOCKET_H
  17 #include <sys/socket.h>
  18 #endif
  19
  20 #include <boost/thread/thread.hpp>
  21 #include <boost/thread/mutex.hpp>
  22 #include <boost/thread/condition.hpp>
  23
  24 #include <ctype.h>
  25 #include <stdio.h>
  26
  27 #include <deque>
  28
  29 #include <yazpp/socket-observer.h>
  30 #include <yaz/log.h>
  31
  32 #include "thread_pool_observer.hpp"
  33 #include "pipe.hpp"
  34
  35 namespace mp = metaproxy_1;
  36
  37 namespace metaproxy_1 {
  38     class ThreadPoolSocketObserver::Worker {
  39     public:
  40         Worker(ThreadPoolSocketObserver *s) : m_s(s) {};
  41         ThreadPoolSocketObserver *m_s;
  42         void operator() (void) {
  43             m_s->run(0);
  44         }
  45     };
  46
  47     class ThreadPoolSocketObserver::Rep : public boost::noncopyable {
  48         friend class ThreadPoolSocketObserver;
  49     public:
  50         Rep(yazpp_1::ISocketObservable *obs);
  51         ~Rep();
  52     private:
  53         yazpp_1::ISocketObservable *m_socketObservable;
  54         Pipe m_pipe;
  55         boost::thread_group m_thrds;
  56         boost::mutex m_mutex_input_data;
  57         boost::condition m_cond_input_data;
  58         boost::mutex m_mutex_output_data;
  59         std::deque<IThreadPoolMsg *> m_input;
  60         std::deque<IThreadPoolMsg *> m_output;
  61         bool m_stop_flag;
  62         int m_no_threads;
  63     };
  64 }
  65
  66
  67 using namespace yazpp_1;
  68 using namespace mp;
  69
  70 ThreadPoolSocketObserver::Rep::Rep(ISocketObservable *obs)
  71     : m_socketObservable(obs), m_pipe(9123)
  72 {
  73 }
  74
  75 ThreadPoolSocketObserver::Rep::~Rep()
  76 {
  77 }
  78
  79 IThreadPoolMsg::~IThreadPoolMsg()
  80 {
  81
  82 }
  83
  84 ThreadPoolSocketObserver::ThreadPoolSocketObserver(ISocketObservable *obs,
  85                                                    int no_threads)
  86     : m_p(new Rep(obs))
  87 {
  88     obs->addObserver(m_p->m_pipe.read_fd(), this);
  89     obs->maskObserver(this, SOCKET_OBSERVE_READ);
  90
  91     m_p->m_stop_flag = false;
  92     m_p->m_no_threads = no_threads;
  93     int i;
  94     for (i = 0; i<no_threads; i++)
  95     {
  96         Worker w(this);
  97         m_p->m_thrds.add_thread(new boost::thread(w));
  98     }
  99 }
 100
 101 ThreadPoolSocketObserver::~ThreadPoolSocketObserver()
 102 {
 103     {
 104         boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
 105         m_p->m_stop_flag = true;
 106         m_p->m_cond_input_data.notify_all();
 107     }
 108     m_p->m_thrds.join_all();
 109
 110     m_p->m_socketObservable->deleteObserver(this);
 111 }
 112
 113 void ThreadPoolSocketObserver::socketNotify(int event)
 114 {
 115     if (event & SOCKET_OBSERVE_READ)
 116     {
 117         char buf[2];
 118         recv(m_p->m_pipe.read_fd(), buf, 1, 0);
 119         IThreadPoolMsg *out;
 120         {
 121             boost::mutex::scoped_lock output_lock(m_p->m_mutex_output_data);
 122             out = m_p->m_output.front();
 123             m_p->m_output.pop_front();
 124         }
 125         if (out)
 126             out->result();
 127     }
 128 }
 129
 130 void ThreadPoolSocketObserver::run(void *p)
 131 {
 132     while(1)
 133     {
 134         IThreadPoolMsg *in = 0;
 135         {
 136             boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
 137             while (!m_p->m_stop_flag && m_p->m_input.size() == 0)
 138                 m_p->m_cond_input_data.wait(input_lock);
 139             if (m_p->m_stop_flag)
 140                 break;
 141
 142             in = m_p->m_input.front();
 143             m_p->m_input.pop_front();
 144         }
 145         IThreadPoolMsg *out = in->handle();
 146         {
 147             boost::mutex::scoped_lock output_lock(m_p->m_mutex_output_data);
 148             m_p->m_output.push_back(out);
 149             send(m_p->m_pipe.write_fd(), "", 1, 0);
 150         }
 151     }
 152 }
 153
 154 void ThreadPoolSocketObserver::put(IThreadPoolMsg *m)
 155 {
 156     boost::mutex::scoped_lock input_lock(m_p->m_mutex_input_data);
 157     m_p->m_input.push_back(m);
 158     m_p->m_cond_input_data.notify_one();
 159 }
 160 /*
 161  * Local variables:
 162  * c-basic-offset: 4
 163  * indent-tabs-mode: nil
 164  * c-file-style: "stroustrup"
 165  * End:
 166  * vim: shiftwidth=4 tabstop=8 expandtab
 167  */
 168