X-Git-Url: http://git.indexdata.com/?a=blobdiff_plain;f=src%2Fyaz-pdu-assoc.cpp;h=566f9b0bf251efea1e679f1f333e8e06b272be0a;hb=7a9cf0a84f1169b52469f736be7e98e7440eeff4;hp=1c009b445b978adf11a009d694d06294494dadf5;hpb=56d92bbfde918ba7c662588fabde1331210f003b;p=yazpp-moved-to-github.git 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