From bd71f8812ca0f38438733efc89ecce1f49dae9e7 Mon Sep 17 00:00:00 2001 From: Adam Dickmeiss Date: Mon, 26 Mar 2001 14:43:49 +0000 Subject: [PATCH] New threaded PDU association. --- buildconf.sh | 15 +- configure.in | 12 +- include/yaz++/yaz-pdu-assoc.h | 30 +++- include/yaz++/yaz-pdu-observer.h | 8 +- include/yaz++/yaz-proxy.h | 8 +- include/yaz++/yaz-z-server.h | 27 ++- src/Makefile.am | 6 +- src/yaz-client.cpp | 12 +- src/yaz-pdu-assoc-thread.cpp | 72 ++++++++ src/yaz-pdu-assoc.cpp | 335 ++++++++++++++++++++------------------ src/yaz-proxy.cpp | 15 +- src/yaz-server.cpp | 30 ++-- src/yaz-socket-manager.cpp | 15 +- src/yaz-z-assoc.cpp | 9 +- src/yaz-z-server.cpp | 13 +- yaz++-config.in | 7 +- yaz.m4 | 9 +- 17 files changed, 394 insertions(+), 229 deletions(-) create mode 100644 src/yaz-pdu-assoc-thread.cpp diff --git a/buildconf.sh b/buildconf.sh index d16ea33..071751e 100755 --- a/buildconf.sh +++ b/buildconf.sh @@ -1,11 +1,14 @@ #!/bin/sh -# $Id: buildconf.sh,v 1.5 2001-02-21 11:25:07 adam Exp $ +# $Id: buildconf.sh,v 1.6 2001-03-26 14:43:49 adam Exp $ dir=`aclocal --print-ac-dir` if [ -f $dir/yaz.m4 ]; then - aclocal + aclocal || exit 1 else - aclocal -I . + aclocal -I . || exit 1 +fi +libtoolize --force >/dev/null 2>&1 || exit 2 +automake -a >/dev/null 2>&1 || exit 3 +autoconf || exit 4 +if [ -f config.cache ]; then + rm config.cache fi -libtoolize --force -automake -a -autoconf diff --git a/configure.in b/configure.in index 9d3674e..cbb90bf 100644 --- a/configure.in +++ b/configure.in @@ -8,15 +8,17 @@ AC_HEADER_STDC AM_DISABLE_SHARED AM_PROG_LIBTOOL -YAZ_INIT +YAZ_INIT(threads) -YAZPP_SRC_ROOT=`pwd` AC_SUBST(YAZPP_SRC_ROOT) +AC_SUBST(YAZPP_BUILD_ROOT) +YAZPP_SRC_ROOT=`cd ${srcdir}; pwd` +YAZPP_BUILD_ROOT=`pwd` -if test -f src/yaz++-config.in; then - rm src/yaz++-config.in +if test -f ${srcdir}/src/yaz++-config.in; then + rm ${srcdir}/src/yaz++-config.in fi -sed s%yaz_echo_source=yes%yaz_echo_source=no%g < yaz++-config.in >src/yaz++-config.in +sed s%yaz_echo_source=yes%yaz_echo_source=no%g < ${srcdir}/yaz++-config.in >${srcdir}/src/yaz++-config.in AC_OUTPUT([ Makefile src/Makefile diff --git a/include/yaz++/yaz-pdu-assoc.h b/include/yaz++/yaz-pdu-assoc.h index 10cfbe9..c3b2b35 100644 --- a/include/yaz++/yaz-pdu-assoc.h +++ b/include/yaz++/yaz-pdu-assoc.h @@ -2,7 +2,7 @@ * Copyright (c) 1998-2000, Index Data. * See the file LICENSE for details. * - * $Id: yaz-pdu-assoc.h,v 1.1 2000-10-11 11:58:16 adam Exp $ + * $Id: yaz-pdu-assoc.h,v 1.2 2001-03-26 14:43:49 adam Exp $ */ #ifndef YAZ_PDU_ASSOC_INCLUDED @@ -19,8 +19,16 @@ listen method. */ class YAZ_EXPORT Yaz_PDU_Assoc : public IYaz_PDU_Observable, IYazSocketObserver { + friend class Yaz_PDU_AssocThread; private: - enum { Connecting, Listen, Ready, Closed } m_state; + enum { + Connecting, + Listen, + Ready, + Closed, + Writing, + Accepting + } m_state; class PDU_Queue { public: PDU_Queue(const char *buf, int len); @@ -42,10 +50,14 @@ class YAZ_EXPORT Yaz_PDU_Assoc : public IYaz_PDU_Observable, IYazSocketObserver int *m_destroyed; int m_idleTime; int m_log; + void init(IYazSocketObservable *socketObservable); public: - COMSTACK comstack(); + COMSTACK comstack(const char *type_and_host, void **vp); /// Create object using specified socketObservable Yaz_PDU_Assoc(IYazSocketObservable *socketObservable); + /// Create Object using existing comstack + Yaz_PDU_Assoc(IYazSocketObservable *socketObservable, + COMSTACK cs); /// Close socket and destroy object. /// virtual ~Yaz_PDU_Assoc(); /// Clone the object @@ -56,8 +68,6 @@ class YAZ_EXPORT Yaz_PDU_Assoc : public IYaz_PDU_Observable, IYazSocketObserver void connect(IYaz_PDU_Observer *observer, const char *addr); /// listen for clients (server role) void listen(IYaz_PDU_Observer *observer, const char *addr); - /// open with existing socket - void socket(IYaz_PDU_Observer *observer, int fd); /// Socket notification void socketNotify(int event); /// Close socket @@ -67,7 +77,15 @@ class YAZ_EXPORT Yaz_PDU_Assoc : public IYaz_PDU_Observable, IYazSocketObserver /// Set Idle Time void idleTime (int timeout); /// Child start... - virtual void childNotify(int fd); + virtual void childNotify(COMSTACK cs); }; +class YAZ_EXPORT Yaz_PDU_AssocThread : public Yaz_PDU_Assoc { + public: + Yaz_PDU_AssocThread(IYazSocketObservable *socketObservable); + private: + void childNotify(COMSTACK cs); + +}; #endif + diff --git a/include/yaz++/yaz-pdu-observer.h b/include/yaz++/yaz-pdu-observer.h index 99cd5a9..61daf78 100644 --- a/include/yaz++/yaz-pdu-observer.h +++ b/include/yaz++/yaz-pdu-observer.h @@ -2,7 +2,7 @@ * Copyright (c) 1998-2000, Index Data. * See the file LICENSE for details. * - * $Id: yaz-pdu-observer.h,v 1.2 2000-11-01 14:22:59 adam Exp $ + * $Id: yaz-pdu-observer.h,v 1.3 2001-03-26 14:43:49 adam Exp $ */ #ifndef YAZ_PDU_OBSERVER_H @@ -32,8 +32,6 @@ class YAZ_EXPORT IYaz_PDU_Observable { virtual void destroy() = 0; /// Set Idle Time virtual void idleTime (int timeout) = 0; - /// open with existing socket - virtual void socket(IYaz_PDU_Observer *observer, int fd) = 0; }; /** Protocol Data Unit Observer. @@ -51,8 +49,8 @@ class YAZ_EXPORT IYaz_PDU_Observer { /// Called whenever there is a timeout virtual void timeoutNotify() = 0; /// Make clone of observer using IYaz_PDU_Observable interface - virtual IYaz_PDU_Observer *clone(IYaz_PDU_Observable *the_PDU_Observable, - int fd) = 0; + virtual IYaz_PDU_Observer *sessionNotify( + IYaz_PDU_Observable *the_PDU_Observable, int fd) = 0; }; #endif diff --git a/include/yaz++/yaz-proxy.h b/include/yaz++/yaz-proxy.h index de8fe0f..e4f3e5e 100644 --- a/include/yaz++/yaz-proxy.h +++ b/include/yaz++/yaz-proxy.h @@ -2,7 +2,7 @@ * Copyright (c) 1998-2000, Index Data. * See the file LICENSE for details. * - * $Id: yaz-proxy.h,v 1.2 2000-11-01 14:22:59 adam Exp $ + * $Id: yaz-proxy.h,v 1.3 2001-03-26 14:43:49 adam Exp $ */ #include @@ -16,7 +16,8 @@ class YAZ_EXPORT Yaz_ProxyClient : public Yaz_Z_Assoc { Yaz_ProxyClient(IYaz_PDU_Observable *the_PDU_Observable); ~Yaz_ProxyClient(); void recv_Z_PDU(Z_APDU *apdu); - IYaz_PDU_Observer* clone(IYaz_PDU_Observable *the_PDU_Observable, int fd); + IYaz_PDU_Observer* sessionNotify + (IYaz_PDU_Observable *the_PDU_Observable, int fd); void shutdown(); Yaz_Proxy *m_server; void failNotify(); @@ -55,7 +56,8 @@ class YAZ_EXPORT Yaz_Proxy : public Yaz_Z_Assoc { Yaz_Proxy(IYaz_PDU_Observable *the_PDU_Observable); ~Yaz_Proxy(); void recv_Z_PDU(Z_APDU *apdu); - IYaz_PDU_Observer* clone(IYaz_PDU_Observable *the_PDU_Observable, int fd); + IYaz_PDU_Observer* sessionNotify + (IYaz_PDU_Observable *the_PDU_Observable, int fd); void failNotify(); void timeoutNotify(); void connectNotify(); diff --git a/include/yaz++/yaz-z-server.h b/include/yaz++/yaz-z-server.h index 5b4a0eb..1e606fa 100644 --- a/include/yaz++/yaz-z-server.h +++ b/include/yaz++/yaz-z-server.h @@ -1,12 +1,34 @@ /* - * Copyright (c) 2000, Index Data. + * Copyright (c) 2000-2001, Index Data. * See the file LICENSE for details. * - * $Id: yaz-z-server.h,v 1.1 2000-10-11 11:58:16 adam Exp $ + * $Id: yaz-z-server.h,v 1.2 2001-03-26 14:43:49 adam Exp $ */ #include +class Yaz_Z_Server; + +class YAZ_EXPORT Yaz_I_Facility { + public: + virtual const char *get_name(Yaz_Z_Server *server) = 0; + virtual int open(Yaz_Z_Server *server, + Z_InitRequest *initRequest, + Z_InitResponse *initResponse) = 0; + virtual int recv_PDU(Yaz_Z_Server *server, Z_APDU *apdu) = 0; + virtual int close(Yaz_Z_Server *server) = 0; +}; + +class YAZ_EXPORT Yaz_Facility_Retrieval : public Yaz_I_Facility { + public: + const char *get_name(Yaz_Z_Server *server); + int open(Yaz_Z_Server *server, + Z_InitRequest *initRequest, + Z_InitResponse *initResponse); + int recv_PDU(Yaz_Z_Server *server, Z_APDU *apdu); + int close(Yaz_Z_Server *server); +}; + class YAZ_EXPORT Yaz_Z_Server : public Yaz_Z_Assoc { public: Yaz_Z_Server(IYaz_PDU_Observable *the_PDU_Observable); @@ -31,7 +53,6 @@ public: void create_surrogateDiagnostics(Z_NamePlusRecord *rec, const char *dbname, int error, char *const addinfo); - private: Z_Records *pack_records (const char *resultSetName, int start, int num, diff --git a/src/Makefile.am b/src/Makefile.am index b2e96e3..c04230c 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1,12 +1,12 @@ -## $Id: Makefile.am,v 1.1 2000-10-11 11:58:16 adam Exp $ +## $Id: Makefile.am,v 1.2 2001-03-26 14:43:49 adam Exp $ -INCLUDES=$(YAZINC) -I../include +INCLUDES=$(YAZINC) -I$(srcdir)/../include lib_LTLIBRARIES = libyaz++.la libyaz___la_SOURCES=yaz-socket-manager.cpp yaz-pdu-assoc.cpp \ yaz-z-assoc.cpp yaz-proxy.cpp yaz-z-query.cpp yaz-ir-assoc.cpp \ - yaz-z-server.cpp + yaz-z-server.cpp yaz-pdu-assoc-thread.cpp libyaz___la_LDFLAGS=-version-info 0:0:0 diff --git a/src/yaz-client.cpp b/src/yaz-client.cpp index 8670d4f..a14956e 100644 --- a/src/yaz-client.cpp +++ b/src/yaz-client.cpp @@ -3,7 +3,10 @@ * See the file LICENSE for details. * * $Log: yaz-client.cpp,v $ - * Revision 1.16 2000-11-01 14:22:59 adam + * Revision 1.17 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.16 2000/11/01 14:22:59 adam * Added fd parameter for method IYaz_PDU_Observer::clone. * * Revision 1.15 2000/10/11 11:58:16 adam @@ -85,7 +88,8 @@ private: public: MyClient(IYaz_PDU_Observable *the_PDU_Observable, Yaz_SocketManager *the_SocketManager); - IYaz_PDU_Observer *clone(IYaz_PDU_Observable *the_PDU_Observable, int fd); + IYaz_PDU_Observer *sessionNotify( + IYaz_PDU_Observable *the_PDU_Observable, int fd); int args(Yaz_SocketManager *socketManager, int argc, char **argv); int interactive(Yaz_SocketManager *socketManager); int wait(); @@ -138,8 +142,8 @@ void MyClient::failNotify() set_lastReceived(-1); } -IYaz_PDU_Observer *MyClient::clone(IYaz_PDU_Observable *the_PDU_Observable, - int fd) +IYaz_PDU_Observer *MyClient::sessionNotify( + IYaz_PDU_Observable *the_PDU_Observable, int fd) { return new MyClient(the_PDU_Observable, m_socketManager); } diff --git a/src/yaz-pdu-assoc-thread.cpp b/src/yaz-pdu-assoc-thread.cpp new file mode 100644 index 0000000..5f7a20e --- /dev/null +++ b/src/yaz-pdu-assoc-thread.cpp @@ -0,0 +1,72 @@ +/* + * Copyright (c) 1998-2001, Index Data. + * See the file LICENSE for details. + * + * $Log: yaz-pdu-assoc-thread.cpp,v $ + * Revision 1.1 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + */ + +#include +#include + +#include +#include + +#ifdef WIN32 +#include +#else +#include +#endif + + +Yaz_PDU_AssocThread::Yaz_PDU_AssocThread( + IYazSocketObservable *socketObservable) + : Yaz_PDU_Assoc(socketObservable) +{ + +} + +#ifdef WIN32 +void __cdecl +#else +void * +#endif +events(void *p) +{ + Yaz_SocketManager *s = (Yaz_SocketManager *) p; + + logf (LOG_LOG, "thread started"); + while (s->processEvent() > 0) + ; + logf (LOG_LOG, "thread finished"); +#ifdef WIN32 +#else + return 0; +#endif +} + +void Yaz_PDU_AssocThread::childNotify(COMSTACK cs) +{ + Yaz_SocketManager *socket_observable = new Yaz_SocketManager; + Yaz_PDU_Assoc *new_observable = new Yaz_PDU_Assoc (socket_observable, cs); + + /// Clone PDU Observer + new_observable->m_PDU_Observer = + m_PDU_Observer->sessionNotify(new_observable, cs_fileno(cs)); +#ifdef WIN32 + long t_id; + t_id = _beginthread (events, 0, socket_observable); + if (t_id == -1) + { + yaz_log (LOG_FATAL|LOG_ERRNO, "_beginthread failed"); + exit (1); + } +#else + pthread_t type; + + int id = pthread_create (&type, 0, events, socket_observable); + yaz_log (LOG_LOG, "pthread_create returned id=%d", id); +#endif +} diff --git a/src/yaz-pdu-assoc.cpp b/src/yaz-pdu-assoc.cpp index 1c009b4..566f9b0 100644 --- a/src/yaz-pdu-assoc.cpp +++ b/src/yaz-pdu-assoc.cpp @@ -1,9 +1,12 @@ /* - * Copyright (c) 1998-2000, Index Data. + * Copyright (c) 1998-2001, Index Data. * See the file LICENSE for details. * * $Log: yaz-pdu-assoc.cpp,v $ - * Revision 1.22 2001-01-29 11:18:24 adam + * Revision 1.23 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.22 2001/01/29 11:18:24 adam * Server sets OPTIONS search and present. * * Revision 1.21 2000/11/20 14:17:36 adam @@ -79,13 +82,14 @@ */ #include +#include +#include +#include #include -#include -#include -Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable) +void Yaz_PDU_Assoc::init(IYazSocketObservable *socketObservable) { m_state = Closed; m_cs = 0; @@ -99,9 +103,43 @@ Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable) m_next = 0; m_destroyed = 0; m_idleTime = 0; - m_log = LOG_LOG; + m_log = LOG_DEBUG; +} + +Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable) +{ + init (socketObservable); +} + +Yaz_PDU_Assoc::Yaz_PDU_Assoc(IYazSocketObservable *socketObservable, + COMSTACK cs) +{ + init(socketObservable); + m_cs = cs; + unsigned mask = 0; + if (cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + m_socketObservable->addObserver(cs_fileno(cs), this); + if (!mask) + { + yaz_log (m_log, "new PDU_Assoc. Ready"); + m_state = Ready; + flush_PDU(); + } + else + { + yaz_log (m_log, "new PDU_Assoc. Accepting"); + // assume comstack is accepting... + m_state = Accepting; + m_socketObservable->addObserver(cs_fileno(cs), this); + m_socketObservable->maskObserver(this, + mask |YAZ_SOCKET_OBSERVE_EXCEPT); + } } + IYaz_PDU_Observable *Yaz_PDU_Assoc::clone() { Yaz_PDU_Assoc *copy = new Yaz_PDU_Assoc(m_socketObservable); @@ -110,26 +148,75 @@ IYaz_PDU_Observable *Yaz_PDU_Assoc::clone() void Yaz_PDU_Assoc::socketNotify(int event) { - logf (m_log, "Yaz_PDU_Assoc::socketNotify p=%p event = %d", this, event); + yaz_log (m_log, "Yaz_PDU_Assoc::socketNotify p=%p state=%d event = %d", + this, m_state, event); + if (event & YAZ_SOCKET_OBSERVE_EXCEPT) + { + close(); + m_PDU_Observer->failNotify(); + return; + } + else if (event & YAZ_SOCKET_OBSERVE_TIMEOUT) + { + m_PDU_Observer->timeoutNotify(); + return; + } switch (m_state) { + case Accepting: + if (!cs_accept (m_cs)) + { + yaz_log (m_log, "Yaz_PDU_Assoc::cs_accept failed"); + m_cs = 0; + close(); + m_PDU_Observer->failNotify(); + } + else + { + unsigned mask = 0; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + if (!mask) + { // accept is complete. turn to ready state and write if needed + m_state = Ready; + flush_PDU(); + } + else + { // accept still incomplete. + m_socketObservable->maskObserver(this, + mask|YAZ_SOCKET_OBSERVE_EXCEPT); + } + } + break; case Connecting: - if (event & (YAZ_SOCKET_OBSERVE_READ|YAZ_SOCKET_OBSERVE_EXCEPT)) + if (event & YAZ_SOCKET_OBSERVE_READ && + event & YAZ_SOCKET_OBSERVE_WRITE) { + // For Unix: if both read and write is set, then connect failed. close(); m_PDU_Observer->failNotify(); } - else if (event & YAZ_SOCKET_OBSERVE_TIMEOUT) - { - m_PDU_Observer->timeoutNotify(); - } - else + else { - m_state = Ready; - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT); - m_PDU_Observer->connectNotify(); - flush_PDU(); + yaz_log (m_log, "cs_connect again"); + int res = cs_connect (m_cs, 0); + if (res == 1) + { + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + m_socketObservable->maskObserver(this, mask); + } + else + { + m_state = Ready; + m_PDU_Observer->connectNotify(); + flush_PDU(); + } } break; case Listen: @@ -142,7 +229,7 @@ void Yaz_PDU_Assoc::socketNotify(int event) return; if (res < 0) { - logf(LOG_FATAL|LOG_ERRNO, "cs_listen failed"); + yaz_log(LOG_FATAL|LOG_ERRNO, "cs_listen failed"); return; } if (!(new_line = cs_accept(m_cs))) @@ -153,28 +240,32 @@ void Yaz_PDU_Assoc::socketNotify(int event) setup observer for child fileid in pdu-assoc 4. start thread */ - int fd = cs_fileno(new_line); - logf (m_log, "accept ok fd = %d", fd); - cs_fileno(new_line) = -1; - cs_close (new_line); - childNotify(fd); + childNotify (new_line); } break; + case Writing: + if (event & (YAZ_SOCKET_OBSERVE_READ|YAZ_SOCKET_OBSERVE_WRITE)) + flush_PDU(); + break; case Ready: - if (event & YAZ_SOCKET_OBSERVE_WRITE) - { - flush_PDU(); - } - if (event & YAZ_SOCKET_OBSERVE_READ) + if (event & (YAZ_SOCKET_OBSERVE_READ|YAZ_SOCKET_OBSERVE_WRITE)) { do { int res = cs_get (m_cs, &m_input_buf, &m_input_len); if (res == 1) + { + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + m_socketObservable->maskObserver(this, mask); return; + } else if (res <= 0) { - logf (m_log, "Connection closed by peer"); + yaz_log (m_log, "Yaz_PDU_Assoc::Connection closed by peer"); close(); m_PDU_Observer->failNotify(); // problem here.. return; @@ -188,10 +279,10 @@ void Yaz_PDU_Assoc::socketNotify(int event) if (destroyed) // it really was destroyed, return now. return; } while (m_cs && cs_more (m_cs)); - } - if (event & YAZ_SOCKET_OBSERVE_TIMEOUT) - { - m_PDU_Observer->timeoutNotify(); + if (m_cs) + m_socketObservable->maskObserver(this, + YAZ_SOCKET_OBSERVE_EXCEPT| + YAZ_SOCKET_OBSERVE_READ); } break; case Closed: @@ -273,8 +364,7 @@ int Yaz_PDU_Assoc::flush_PDU() { int r; - logf (m_log, "Yaz_PDU_Assoc::flush_PDU"); - if (m_state != Ready) + if (m_state != Ready && m_state != Writing) { logf (m_log, "YAZ_PDU_Assoc::flush_PDU, not ready"); return 1; @@ -282,27 +372,37 @@ int Yaz_PDU_Assoc::flush_PDU() PDU_Queue *q = m_queue_out; if (!q) { + m_state = Ready; + logf (m_log, "YAZ_PDU_Assoc::flush_PDU queue empty"); m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| + YAZ_SOCKET_OBSERVE_WRITE| YAZ_SOCKET_OBSERVE_EXCEPT); return 0; } r = cs_put (m_cs, q->m_buf, q->m_len); if (r < 0) { + logf (m_log, "Yaz_PDU_Assoc::flush_PDU cs_put failed"); close(); m_PDU_Observer->failNotify(); return r; } if (r == 1) { - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT| - YAZ_SOCKET_OBSERVE_WRITE); - logf (m_log, "Yaz_PDU_Assoc::flush_PDU put %d bytes (incomplete)", + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + m_state = Writing; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + + m_socketObservable->maskObserver(this, mask); + logf (m_log, "Yaz_PDU_Assoc::flush_PDU cs_put %d bytes (incomplete)", q->m_len); return r; - } - logf (m_log, "Yaz_PDU_Assoc::flush_PDU put %d bytes", q->m_len); + } + m_state = Ready; + logf (m_log, "Yaz_PDU_Assoc::flush_PDU cs_put %d bytes", q->m_len); // whole packet sent... delete this and proceed to next ... m_queue_out = q->m_next; delete q; @@ -335,36 +435,30 @@ int Yaz_PDU_Assoc::send_PDU(const char *buf, int len) return 0; } -COMSTACK Yaz_PDU_Assoc::comstack() +COMSTACK Yaz_PDU_Assoc::comstack(const char *type_and_host, void **vp) { - if (!m_cs) - { - CS_TYPE cs_type = tcpip_type; - m_cs = cs_create (cs_type, 0, PROTO_Z3950); - } - return m_cs; + return cs_create_host(type_and_host, 0, vp); } void Yaz_PDU_Assoc::listen(IYaz_PDU_Observer *observer, const char *addr) { close(); - void *ap; - COMSTACK cs = comstack(); - logf (m_log, "Yaz_PDU_Assoc::listen %s", addr); + logf (LOG_LOG, "Adding listener %s", addr); + m_PDU_Observer = observer; - if (!cs) - return; - ap = cs_straddr (cs, addr); - if (!ap) + void *ap; + m_cs = comstack(addr, &ap); + + if (!m_cs) return; - if (cs_bind(cs, ap, CS_SERVER) < 0) + if (cs_bind(m_cs, ap, CS_SERVER) < 0) return; - m_socketObservable->addObserver(cs_fileno(cs), this); + m_socketObservable->addObserver(cs_fileno(m_cs), this); m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| YAZ_SOCKET_OBSERVE_EXCEPT); - logf (m_log, "Yaz_PDU_Assoc::listen ok fd=%d", cs_fileno(cs)); + logf (m_log, "Yaz_PDU_Assoc::listen ok fd=%d", cs_fileno(m_cs)); m_state = Listen; } @@ -381,111 +475,38 @@ void Yaz_PDU_Assoc::connect(IYaz_PDU_Observer *observer, logf (m_log, "Yaz_PDU_Assoc::connect %s", addr); close(); m_PDU_Observer = observer; - COMSTACK cs = comstack(); - void *ap = cs_straddr (cs, addr); - if (!ap) - { - logf (m_log, "cs_straddr failed"); - return; - } - int res = cs_connect (cs, ap); - logf (m_log, "Yaz_PDU_Assoc::connect fd=%d res=%d", cs_fileno(cs), res); - m_socketObservable->addObserver(cs_fileno(cs), this); - m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_READ| - YAZ_SOCKET_OBSERVE_EXCEPT| - YAZ_SOCKET_OBSERVE_WRITE); + void *ap; + m_cs = comstack(addr, &ap); + int res = cs_connect (m_cs, ap); + logf (m_log, "Yaz_PDU_Assoc::connect fd=%d res=%d", cs_fileno(m_cs), res); + m_socketObservable->addObserver(cs_fileno(m_cs), this); + if (res >= 0) + { // Connect pending or complet m_state = Connecting; - // if res < 0, then cs_connect failed immediately -> m_state is Closed.. -} - -void Yaz_PDU_Assoc::socket(IYaz_PDU_Observer *observer, int fd) -{ - close(); - m_PDU_Observer = observer; - if (fd >= 0) - { - CS_TYPE cs_type = tcpip_type; - m_cs = cs_createbysocket(fd, cs_type, 0, PROTO_Z3950); - m_state = Ready; - m_socketObservable->addObserver(fd, this); - m_socketObservable->maskObserver(this, - YAZ_SOCKET_OBSERVE_READ| + unsigned mask = YAZ_SOCKET_OBSERVE_EXCEPT; + if (m_cs->io_pending & CS_WANT_WRITE) + mask |= YAZ_SOCKET_OBSERVE_WRITE; + if (m_cs->io_pending & CS_WANT_READ) + mask |= YAZ_SOCKET_OBSERVE_READ; + m_socketObservable->maskObserver(this, mask); + } + else + { // Connect failed immediately + // Since m_state is Closed we can distinguish this case from + // normal connect in socketNotify handler + m_socketObservable->maskObserver(this, YAZ_SOCKET_OBSERVE_WRITE| YAZ_SOCKET_OBSERVE_EXCEPT); - m_socketObservable->timeoutObserver(this, m_idleTime); } } -#if 1 - // 1 = single-threaded - // 0 = multi-threaded - // Single-threaded... Only useful for non-blocking handlers -void Yaz_PDU_Assoc::childNotify(int fd) -{ - // Clone PDU Observable (keep socket manager) - IYaz_PDU_Observable *new_observable = clone(); - - // Clone PDU Observer - IYaz_PDU_Observer *observer = m_PDU_Observer->clone(new_observable, fd); - - // Attach new socket to it - new_observable->socket(observer, fd); -} -#else - -#include - -#ifdef WIN32 -#include -#else -#include -#endif - -#ifdef WIN32 -void __cdecl -#else -void * -#endif - events(void *p) -{ - Yaz_SocketManager *s = (Yaz_SocketManager *) p; - - logf (LOG_LOG, "thread started"); - while (s->processEvent() > 0) - ; - logf (LOG_LOG, "thread finished"); -#ifdef WIN32 -#else - return 0; -#endif -} - -void Yaz_PDU_Assoc::childNotify(int fd) +void Yaz_PDU_Assoc::childNotify(COMSTACK cs) { - Yaz_SocketManager *socket_observable = new Yaz_SocketManager; - Yaz_PDU_Assoc *new_observable = new Yaz_PDU_Assoc (socket_observable); + Yaz_PDU_Assoc *new_observable = + new Yaz_PDU_Assoc (m_socketObservable, cs); - /// Clone PDU Observer - IYaz_PDU_Observer *observer = m_PDU_Observer->clone(new_observable, fd); - - /// Attach new socket to it - new_observable->socket(observer, fd); - -#ifdef WIN32 - long t_id; - t_id = _beginthread (events, 0, socket_observable); - if (t_id == -1) - { - logf (LOG_FATAL|LOG_ERRNO, "_beginthread failed"); - exit (1); - } -#else - pthread_t type; - - int id = pthread_create (&type, 0, events, socket_observable); - logf (LOG_LOG, "pthread_create returned id=%d", id); -#endif + // Clone PDU Observer + new_observable->m_PDU_Observer = m_PDU_Observer->sessionNotify + (new_observable, cs_fileno(cs)); } -// Threads end -#endif diff --git a/src/yaz-proxy.cpp b/src/yaz-proxy.cpp index e0ec329..988eb91 100644 --- a/src/yaz-proxy.cpp +++ b/src/yaz-proxy.cpp @@ -1,9 +1,12 @@ /* - * Copyright (c) 1998-2000, Index Data. + * Copyright (c) 1998-2001, Index Data. * See the file LICENSE for details. * * $Log: yaz-proxy.cpp,v $ - * Revision 1.22 2000-11-20 11:27:33 adam + * Revision 1.23 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.22 2000/11/20 11:27:33 adam * Fixes for connect operation (timeout and notify fix). * * Revision 1.21 2000/11/01 14:22:59 adam @@ -113,8 +116,8 @@ void Yaz_Proxy::set_proxyTarget(const char *target) m_proxyTarget = (char *) xstrdup (target); } -IYaz_PDU_Observer *Yaz_Proxy::clone(IYaz_PDU_Observable - *the_PDU_Observable, int fd) +IYaz_PDU_Observer *Yaz_Proxy::sessionNotify(IYaz_PDU_Observable + *the_PDU_Observable, int fd) { Yaz_Proxy *new_proxy = new Yaz_Proxy(the_PDU_Observable); new_proxy->m_parent = this; @@ -486,8 +489,8 @@ void Yaz_ProxyClient::connectNotify() timeout(600); } -IYaz_PDU_Observer *Yaz_ProxyClient::clone(IYaz_PDU_Observable - *the_PDU_Observable, int fd) +IYaz_PDU_Observer *Yaz_ProxyClient::sessionNotify(IYaz_PDU_Observable + *the_PDU_Observable, int fd) { return new Yaz_ProxyClient(the_PDU_Observable); } diff --git a/src/yaz-server.cpp b/src/yaz-server.cpp index cc503dc..f64a6b4 100644 --- a/src/yaz-server.cpp +++ b/src/yaz-server.cpp @@ -1,9 +1,12 @@ /* - * Copyright (c) 1998-2000, Index Data. + * Copyright (c) 1998-2001, Index Data. * See the file LICENSE for details. * * $Log: yaz-server.cpp,v $ - * Revision 1.14 2000-11-01 14:22:59 adam + * Revision 1.15 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.14 2000/11/01 14:22:59 adam * Added fd parameter for method IYaz_PDU_Observer::clone. * * Revision 1.13 2000/10/11 11:58:16 adam @@ -72,8 +75,8 @@ public: Z_RecordComposition *comp, Z_NamePlusRecord *namePlusRecord, Z_Records *records); - IYaz_PDU_Observer* clone(IYaz_PDU_Observable *the_PDU_Observable, - int fd); + IYaz_PDU_Observer* sessionNotify(IYaz_PDU_Observable *the_PDU_Observable, + int fd); void failNotify(); void timeoutNotify(); void connectNotify(); @@ -1557,7 +1560,7 @@ static char *marc_records[] = { void MyServer::recv_Z_init (Z_InitRequest *initRequest, Z_InitResponse *initResponse) { - logf (LOG_LOG, "MyServer::recv_Z_init"); + yaz_log (LOG_LOG, "MyServer::recv_Z_init"); } static MyServer *myServer = 0; @@ -1565,7 +1568,7 @@ static MyServer *myServer = 0; void MyServer::recv_Z_search (Z_SearchRequest *searchRequest, Z_SearchResponse *searchResponse) { - logf (LOG_LOG, "MyServer::recv_Z_search"); + yaz_log (LOG_LOG, "MyServer::recv_Z_search"); if (searchRequest->query->which == Z_Query_type_1) { Z_RPNStructure *s = searchRequest->query->u.type_1->RPNStructure; @@ -1586,7 +1589,7 @@ void MyServer::recv_Z_search (Z_SearchRequest *searchRequest, void MyServer::recv_Z_present (Z_PresentRequest *presentRequest, Z_PresentResponse *presentResponse) { - logf (LOG_LOG, "MyServer::recv_Z_present"); + yaz_log (LOG_LOG, "MyServer::recv_Z_present"); } void MyServer::recv_Z_record (const char *resultSetName, @@ -1596,7 +1599,7 @@ void MyServer::recv_Z_record (const char *resultSetName, Z_NamePlusRecord *namePlusRecord, Z_Records *records) { - logf (LOG_LOG, "MyServer::recv_Z_record"); + yaz_log (LOG_LOG, "MyServer::recv_Z_record"); int max = sizeof(marc_records) / sizeof(*marc_records); int eff_pos = (position-1) % max; create_databaseRecord (namePlusRecord, 0, VAL_USMARC, @@ -1604,11 +1607,10 @@ void MyServer::recv_Z_record (const char *resultSetName, strlen(marc_records[eff_pos])); } -IYaz_PDU_Observer *MyServer::clone(IYaz_PDU_Observable *the_PDU_Observable, - int fd) +IYaz_PDU_Observer *MyServer::sessionNotify( + IYaz_PDU_Observable *the_PDU_Observable, int fd) { MyServer *new_server; - logf (LOG_LOG, "child no %d", m_no); m_no++; new_server = new MyServer(the_PDU_Observable); new_server->timeout(900); @@ -1623,13 +1625,13 @@ MyServer::MyServer(IYaz_PDU_Observable *the_PDU_Observable) : void MyServer::timeoutNotify() { - logf (LOG_LOG, "connection timed out"); + yaz_log (LOG_LOG, "connection timed out"); delete this; } void MyServer::failNotify() { - logf (LOG_LOG, "connection closed by client"); + yaz_log (LOG_LOG, "connection closed by client"); delete this; } @@ -1642,7 +1644,7 @@ int main(int argc, char **argv) while (1) { Yaz_SocketManager mySocketManager; - Yaz_PDU_Assoc *my_PDU_Assoc = new Yaz_PDU_Assoc(&mySocketManager); + Yaz_PDU_AssocThread *my_PDU_Assoc = new Yaz_PDU_AssocThread(&mySocketManager); myServer = new MyServer(my_PDU_Assoc); diff --git a/src/yaz-socket-manager.cpp b/src/yaz-socket-manager.cpp index a64b9bc..669597a 100644 --- a/src/yaz-socket-manager.cpp +++ b/src/yaz-socket-manager.cpp @@ -3,7 +3,10 @@ * See the file LICENSE for details. * * $Log: yaz-socket-manager.cpp,v $ - * Revision 1.14 2000-11-20 14:17:36 adam + * Revision 1.15 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.14 2000/11/20 14:17:36 adam * Yet another WIN32 fix for connect notify. * * Revision 1.13 2000/11/20 11:27:33 adam @@ -143,7 +146,7 @@ int Yaz_SocketManager::processEvent() YazSocketEntry *p; YazSocketEvent *event = getEvent(); unsigned timeout = 0; - logf (m_log, "processEvent"); + logf (m_log, "Yaz_SocketManager::processEvent manager=%p", this); if (event) { event->observer->socketNotify(event->event); @@ -168,17 +171,17 @@ int Yaz_SocketManager::processEvent() no++; if (p->mask & YAZ_SOCKET_OBSERVE_READ) { - yaz_log (m_log, "select fd=%d: read fd", fd); + yaz_log (m_log, "Yaz_SocketManager::select fd=%d read", fd); FD_SET(fd, &in); } if (p->mask & YAZ_SOCKET_OBSERVE_WRITE) { - yaz_log (m_log, "select fd=%d: write fd", fd); + yaz_log (m_log, "Yaz_SocketManager::select fd=%d write", fd); FD_SET(fd, &out); } if (p->mask & YAZ_SOCKET_OBSERVE_EXCEPT) { - yaz_log (m_log, "select fd=%d: except fd", fd); + yaz_log (m_log, "Yaz_SocketManager::select fd=%d except", fd); FD_SET(fd, &except); } if (fd > max) @@ -209,7 +212,7 @@ int Yaz_SocketManager::processEvent() to.tv_sec = timeout; to.tv_usec = 0; - logf (m_log, "select pending=%d timeout=%d", no, timeout); + logf (m_log, "Yaz_SocketManager::select no=%d timeout=%d", no, timeout); while ((res = select(max + 1, &in, &out, &except, timeout ? &to : 0)) < 0) if (errno != EINTR) return -1; diff --git a/src/yaz-z-assoc.cpp b/src/yaz-z-assoc.cpp index c8f529b..5034821 100644 --- a/src/yaz-z-assoc.cpp +++ b/src/yaz-z-assoc.cpp @@ -3,7 +3,10 @@ * See the file LICENSE for details. * * $Log: yaz-z-assoc.cpp,v $ - * Revision 1.17 2001-01-04 14:25:25 heikki + * Revision 1.18 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.17 2001/01/04 14:25:25 heikki * No SIGPIPES in windows... * * Revision 1.16 2000/12/14 16:00:39 adam @@ -141,6 +144,10 @@ void Yaz_Z_Assoc::recv_PDU(const char *buf, int len) { recv_Z_PDU (apdu); } + else + { + close(); + } } Z_APDU *Yaz_Z_Assoc::create_Z_PDU(int type) diff --git a/src/yaz-z-server.cpp b/src/yaz-z-server.cpp index 096cfa0..b6b0a10 100644 --- a/src/yaz-z-server.cpp +++ b/src/yaz-z-server.cpp @@ -1,9 +1,12 @@ /* - * Copyright (c) 2000, Index Data. + * Copyright (c) 2000-2001, Index Data. * See the file LICENSE for details. * * $Log: yaz-z-server.cpp,v $ - * Revision 1.6 2001-01-29 11:18:24 adam + * Revision 1.7 2001-03-26 14:43:49 adam + * New threaded PDU association. + * + * Revision 1.6 2001/01/29 11:18:24 adam * Server sets OPTIONS search and present. * * Revision 1.5 2000/10/24 12:29:57 adam @@ -287,7 +290,7 @@ void Yaz_Z_Server::recv_Z_PDU (Z_APDU *apdu_request) switch (apdu_request->which) { case Z_APDU_initRequest: - logf (LOG_LOG, "got InitRequest p=%p", this); + yaz_log (LOG_LOG, "got InitRequest p=%p", this); apdu_response = create_Z_PDU(Z_APDU_initResponse); req = apdu_request->u.initRequest->options; res = apdu_response->u.initResponse->options; @@ -305,7 +308,7 @@ void Yaz_Z_Server::recv_Z_PDU (Z_APDU *apdu_request) send_Z_PDU(apdu_response); break; case Z_APDU_searchRequest: - logf (LOG_LOG, "got SearchRequest p=%p", this); + yaz_log (LOG_LOG, "got SearchRequest p=%p", this); apdu_response = create_Z_PDU(Z_APDU_searchResponse); recv_Z_search (apdu_request->u.searchRequest, apdu_response->u.searchResponse); @@ -317,7 +320,7 @@ void Yaz_Z_Server::recv_Z_PDU (Z_APDU *apdu_request) send_Z_PDU(apdu_response); break; case Z_APDU_presentRequest: - logf (LOG_LOG, "got PresentRequest p=%p", this); + yaz_log (LOG_LOG, "got PresentRequest p=%p", this); apdu_response = create_Z_PDU(Z_APDU_presentResponse); recv_Z_present (apdu_request->u.presentRequest, apdu_response->u.presentResponse); diff --git a/yaz++-config.in b/yaz++-config.in index 385ca00..1ff5a31 100644 --- a/yaz++-config.in +++ b/yaz++-config.in @@ -1,5 +1,5 @@ #!/bin/sh -# $Id: yaz++-config.in,v 1.2 2000-10-11 12:21:00 adam Exp $ +# $Id: yaz++-config.in,v 1.3 2001-03-26 14:43:49 adam Exp $ yazppprefix=@prefix@ yaz_echo_cflags=no yaz_echo_libs=no @@ -8,6 +8,7 @@ yaz_echo_tabs=no yaz_echo_source=yes yaz_echo_lalibs=no yazpp_src_root=@YAZPP_SRC_ROOT@ +yazpp_build_root=@YAZPP_BUILD_ROOT@ yazlibs="@YAZLIB@ @LIBS@" YAZPPVERSION=@VERSION@ @@ -68,8 +69,8 @@ while test $# -gt 0; do done if test "$yaz_echo_source" = "yes"; then - YAZPPLIB="-L${yazpp_src_root}/src/.libs -lyaz++ $yazlibs" - YAZPPLALIB="${yazpp_src_root}/src/libyaz++.la $yazlibs" + YAZPPLIB="-L${yazpp_build_root}/src/.libs -lyaz++ $yazlibs" + YAZPPLALIB="${yazpp_build_root}/src/libyaz++.la $yazlibs" YAZPPINC="@YAZINC@ -I${yazpp_src_root}/include" else diff --git a/yaz.m4 b/yaz.m4 index 32bfdb7..f4e8e6c 100644 --- a/yaz.m4 +++ b/yaz.m4 @@ -1,4 +1,4 @@ -## $Id: yaz.m4,v 1.1 2001-02-21 11:25:07 adam Exp $ +## $Id: yaz.m4,v 1.2 2001-03-26 14:43:49 adam Exp $ ## # Use this m4 funciton for autoconf if you use YAZ in your own # configure script. @@ -16,7 +16,12 @@ AC_DEFUN([YAZ_INIT], if test "x$yazpath" != "xNONE"; then yazconfig=$yazpath/yaz-config else - for i in ../yaz* ../yaz; do + if test "x$srcdir" = "x"; then + yazsrcdir=. + else + yazsrcdir=$srcdir + fi + for i in ${yazsrcdir}/../yaz* ${yazsrcdir}/../yaz; do if test -d $i; then if test -r $i/yaz-config; then yazconfig=$i/yaz-config -- 1.7.10.4