src/tstthreads.cpp

   1 /* $Id: tstthreads.cpp,v 1.2 2005-05-20 21:32:31 adam Exp $
   2    Copyright (c) 1998-2005, Index Data.
   3
   4 This file is part of the yaz-proxy.
   5
   6 YAZ proxy is free software; you can redistribute it and/or modify it under
   7 the terms of the GNU General Public License as published by the Free
   8 Software Foundation; either version 2, or (at your option) any later
   9 version.
  10
  11 YAZ proxy is distributed in the hope that it will be useful, but WITHOUT ANY
  12 WARRANTY; without even the implied warranty of MERCHANTABILITY or
  13 FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
  14 for more details.
  15
  16 You should have received a copy of the GNU General Public License
  17 along with YAZ proxy; see the file LICENSE.  If not, write to the
  18 Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA
  19 02111-1307, USA.
  20  */
  21
  22 #include <stdlib.h>
  23 #include <pthread.h>
  24 #include <unistd.h>
  25 #include <ctype.h>
  26
  27 #if HAVE_DLFCN_H
  28 #include <dlfcn.h>
  29 #endif
  30
  31 #include <yaz++/socket-observer.h>
  32 #include <yaz++/socket-manager.h>
  33 #include <yaz/log.h>
  34 #include "proxyp.h"
  35
  36 class Proxy_Msg {
  37 public:
  38     virtual void destroy() = 0;
  39     virtual Proxy_Msg *handle() = 0;
  40     virtual void result() = 0;
  41 };
  42
  43 class Proxy_Msg_Queue_List {
  44     friend class Proxy_Msg_Queue;
  45  private:
  46     Proxy_Msg *m_item;
  47     Proxy_Msg_Queue_List *m_next;
  48 };
  49
  50 class Proxy_Msg_Queue {
  51  public:
  52     Proxy_Msg_Queue();
  53     void enqueue(Proxy_Msg *in);
  54     Proxy_Msg *dequeue();
  55     int size();
  56  private:
  57     Proxy_Msg_Queue_List *m_list;
  58 };
  59
  60 Proxy_Msg_Queue::Proxy_Msg_Queue()
  61 {
  62     m_list = 0;
  63 }
  64
  65 int Proxy_Msg_Queue::size()
  66 {
  67     int no = 0;
  68     Proxy_Msg_Queue_List *l;
  69     for (l = m_list; l; l = l->m_next)
  70         no++;
  71     return no;
  72 }
  73
  74 void Proxy_Msg_Queue::enqueue(Proxy_Msg *m)
  75 {
  76     Proxy_Msg_Queue_List *l = new Proxy_Msg_Queue_List;
  77     l->m_next = m_list;
  78     l->m_item = m;
  79     m_list = l;
  80 }
  81
  82 Proxy_Msg *Proxy_Msg_Queue::dequeue()
  83 {
  84     Proxy_Msg_Queue_List **l = &m_list;
  85     if (!*l)
  86         return 0;
  87     while ((*l)->m_next)
  88         l = &(*l)->m_next;
  89     Proxy_Msg *m = (*l)->m_item;
  90     delete *l;
  91     *l = 0;
  92     return m;
  93 }
  94
  95 class Proxy_Thread : public IYazSocketObserver {
  96 public:
  97     Proxy_Thread(IYazSocketObservable *obs);
  98     virtual ~Proxy_Thread();
  99     void socketNotify(int event);
 100     void put(Proxy_Msg *m);
 101     Proxy_Msg *get();
 102     void run(void *p);
 103 private:
 104     IYazSocketObservable *m_obs;
 105     int m_fd[2];
 106     pthread_t m_thread_id;
 107     Proxy_Msg_Queue m_input;
 108     Proxy_Msg_Queue m_output;
 109     pthread_mutex_t m_mutex_input_data;
 110     pthread_cond_t m_cond_input_data;
 111     pthread_mutex_t m_mutex_output_data;
 112     pthread_cond_t m_cond_output_data;
 113 };
 114
 115 static void *tfunc(void *p)
 116 {
 117     Proxy_Thread *pt = (Proxy_Thread *) p;
 118     pt->run(0);
 119     return 0;
 120 }
 121
 122
 123 Proxy_Thread::Proxy_Thread(IYazSocketObservable *obs)
 124     : m_obs(obs)
 125 {
 126     pthread_mutex_init(&m_mutex_input_data, 0);
 127     pthread_cond_init(&m_cond_input_data, 0);
 128     pthread_mutex_init(&m_mutex_output_data, 0);
 129     pthread_cond_init(&m_cond_output_data, 0);
 130     m_fd[0] = m_fd[1] = -1;
 131     if (pipe(m_fd) != 0)
 132         return;
 133     m_obs->addObserver(m_fd[0], this);
 134     m_obs->timeoutObserver(this, 2000);
 135     m_obs->maskObserver(this, YAZ_SOCKET_OBSERVE_READ);
 136
 137     pthread_create(&m_thread_id, 0, tfunc, this);
 138 }
 139
 140 Proxy_Thread::~Proxy_Thread()
 141 {
 142
 143 }
 144
 145 void Proxy_Thread::socketNotify(int event)
 146 {
 147     char buf[2];
 148     read(m_fd[0], buf, 1);
 149     pthread_mutex_lock(&m_mutex_output_data);
 150     Proxy_Msg *out = m_output.dequeue();
 151     pthread_mutex_unlock(&m_mutex_output_data);
 152     if (out)
 153         out->result();
 154 }
 155
 156 void Proxy_Thread::run(void *p)
 157 {
 158     while(1)
 159     {
 160         pthread_mutex_lock(&m_mutex_input_data);
 161         pthread_cond_wait(&m_cond_input_data, &m_mutex_input_data);
 162         while(1)
 163         {
 164             Proxy_Msg *in = m_input.dequeue();
 165             pthread_mutex_unlock(&m_mutex_input_data);
 166             if (!in)
 167                 break;
 168             Proxy_Msg *out = in->handle();
 169             pthread_mutex_lock(&m_mutex_output_data);
 170             m_output.enqueue(out);
 171             pthread_cond_signal(&m_cond_output_data);
 172             pthread_mutex_unlock(&m_mutex_output_data);
 173             write(m_fd[1], "", 1);
 174
 175             pthread_mutex_lock(&m_mutex_input_data);
 176         }
 177     }
 178 }
 179
 180 void Proxy_Thread::put(Proxy_Msg *m)
 181 {
 182     pthread_mutex_lock(&m_mutex_input_data);
 183     m_input.enqueue(m);
 184     pthread_cond_signal(&m_cond_input_data);
 185     int in_size = m_input.size();
 186     pthread_mutex_unlock(&m_mutex_input_data);
 187     int out_size = m_output.size();
 188     printf("in-size=%d out-size=%d\n", in_size, out_size);
 189 }
 190
 191 class My_Msg : public Proxy_Msg {
 192 public:
 193     void destroy();
 194     Proxy_Msg *handle();
 195     void result();
 196     int m_val;
 197 };
 198
 199 class My_Thread : public Proxy_Thread {
 200 public:
 201     My_Thread(IYazSocketObservable *obs);
 202 };
 203
 204 My_Thread::My_Thread(IYazSocketObservable *obs) : Proxy_Thread(obs)
 205 {
 206 }
 207
 208 Proxy_Msg *My_Msg::handle()
 209 {
 210     My_Msg *res = new My_Msg;
 211     int sl = rand() % 5;
 212
 213     res->m_val = m_val;
 214     printf("My_Msg::handle val=%d sleep=%d\n", m_val, sl);
 215     sleep(sl);
 216     return res;
 217 }
 218
 219
 220 void My_Msg::result()
 221 {
 222     printf("My_Msg::result val=%d\n", m_val);
 223 }
 224
 225 void My_Msg::destroy()
 226 {
 227     delete this;
 228 }
 229
 230 class My_Timer_Thread : public IYazSocketObserver {
 231 private:
 232     IYazSocketObservable *m_obs;
 233     int m_fd[2];
 234     My_Thread *m_t;
 235 public:
 236     My_Timer_Thread(IYazSocketObservable *obs, My_Thread *t);
 237     void socketNotify(int event);
 238 };
 239
 240 My_Timer_Thread::My_Timer_Thread(IYazSocketObservable *obs,
 241                                  My_Thread *t) : m_obs(obs)
 242 {
 243     pipe(m_fd);
 244     m_t = t;
 245     obs->addObserver(m_fd[0], this);
 246     obs->timeoutObserver(this, 2);
 247 }
 248
 249 void My_Timer_Thread::socketNotify(int event)
 250 {
 251     static int seq = 1;
 252     printf("Add %d\n", seq);
 253     My_Msg *m = new My_Msg;
 254     m->m_val = seq++;
 255     m_t->put(m);
 256 }
 257
 258 int main(int argc, char **argv)
 259 {
 260     Yaz_SocketManager mySocketManager;
 261
 262     My_Thread m(&mySocketManager);
 263     My_Timer_Thread t(&mySocketManager, &m);
 264     while (mySocketManager.processEvent() > 0)
 265         ;
 266     return 0;
 267 }