X-Git-Url: http://git.indexdata.com/?a=blobdiff_plain;f=src%2Fyaz-socket-manager.cpp;h=c78c529ef27c81cd3a3d8f1b6e5b668134a85383;hb=805b4fdf7b70b0e9b298064f24ceff0c1d38e05b;hp=8f209ac109a0f9df03181c24b0f88f531e608ab1;hpb=97118338f9ac93e767e5589d449d3f9abacb3190;p=yazpp-moved-to-github.git diff --git a/src/yaz-socket-manager.cpp b/src/yaz-socket-manager.cpp index 8f209ac..c78c529 100644 --- a/src/yaz-socket-manager.cpp +++ b/src/yaz-socket-manager.cpp @@ -1,244 +1,385 @@ -/* - * Copyright (c) 1998-1999, Index Data. +/* This file is part of the yazpp toolkit. + * Copyright (C) 1998-2013 Index Data and Mike Taylor * See the file LICENSE for details. - * Sebastian Hammer, Adam Dickmeiss - * - * $Log: yaz-socket-manager.cpp,v $ - * Revision 1.2 1999-01-28 13:08:48 adam - * Yaz_PDU_Assoc better encapsulated. Memory leak fix in - * yaz-socket-manager.cc. - * - * Revision 1.1.1.1 1999/01/28 09:41:07 adam - * First implementation of YAZ++. - * */ -#include -#ifdef WINDOWS -#include -#else + +#if HAVE_CONFIG_H +#include +#endif +#if HAVE_SYS_TIME_H #include +#endif +#if HAVE_SYS_TYPES_H #include +#endif +#if HAVE_UNISTD_H #include #endif + #include +#include +#include +#include +#include + +#include + +#include +#include + +using namespace yazpp_1; + +struct SocketManager::SocketEntry { + ISocketObserver *observer; + int fd; + unsigned mask; + int timeout; + int timeout_this; + time_t last_activity; + SocketEntry *next; +}; -#include -#include +struct SocketManager::SocketEvent { + ISocketObserver *observer; + int event; + SocketEvent *next; // front in queue + SocketEvent *prev; // back in queue +}; +struct SocketManager::Rep { + void putEvent(SocketEvent *event); + SocketEvent *getEvent(); + void removeEvent(ISocketObserver *observer); + void inspect_poll_result(int res, struct yaz_poll_fd *fds, int no_fds, + int timeout); + SocketEntry **lookupObserver(ISocketObserver *observer); + SocketEntry *observers; // all registered observers + SocketEvent *queue_front; + SocketEvent *queue_back; + int log; +}; -Yaz_SocketManager::YazSocketEntry **Yaz_SocketManager::lookupObserver( - IYazSocketObserver *observer) +SocketManager::SocketEntry **SocketManager::Rep::lookupObserver( + ISocketObserver *observer) { - YazSocketEntry **se; - - for (se = &m_observers; *se; se = &(*se)->next) - if ((*se)->observer == observer) - break; + SocketEntry **se; + + for (se = &observers; *se; se = &(*se)->next) + if ((*se)->observer == observer) + break; return se; } -void Yaz_SocketManager::addObserver(int fd, IYazSocketObserver *observer) +int SocketManager::getNumberOfObservers() +{ + int i = 0; + SocketEntry *se; + for (se = m_p->observers; se; se = se->next, i++) + ; + return i; +} + +void SocketManager::addObserver(int fd, ISocketObserver *observer) { - YazSocketEntry *se; + SocketEntry *se; - se = *lookupObserver(observer); + se = *m_p->lookupObserver(observer); if (!se) { - se = new YazSocketEntry; - se->next= m_observers; - m_observers = se; - se->observer = observer; + se = new SocketEntry; + se->next= m_p->observers; + m_p->observers = se; + se->observer = observer; } se->fd = fd; se->mask = 0; + se->last_activity = 0; + se->timeout = -1; } -void Yaz_SocketManager::deleteObserver(IYazSocketObserver *observer) +void SocketManager::deleteObserver(ISocketObserver *observer) { - YazSocketEntry **se = lookupObserver(observer); + SocketEntry **se = m_p->lookupObserver(observer); if (*se) { - removeEvent (observer); - YazSocketEntry *se_tmp = *se; - *se = (*se)->next; - delete se_tmp; + m_p->removeEvent(observer); + SocketEntry *se_tmp = *se; + *se = (*se)->next; + delete se_tmp; } } -void Yaz_SocketManager::deleteObservers() +void SocketManager::deleteObservers() { - YazSocketEntry *se = m_observers; - + SocketEntry *se = m_p->observers; + while (se) { - YazSocketEntry *se_next = se->next; - delete se; - se = se_next; + SocketEntry *se_next = se->next; + delete se; + se = se_next; } - m_observers = 0; + m_p->observers = 0; } -void Yaz_SocketManager::maskObserver(IYazSocketObserver *observer, int mask) +void SocketManager::maskObserver(ISocketObserver *observer, int mask) { - YazSocketEntry *se; + SocketEntry *se; + + yaz_log(m_p->log, "obs=%p read=%d write=%d except=%d", observer, + mask & SOCKET_OBSERVE_READ, + mask & SOCKET_OBSERVE_WRITE, + mask & SOCKET_OBSERVE_EXCEPT); - se = *lookupObserver(observer); + se = *m_p->lookupObserver(observer); if (se) - se->mask = mask; + se->mask = mask; } -void Yaz_SocketManager::timeoutObserver(IYazSocketObserver *observer, - unsigned timeout) +void SocketManager::timeoutObserver(ISocketObserver *observer, + int timeout) { - YazSocketEntry *se; + SocketEntry *se; - se = *lookupObserver(observer); + se = *m_p->lookupObserver(observer); if (se) - se->timeout = timeout; + se->timeout = timeout; } -int Yaz_SocketManager::processEvent() +void SocketManager::Rep::inspect_poll_result(int res, struct yaz_poll_fd *fds, + int no_fds, int timeout) + { - YazSocketEvent *event = getEvent(); - if (event) + yaz_log(log, "yaz_poll returned res=%d", res); + time_t now = time(0); + int i; + int no_put_events = 0; + int no_lost_observers = 0; + + for (i = 0; i < no_fds; i++) { - event->observer->socketNotify(event->event); - delete event; - return 1; - } + SocketEntry *p; + for (p = observers; p; p = p->next) + if (p->fd == fds[i].fd) + break; + if (!p) + { + // m_p->observers list changed since poll started + no_lost_observers++; + continue; + } - fd_set in, out, except; - int res; - int max = 0; - int no = 0; - struct timeval to; - struct timeval *timeout = &to; + enum yaz_poll_mask output_mask = fds[i].output_mask; - FD_ZERO(&in); - FD_ZERO(&out); - FD_ZERO(&except); + int mask = 0; + if (output_mask & yaz_poll_read) + mask |= SOCKET_OBSERVE_READ; - timeout = &to; /* hang on select */ - to.tv_sec = 60; - to.tv_usec = 0; + if (output_mask & yaz_poll_write) + mask |= SOCKET_OBSERVE_WRITE; - for (YazSocketEntry *p = m_observers; p; p = p->next) + if (output_mask & yaz_poll_except) + mask |= SOCKET_OBSERVE_EXCEPT; + + if (mask) + { + SocketEvent *event = new SocketEvent; + p->last_activity = now; + event->observer = p->observer; + event->event = mask; + putEvent(event); + no_put_events++; + yaz_log(log, "putEvent I/O mask=%d", mask); + } + else if (res == 0 && p->timeout_this == timeout) + { + SocketEvent *event = new SocketEvent; + assert(p->last_activity); + yaz_log(log, "putEvent timeout fd=%d, now = %ld " + "last_activity=%ld timeout=%d", + p->fd, now, p->last_activity, p->timeout); + p->last_activity = now; + event->observer = p->observer; + event->event = SOCKET_OBSERVE_TIMEOUT; + putEvent(event); + no_put_events++; + + } + } + SocketEvent *event = getEvent(); + if (event) + { + event->observer->socketNotify(event->event); + delete event; + } + else + { + if (no_lost_observers == 0) + { + // bug #2035 + yaz_log(YLOG_WARN, "unhandled socket event. yaz_poll returned %d", + res); + yaz_log(YLOG_WARN, "no_put_events=%d no_fds=%d i=%d timeout=%d", + no_put_events, no_fds, i, timeout); + } + } +} + +int SocketManager::processEvent() +{ + SocketEntry *p; + SocketEvent *event = m_p->getEvent(); + int timeout = -1; + yaz_log(m_p->log, "SocketManager::processEvent manager=%p", this); + if (event) { - int fd = p->fd; - logf (LOG_LOG, "fd = %d mask=%d", fd, p->mask); - if (p->mask) - no++; - if (p->mask & YAZ_SOCKET_OBSERVE_READ) - FD_SET(fd, &in); - if (p->mask & YAZ_SOCKET_OBSERVE_WRITE) - FD_SET(fd, &out); - if (p->mask & YAZ_SOCKET_OBSERVE_EXCEPT) - FD_SET(fd, &except); - if (fd > max) - max = fd; + event->observer->socketNotify(event->event); + delete event; + return 1; } - if (!no) - return 0; - while ((res = select(max + 1, &in, &out, &except, timeout)) < 0) - if (errno != EINTR) - return -1; - for (YazSocketEntry * p = m_observers; p; p = p->next) + int res; + time_t now = time(0); + int i; + int no_fds = 0; + for (p = m_p->observers; p; p = p->next) + no_fds++; + + if (!no_fds) + return 0; + struct yaz_poll_fd *fds = new yaz_poll_fd [no_fds]; + for (i = 0, p = m_p->observers; p; p = p->next, i++) { - int fd = p->fd; - int mask = 0; - if (FD_ISSET(fd, &in)) - mask |= YAZ_SOCKET_OBSERVE_READ; - - if (FD_ISSET(fd, &out)) - mask |= YAZ_SOCKET_OBSERVE_WRITE; - - if (FD_ISSET(fd, &except)) - mask |= YAZ_SOCKET_OBSERVE_EXCEPT; - - if (mask) - { - YazSocketEvent *event = new YazSocketEvent; - event->observer = p->observer; - event->event = mask; - putEvent (event); - } + fds[i].fd = p->fd; + int input_mask = 0; + if (p->mask & SOCKET_OBSERVE_READ) + input_mask += yaz_poll_read; + if (p->mask & SOCKET_OBSERVE_WRITE) + input_mask += yaz_poll_write; + if (p->mask & SOCKET_OBSERVE_EXCEPT) + input_mask += yaz_poll_except; + if (p->timeout > 0 || + (p->timeout == 0 && (p->mask & SOCKET_OBSERVE_WRITE) == 0)) + { + int timeout_this; + timeout_this = p->timeout; + if (p->last_activity) + timeout_this -= now - p->last_activity; + else + p->last_activity = now; + if (timeout_this < 0 || timeout_this > 2147483646) + timeout_this = 0; + if (timeout == -1 || timeout_this < timeout) + timeout = timeout_this; + p->timeout_this = timeout_this; + yaz_log(m_p->log, "SocketManager::select timeout_this=%d", + p->timeout_this); + } + else + p->timeout_this = -1; + fds[i].input_mask = (enum yaz_poll_mask) input_mask; } - if ((event = getEvent())) + + int pass = 0; + while ((res = yaz_poll(fds, no_fds, timeout, 0)) < 0 && pass < 10) { - event->observer->socketNotify(event->event); - delete event; - return 1; + if (errno == EINTR) + { + delete [] fds; + return 1; + } + yaz_log(YLOG_ERRNO|YLOG_WARN, "yaz_poll"); + yaz_log(YLOG_WARN, "errno=%d timeout=%d", errno, timeout); } - return 0; + + if (res >= 0) + m_p->inspect_poll_result(res, fds, no_fds, timeout); + + delete [] fds; + return res >= 0 ? 1 : -1; } -void Yaz_SocketManager::putEvent(YazSocketEvent *event) +// n p n p ...... n p n p +// front back + +void SocketManager::Rep::putEvent(SocketEvent *event) { // put in back of queue - if (m_queue_back) + if (queue_back) { - m_queue_back->prev = event; - assert (m_queue_front); + queue_back->prev = event; + assert(queue_front); } else { - assert (!m_queue_front); - m_queue_front = event; + assert(!queue_front); + queue_front = event; } - event->next = m_queue_back; + event->next = queue_back; event->prev = 0; - m_queue_back = event; + queue_back = event; } -Yaz_SocketManager::YazSocketEvent *Yaz_SocketManager::getEvent() +SocketManager::SocketEvent *SocketManager::Rep::getEvent() { // get from front of queue - YazSocketEvent *event = m_queue_front; + SocketEvent *event = queue_front; if (!event) - return 0; - assert (m_queue_back); - m_queue_front = event->prev; - if (m_queue_front) + return 0; + assert(queue_back); + queue_front = event->prev; + if (queue_front) { - assert (m_queue_back); - m_queue_front->next = 0; + assert(queue_back); + queue_front->next = 0; } else - m_queue_back = 0; + queue_back = 0; return event; } -void Yaz_SocketManager::removeEvent(IYazSocketObserver *observer) +void SocketManager::Rep::removeEvent(ISocketObserver *observer) { - YazSocketEvent *ev = m_queue_back; + SocketEvent *ev = queue_back; while (ev) { - YazSocketEvent *ev_next = ev->next; - if (observer == ev->observer) - { - if (ev->prev) - ev->prev->next = ev->next; - else - m_queue_back = ev->next; - if (ev->next) - ev->next->prev = ev->prev; - else - m_queue_front = ev->prev; - delete ev; - } - ev = ev_next; + SocketEvent *ev_next = ev->next; + if (observer == ev->observer) + { + if (ev->prev) + ev->prev->next = ev->next; + else + queue_back = ev->next; + if (ev->next) + ev->next->prev = ev->prev; + else + queue_front = ev->prev; + delete ev; + } + ev = ev_next; } } -Yaz_SocketManager::Yaz_SocketManager() +SocketManager::SocketManager() { - m_observers = 0; - m_queue_front = 0; - m_queue_back = 0; + m_p = new Rep; + m_p->observers = 0; + m_p->queue_front = 0; + m_p->queue_back = 0; + m_p->log = YLOG_DEBUG; } -Yaz_SocketManager::~Yaz_SocketManager() +SocketManager::~SocketManager() { deleteObservers(); + delete m_p; } +/* + * Local variables: + * c-basic-offset: 4 + * c-file-style: "Stroustrup" + * indent-tabs-mode: nil + * End: + * vim: shiftwidth=4 tabstop=8 expandtab + */ +