From: Adam Dickmeiss Date: Mon, 30 May 2005 20:08:58 +0000 (+0000) Subject: Implemented new class Msg_Thread which allows for threaded facilities X-Git-Tag: YAZPROXY.ERE2~14 X-Git-Url: http://git.indexdata.com/?p=yazproxy-moved-to-github.git;a=commitdiff_plain;h=7f81d5b3aa6b9d7e4c5c1e02c91dd416eb97d19e Implemented new class Msg_Thread which allows for threaded facilities combined with the socketObservable interface. That is it allows for threading combined with select/poll. --- diff --git a/src/msg-thread.cpp b/src/msg-thread.cpp new file mode 100644 index 0000000..3315e26 --- /dev/null +++ b/src/msg-thread.cpp @@ -0,0 +1,157 @@ +/* $Id: msg-thread.cpp,v 1.1 2005-05-30 20:08:58 adam Exp $ + Copyright (c) 1998-2005, Index Data. + +This file is part of the yaz-proxy. + +YAZ proxy is free software; you can redistribute it and/or modify it under +the terms of the GNU General Public License as published by the Free +Software Foundation; either version 2, or (at your option) any later +version. + +YAZ proxy is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or +FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License +for more details. + +You should have received a copy of the GNU General Public License +along with YAZ proxy; see the file LICENSE. If not, write to the +Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA +02111-1307, USA. + */ +#include +#include +#include + +#include +#include + +#include "msg-thread.h" + +Msg_Thread_Queue::Msg_Thread_Queue() +{ + m_list = 0; +} + +int Msg_Thread_Queue::size() +{ + int no = 0; + Msg_Thread_Queue_List *l; + for (l = m_list; l; l = l->m_next) + no++; + return no; +} + +void Msg_Thread_Queue::enqueue(IMsg_Thread *m) +{ + Msg_Thread_Queue_List *l = new Msg_Thread_Queue_List; + l->m_next = m_list; + l->m_item = m; + m_list = l; +} + +IMsg_Thread *Msg_Thread_Queue::dequeue() +{ + Msg_Thread_Queue_List **l = &m_list; + if (!*l) + return 0; + while ((*l)->m_next) + l = &(*l)->m_next; + IMsg_Thread *m = (*l)->m_item; + delete *l; + *l = 0; + return m; +} + +static void *tfunc(void *p) +{ + Msg_Thread *pt = (Msg_Thread *) p; + pt->run(0); + return 0; +} + + +Msg_Thread::Msg_Thread(IYazSocketObservable *obs) + : m_SocketObservable(obs) +{ + pipe(m_fd); + obs->addObserver(m_fd[0], this); + obs->maskObserver(this, YAZ_SOCKET_OBSERVE_READ); + + m_stop_flag = false; + pthread_mutex_init(&m_mutex_input_data, 0); + pthread_cond_init(&m_cond_input_data, 0); + pthread_mutex_init(&m_mutex_output_data, 0); + pthread_cond_init(&m_cond_output_data, 0); + pthread_create(&m_thread_id, 0, tfunc, this); +} + +Msg_Thread::~Msg_Thread() +{ + pthread_mutex_lock(&m_mutex_input_data); + m_stop_flag = true; + pthread_cond_signal(&m_cond_input_data); + pthread_mutex_unlock(&m_mutex_input_data); + + pthread_join(m_thread_id, 0); + + m_SocketObservable->deleteObserver(this); + + pthread_cond_destroy(&m_cond_input_data); + pthread_mutex_destroy(&m_mutex_input_data); + pthread_cond_destroy(&m_cond_output_data); + pthread_mutex_destroy(&m_mutex_output_data); + close(m_fd[0]); + close(m_fd[1]); +} + +void Msg_Thread::socketNotify(int event) +{ + if (event & YAZ_SOCKET_OBSERVE_READ) + { + char buf[2]; + read(m_fd[0], buf, 1); + pthread_mutex_lock(&m_mutex_output_data); + IMsg_Thread *out = m_output.dequeue(); + pthread_mutex_unlock(&m_mutex_output_data); + if (out) + out->result(); + } +} + +void Msg_Thread::run(void *p) +{ + while(1) + { + pthread_mutex_lock(&m_mutex_input_data); + pthread_cond_wait(&m_cond_input_data, &m_mutex_input_data); + while (1) + { + if (m_stop_flag) + { + pthread_mutex_unlock(&m_mutex_input_data); + return; + } + IMsg_Thread *in = m_input.dequeue(); + pthread_mutex_unlock(&m_mutex_input_data); + if (!in) + break; + IMsg_Thread *out = in->handle(); + pthread_mutex_lock(&m_mutex_output_data); + m_output.enqueue(out); + pthread_cond_signal(&m_cond_output_data); + pthread_mutex_unlock(&m_mutex_output_data); + + write(m_fd[1], "", 1); + + pthread_mutex_lock(&m_mutex_input_data); + } + } +} + +void Msg_Thread::put(IMsg_Thread *m) +{ + pthread_mutex_lock(&m_mutex_input_data); + m_input.enqueue(m); + pthread_cond_signal(&m_cond_input_data); + pthread_mutex_unlock(&m_mutex_input_data); +} diff --git a/src/msg-thread.h b/src/msg-thread.h new file mode 100644 index 0000000..0d0729f --- /dev/null +++ b/src/msg-thread.h @@ -0,0 +1,76 @@ +/* $Id: msg-thread.h,v 1.1 2005-05-30 20:08:58 adam Exp $ + Copyright (c) 1998-2005, Index Data. + +This file is part of the yaz-proxy. + +YAZ proxy is free software; you can redistribute it and/or modify it under +the terms of the GNU General Public License as published by the Free +Software Foundation; either version 2, or (at your option) any later +version. + +YAZ proxy is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or +FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License +for more details. + +You should have received a copy of the GNU General Public License +along with YAZ proxy; see the file LICENSE. If not, write to the +Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA +02111-1307, USA. + */ + +#include +#include +#include + +#if HAVE_DLFCN_H +#include +#endif + +#include +#include + +class IMsg_Thread { +public: + virtual IMsg_Thread *handle() = 0; + virtual void result() = 0; +}; + +class Msg_Thread_Queue_List { + friend class Msg_Thread_Queue; + private: + IMsg_Thread *m_item; + Msg_Thread_Queue_List *m_next; +}; + +class Msg_Thread_Queue { + public: + Msg_Thread_Queue(); + void enqueue(IMsg_Thread *in); + IMsg_Thread *dequeue(); + int size(); + private: + Msg_Thread_Queue_List *m_list; +}; + +class Msg_Thread : public IYazSocketObserver { + public: + Msg_Thread(IYazSocketObservable *obs); + virtual ~Msg_Thread(); + void socketNotify(int event); + void put(IMsg_Thread *m); + IMsg_Thread *get(); + void run(void *p); + int m_fd[2]; +private: + IYazSocketObservable *m_SocketObservable; + pthread_t m_thread_id; + Msg_Thread_Queue m_input; + Msg_Thread_Queue m_output; + pthread_mutex_t m_mutex_input_data; + pthread_cond_t m_cond_input_data; + pthread_mutex_t m_mutex_output_data; + pthread_cond_t m_cond_output_data; + bool m_stop_flag; +}; + diff --git a/src/tstthreads.cpp b/src/tstthreads.cpp index f1bf7f0..ae54b79 100644 --- a/src/tstthreads.cpp +++ b/src/tstthreads.cpp @@ -1,4 +1,4 @@ -/* $Id: tstthreads.cpp,v 1.2 2005-05-20 21:32:31 adam Exp $ +/* $Id: tstthreads.cpp,v 1.3 2005-05-30 20:08:58 adam Exp $ Copyright (c) 1998-2005, Index Data. This file is part of the yaz-proxy. @@ -20,192 +20,21 @@ Free Software Foundation, 59 Temple Place - Suite 330, Boston, MA */ #include -#include -#include #include -#if HAVE_DLFCN_H -#include -#endif - -#include +#include #include #include -#include "proxyp.h" - -class Proxy_Msg { -public: - virtual void destroy() = 0; - virtual Proxy_Msg *handle() = 0; - virtual void result() = 0; -}; - -class Proxy_Msg_Queue_List { - friend class Proxy_Msg_Queue; - private: - Proxy_Msg *m_item; - Proxy_Msg_Queue_List *m_next; -}; - -class Proxy_Msg_Queue { - public: - Proxy_Msg_Queue(); - void enqueue(Proxy_Msg *in); - Proxy_Msg *dequeue(); - int size(); - private: - Proxy_Msg_Queue_List *m_list; -}; - -Proxy_Msg_Queue::Proxy_Msg_Queue() -{ - m_list = 0; -} - -int Proxy_Msg_Queue::size() -{ - int no = 0; - Proxy_Msg_Queue_List *l; - for (l = m_list; l; l = l->m_next) - no++; - return no; -} - -void Proxy_Msg_Queue::enqueue(Proxy_Msg *m) -{ - Proxy_Msg_Queue_List *l = new Proxy_Msg_Queue_List; - l->m_next = m_list; - l->m_item = m; - m_list = l; -} - -Proxy_Msg *Proxy_Msg_Queue::dequeue() -{ - Proxy_Msg_Queue_List **l = &m_list; - if (!*l) - return 0; - while ((*l)->m_next) - l = &(*l)->m_next; - Proxy_Msg *m = (*l)->m_item; - delete *l; - *l = 0; - return m; -} - -class Proxy_Thread : public IYazSocketObserver { -public: - Proxy_Thread(IYazSocketObservable *obs); - virtual ~Proxy_Thread(); - void socketNotify(int event); - void put(Proxy_Msg *m); - Proxy_Msg *get(); - void run(void *p); -private: - IYazSocketObservable *m_obs; - int m_fd[2]; - pthread_t m_thread_id; - Proxy_Msg_Queue m_input; - Proxy_Msg_Queue m_output; - pthread_mutex_t m_mutex_input_data; - pthread_cond_t m_cond_input_data; - pthread_mutex_t m_mutex_output_data; - pthread_cond_t m_cond_output_data; -}; - -static void *tfunc(void *p) -{ - Proxy_Thread *pt = (Proxy_Thread *) p; - pt->run(0); - return 0; -} - - -Proxy_Thread::Proxy_Thread(IYazSocketObservable *obs) - : m_obs(obs) -{ - pthread_mutex_init(&m_mutex_input_data, 0); - pthread_cond_init(&m_cond_input_data, 0); - pthread_mutex_init(&m_mutex_output_data, 0); - pthread_cond_init(&m_cond_output_data, 0); - m_fd[0] = m_fd[1] = -1; - if (pipe(m_fd) != 0) - return; - m_obs->addObserver(m_fd[0], this); - m_obs->timeoutObserver(this, 2000); - m_obs->maskObserver(this, YAZ_SOCKET_OBSERVE_READ); - - pthread_create(&m_thread_id, 0, tfunc, this); -} - -Proxy_Thread::~Proxy_Thread() -{ - -} - -void Proxy_Thread::socketNotify(int event) -{ - char buf[2]; - read(m_fd[0], buf, 1); - pthread_mutex_lock(&m_mutex_output_data); - Proxy_Msg *out = m_output.dequeue(); - pthread_mutex_unlock(&m_mutex_output_data); - if (out) - out->result(); -} - -void Proxy_Thread::run(void *p) -{ - while(1) - { - pthread_mutex_lock(&m_mutex_input_data); - pthread_cond_wait(&m_cond_input_data, &m_mutex_input_data); - while(1) - { - Proxy_Msg *in = m_input.dequeue(); - pthread_mutex_unlock(&m_mutex_input_data); - if (!in) - break; - Proxy_Msg *out = in->handle(); - pthread_mutex_lock(&m_mutex_output_data); - m_output.enqueue(out); - pthread_cond_signal(&m_cond_output_data); - pthread_mutex_unlock(&m_mutex_output_data); - write(m_fd[1], "", 1); +#include "msg-thread.h" - pthread_mutex_lock(&m_mutex_input_data); - } - } -} - -void Proxy_Thread::put(Proxy_Msg *m) -{ - pthread_mutex_lock(&m_mutex_input_data); - m_input.enqueue(m); - pthread_cond_signal(&m_cond_input_data); - int in_size = m_input.size(); - pthread_mutex_unlock(&m_mutex_input_data); - int out_size = m_output.size(); - printf("in-size=%d out-size=%d\n", in_size, out_size); -} - -class My_Msg : public Proxy_Msg { +class My_Msg : public IMsg_Thread { public: - void destroy(); - Proxy_Msg *handle(); + IMsg_Thread *handle(); void result(); int m_val; }; -class My_Thread : public Proxy_Thread { -public: - My_Thread(IYazSocketObservable *obs); -}; - -My_Thread::My_Thread(IYazSocketObservable *obs) : Proxy_Thread(obs) -{ -} - -Proxy_Msg *My_Msg::handle() +IMsg_Thread *My_Msg::handle() { My_Msg *res = new My_Msg; int sl = rand() % 5; @@ -216,33 +45,28 @@ Proxy_Msg *My_Msg::handle() return res; } - void My_Msg::result() { printf("My_Msg::result val=%d\n", m_val); } -void My_Msg::destroy() -{ - delete this; -} - class My_Timer_Thread : public IYazSocketObserver { private: IYazSocketObservable *m_obs; int m_fd[2]; - My_Thread *m_t; + Msg_Thread *m_t; public: - My_Timer_Thread(IYazSocketObservable *obs, My_Thread *t); + My_Timer_Thread(IYazSocketObservable *obs, Msg_Thread *t); void socketNotify(int event); }; My_Timer_Thread::My_Timer_Thread(IYazSocketObservable *obs, - My_Thread *t) : m_obs(obs) + Msg_Thread *t) : m_obs(obs) { pipe(m_fd); m_t = t; obs->addObserver(m_fd[0], this); + obs->maskObserver(this, YAZ_SOCKET_OBSERVE_READ); obs->timeoutObserver(this, 2); } @@ -259,9 +83,10 @@ int main(int argc, char **argv) { Yaz_SocketManager mySocketManager; - My_Thread m(&mySocketManager); - My_Timer_Thread t(&mySocketManager, &m); - while (mySocketManager.processEvent() > 0) + Msg_Thread m(&mySocketManager); + My_Timer_Thread t(&mySocketManager, &m) ; + int i = 0; + while (++i < 5 && mySocketManager.processEvent() > 0) ; return 0; }