src/msg-thread.cpp

   1 /* $Id: msg-thread.cpp,v 1.14 2006-03-30 14:16:34 adam Exp $
   2    Copyright (c) 1998-2006, Index Data.
   3
   4 This file is part of the yazproxy.
   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 #if YAZ_POSIX_THREADS
  23 #include <pthread.h>
  24 #endif
  25
  26 #if HAVE_UNISTD_H
  27 #include <unistd.h>
  28 #endif
  29
  30 #include <ctype.h>
  31 #include <stdio.h>
  32
  33 #include <yazpp/socket-observer.h>
  34 #include <yaz/log.h>
  35
  36 #include "msg-thread.h"
  37
  38 using namespace yazpp_1;
  39
  40 class Msg_Thread::Private {
  41 public:
  42     int m_no_threads;
  43     Msg_Thread_Queue m_input;
  44     Msg_Thread_Queue m_output;
  45 #if YAZ_POSIX_THREADS
  46     int m_fd[2];
  47     yazpp_1::ISocketObservable *m_SocketObservable;
  48     pthread_t *m_thread_id;
  49     pthread_mutex_t m_mutex_input_data;
  50     pthread_cond_t m_cond_input_data;
  51     pthread_mutex_t m_mutex_output_data;
  52     bool m_stop_flag;
  53 #endif
  54 };
  55
  56 IMsg_Thread::~IMsg_Thread()
  57 {
  58
  59 }
  60
  61 Msg_Thread_Queue::Msg_Thread_Queue()
  62 {
  63     m_list = 0;
  64 }
  65
  66 int Msg_Thread_Queue::size()
  67 {
  68     int no = 0;
  69     Msg_Thread_Queue_List *l;
  70     for (l = m_list; l; l = l->m_next)
  71         no++;
  72     return no;
  73 }
  74
  75 void Msg_Thread_Queue::enqueue(IMsg_Thread *m)
  76 {
  77     Msg_Thread_Queue_List *l = new Msg_Thread_Queue_List;
  78     l->m_next = m_list;
  79     l->m_item = m;
  80     m_list = l;
  81 }
  82
  83 IMsg_Thread *Msg_Thread_Queue::dequeue()
  84 {
  85     Msg_Thread_Queue_List **l = &m_list;
  86     if (!*l)
  87         return 0;
  88     while ((*l)->m_next)
  89         l = &(*l)->m_next;
  90     IMsg_Thread *m = (*l)->m_item;
  91     delete *l;
  92     *l = 0;
  93     return m;
  94 }
  95
  96 #if YAZ_POSIX_THREADS
  97 static void *tfunc(void *p)
  98 {
  99     Msg_Thread *pt = (Msg_Thread *) p;
 100     pt->run(0);
 101     return 0;
 102 }
 103 #endif
 104
 105 Msg_Thread::Msg_Thread(ISocketObservable *obs, int no_threads)
 106 {
 107     m_p = new Private;
 108
 109 #if YAZ_POSIX_THREADS
 110     m_p->m_SocketObservable = obs;
 111
 112     pipe(m_p->m_fd);
 113     obs->addObserver(m_p->m_fd[0], this);
 114     obs->maskObserver(this, SOCKET_OBSERVE_READ);
 115
 116     m_p->m_stop_flag = false;
 117     pthread_mutex_init(&m_p->m_mutex_input_data, 0);
 118     pthread_cond_init(&m_p->m_cond_input_data, 0);
 119     pthread_mutex_init(&m_p->m_mutex_output_data, 0);
 120
 121     m_p->m_no_threads = no_threads;
 122     m_p->m_thread_id = new pthread_t[no_threads];
 123     int i;
 124     for (i = 0; i<m_p->m_no_threads; i++)
 125         pthread_create(&m_p->m_thread_id[i], 0, tfunc, this);
 126 #endif
 127 }
 128
 129 Msg_Thread::~Msg_Thread()
 130 {
 131 #if YAZ_POSIX_THREADS
 132     pthread_mutex_lock(&m_p->m_mutex_input_data);
 133     m_p->m_stop_flag = true;
 134     pthread_cond_broadcast(&m_p->m_cond_input_data);
 135     pthread_mutex_unlock(&m_p->m_mutex_input_data);
 136
 137     int i;
 138     for (i = 0; i<m_p->m_no_threads; i++)
 139         pthread_join(m_p->m_thread_id[i], 0);
 140     delete [] m_p->m_thread_id;
 141
 142     m_p->m_SocketObservable->deleteObserver(this);
 143
 144     pthread_cond_destroy(&m_p->m_cond_input_data);
 145     pthread_mutex_destroy(&m_p->m_mutex_input_data);
 146     pthread_mutex_destroy(&m_p->m_mutex_output_data);
 147     close(m_p->m_fd[0]);
 148     close(m_p->m_fd[1]);
 149 #endif
 150
 151     delete m_p;
 152 }
 153
 154 void Msg_Thread::socketNotify(int event)
 155 {
 156 #if YAZ_POSIX_THREADS
 157     if (event & SOCKET_OBSERVE_READ)
 158     {
 159         char buf[2];
 160         read(m_p->m_fd[0], buf, 1);
 161         pthread_mutex_lock(&m_p->m_mutex_output_data);
 162         IMsg_Thread *out = m_p->m_output.dequeue();
 163         pthread_mutex_unlock(&m_p->m_mutex_output_data);
 164         if (out)
 165             out->result();
 166     }
 167 #endif
 168 }
 169
 170 #if YAZ_POSIX_THREADS
 171 void Msg_Thread::run(void *p)
 172 {
 173     while(1)
 174     {
 175         pthread_mutex_lock(&m_p->m_mutex_input_data);
 176         while (!m_p->m_stop_flag && m_p->m_input.size() == 0)
 177             pthread_cond_wait(&m_p->m_cond_input_data, &m_p->m_mutex_input_data);
 178         if (m_p->m_stop_flag)
 179         {
 180             pthread_mutex_unlock(&m_p->m_mutex_input_data);
 181             break;
 182         }
 183         IMsg_Thread *in = m_p->m_input.dequeue();
 184         pthread_mutex_unlock(&m_p->m_mutex_input_data);
 185
 186         IMsg_Thread *out = in->handle();
 187         pthread_mutex_lock(&m_p->m_mutex_output_data);
 188         m_p->m_output.enqueue(out);
 189
 190         write(m_p->m_fd[1], "", 1);
 191         pthread_mutex_unlock(&m_p->m_mutex_output_data);
 192     }
 193 }
 194 #endif
 195
 196 void Msg_Thread::put(IMsg_Thread *m)
 197 {
 198 #if YAZ_POSIX_THREADS
 199     pthread_mutex_lock(&m_p->m_mutex_input_data);
 200     m_p->m_input.enqueue(m);
 201     pthread_cond_signal(&m_p->m_cond_input_data);
 202     pthread_mutex_unlock(&m_p->m_mutex_input_data);
 203 #else
 204     IMsg_Thread *out = m->handle();
 205     if (out)
 206         out->result();
 207 #endif
 208 }
 209 /*
 210  * Local variables:
 211  * c-basic-offset: 4
 212  * indent-tabs-mode: nil
 213  * End:
 214  * vim: shiftwidth=4 tabstop=8 expandtab
 215  */
 216