From 33dbb42d6462b1b534212262f2df37af83b6e794 Mon Sep 17 00:00:00 2001 From: Adam Dickmeiss Date: Thu, 19 May 2005 21:29:58 +0000 Subject: [PATCH] New test for thread+socket manager communication --- src/Makefile.am | 4 +- src/tstthreads.cpp | 266 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 269 insertions(+), 1 deletion(-) create mode 100644 src/tstthreads.cpp diff --git a/src/Makefile.am b/src/Makefile.am index 3436bed..1b8a5ca 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1,4 +1,4 @@ -## $Id: Makefile.am,v 1.6 2005-05-04 08:31:44 adam Exp $ +## $Id: Makefile.am,v 1.7 2005-05-19 21:29:58 adam Exp $ AM_CXXFLAGS = $(YAZPPINC) -I$(srcdir)/../include $(XSLT_CFLAGS) $(USEMARCONINC) @@ -10,11 +10,13 @@ libyazproxy_la_SOURCES= yaz-proxy.cpp yaz-proxy-config.cpp yaz-bw.cpp \ bin_PROGRAMS = yazproxy check_PROGRAMS = cdetails +noinst_PROGRAMS = tstthreads TESTS=$(check_PROGRAMS) yazproxy_SOURCES=yaz-proxy-main.cpp cdetails_SOURCES=cdetails.cpp +tstthreads_SOURCES=tstthreads.cpp LDADD=libyazproxy.la $(YAZPPLALIB) $(XSLT_LIBS) $(USEMARCONLALIB) libyazproxy_la_LIBADD = $(XSLT_LIBS) diff --git a/src/tstthreads.cpp b/src/tstthreads.cpp new file mode 100644 index 0000000..b07fe03 --- /dev/null +++ b/src/tstthreads.cpp @@ -0,0 +1,266 @@ +/* $Id: tstthreads.cpp,v 1.1 2005-05-19 21:29: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 +#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); + + 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 { +public: + void destroy(); + Proxy_Msg *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() +{ + My_Msg *res = new My_Msg; + int sl = rand() % 5; + + res->m_val = m_val; + printf("My_Msg::handle val=%d sleep=%d\n", m_val, sl); + sleep(sl); + 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; +public: + My_Timer_Thread(IYazSocketObservable *obs, My_Thread *t); + void socketNotify(int event); +}; + +My_Timer_Thread::My_Timer_Thread(IYazSocketObservable *obs, + My_Thread *t) : m_obs(obs) +{ + pipe(m_fd); + m_t = t; + obs->addObserver(m_fd[0], this); + obs->timeoutObserver(this, 2); +} + +void My_Timer_Thread::socketNotify(int event) +{ + static int seq = 1; + printf("Add %d\n", seq); + My_Msg *m = new My_Msg; + m->m_val = seq++; + m_t->put(m); +} + +int main(int argc, char **argv) +{ + Yaz_SocketManager mySocketManager; + + My_Thread m(&mySocketManager); + My_Timer_Thread t(&mySocketManager, &m); + while (mySocketManager.processEvent() > 0) + ; + return 0; +} -- 1.7.10.4