1 /* This file is part of YAZ proxy
2 Copyright (C) 1998-2009 Index Data
4 YAZ proxy is free software; you can redistribute it and/or modify it under
5 the terms of the GNU General Public License as published by the Free
6 Software Foundation; either version 2, or (at your option) any later
9 YAZ proxy is distributed in the hope that it will be useful, but WITHOUT ANY
10 WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
14 You should have received a copy of the GNU General Public License
15 along with this program; if not, write to the Free Software
16 Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
30 #include <yazpp/socket-observer.h>
33 #include "msg-thread.h"
35 using namespace yazpp_1;
37 class Msg_Thread::Private {
40 Msg_Thread_Queue m_input;
41 Msg_Thread_Queue m_output;
44 yazpp_1::ISocketObservable *m_SocketObservable;
45 pthread_t *m_thread_id;
46 pthread_mutex_t m_mutex_input_data;
47 pthread_cond_t m_cond_input_data;
48 pthread_mutex_t m_mutex_output_data;
53 IMsg_Thread::~IMsg_Thread()
58 Msg_Thread_Queue::Msg_Thread_Queue()
63 int Msg_Thread_Queue::size()
66 Msg_Thread_Queue_List *l;
67 for (l = m_list; l; l = l->m_next)
72 void Msg_Thread_Queue::enqueue(IMsg_Thread *m)
74 Msg_Thread_Queue_List *l = new Msg_Thread_Queue_List;
80 IMsg_Thread *Msg_Thread_Queue::dequeue()
82 Msg_Thread_Queue_List **l = &m_list;
87 IMsg_Thread *m = (*l)->m_item;
94 static void *tfunc(void *p)
96 Msg_Thread *pt = (Msg_Thread *) p;
102 Msg_Thread::Msg_Thread(ISocketObservable *obs, int no_threads)
106 #if YAZ_POSIX_THREADS
107 m_p->m_SocketObservable = obs;
110 obs->addObserver(m_p->m_fd[0], this);
111 obs->maskObserver(this, SOCKET_OBSERVE_READ);
113 m_p->m_stop_flag = false;
114 pthread_mutex_init(&m_p->m_mutex_input_data, 0);
115 pthread_cond_init(&m_p->m_cond_input_data, 0);
116 pthread_mutex_init(&m_p->m_mutex_output_data, 0);
118 m_p->m_no_threads = no_threads;
119 m_p->m_thread_id = new pthread_t[no_threads];
121 for (i = 0; i<m_p->m_no_threads; i++)
122 pthread_create(&m_p->m_thread_id[i], 0, tfunc, this);
126 Msg_Thread::~Msg_Thread()
128 #if YAZ_POSIX_THREADS
129 pthread_mutex_lock(&m_p->m_mutex_input_data);
130 m_p->m_stop_flag = true;
131 pthread_cond_broadcast(&m_p->m_cond_input_data);
132 pthread_mutex_unlock(&m_p->m_mutex_input_data);
135 for (i = 0; i<m_p->m_no_threads; i++)
136 pthread_join(m_p->m_thread_id[i], 0);
137 delete [] m_p->m_thread_id;
139 m_p->m_SocketObservable->deleteObserver(this);
141 pthread_cond_destroy(&m_p->m_cond_input_data);
142 pthread_mutex_destroy(&m_p->m_mutex_input_data);
143 pthread_mutex_destroy(&m_p->m_mutex_output_data);
151 void Msg_Thread::socketNotify(int event)
153 #if YAZ_POSIX_THREADS
154 if (event & SOCKET_OBSERVE_READ)
157 read(m_p->m_fd[0], buf, 1);
158 pthread_mutex_lock(&m_p->m_mutex_output_data);
159 IMsg_Thread *out = m_p->m_output.dequeue();
160 pthread_mutex_unlock(&m_p->m_mutex_output_data);
167 #if YAZ_POSIX_THREADS
168 void Msg_Thread::run(void *p)
172 pthread_mutex_lock(&m_p->m_mutex_input_data);
173 while (!m_p->m_stop_flag && m_p->m_input.size() == 0)
174 pthread_cond_wait(&m_p->m_cond_input_data, &m_p->m_mutex_input_data);
175 if (m_p->m_stop_flag)
177 pthread_mutex_unlock(&m_p->m_mutex_input_data);
180 IMsg_Thread *in = m_p->m_input.dequeue();
181 pthread_mutex_unlock(&m_p->m_mutex_input_data);
183 IMsg_Thread *out = in->handle();
184 pthread_mutex_lock(&m_p->m_mutex_output_data);
185 m_p->m_output.enqueue(out);
187 write(m_p->m_fd[1], "", 1);
188 pthread_mutex_unlock(&m_p->m_mutex_output_data);
193 void Msg_Thread::put(IMsg_Thread *m)
195 #if YAZ_POSIX_THREADS
196 pthread_mutex_lock(&m_p->m_mutex_input_data);
197 m_p->m_input.enqueue(m);
198 pthread_cond_signal(&m_p->m_cond_input_data);
199 pthread_mutex_unlock(&m_p->m_mutex_input_data);
201 IMsg_Thread *out = m->handle();
209 * c-file-style: "Stroustrup"
210 * indent-tabs-mode: nil
212 * vim: shiftwidth=4 tabstop=8 expandtab