src/yaz-socket-manager.cpp

   1 /* This file is part of the yazpp toolkit.
   2  * Copyright (C) 1998-2012 Index Data and Mike Taylor
   3  * See the file LICENSE for details.
   4  */
   5
   6 #if HAVE_CONFIG_H
   7 #include <config.h>
   8 #endif
   9 #if HAVE_SYS_TIME_H
  10 #include <sys/time.h>
  11 #endif
  12 #if HAVE_SYS_TYPES_H
  13 #include <sys/types.h>
  14 #endif
  15 #if HAVE_UNISTD_H
  16 #include <unistd.h>
  17 #endif
  18
  19 #include <errno.h>
  20 #include <string.h>
  21 #include <assert.h>
  22 #include <stdlib.h>
  23 #include <time.h>
  24
  25 #include <yaz/log.h>
  26
  27 #include <yazpp/socket-manager.h>
  28 #include <yaz/poll.h>
  29
  30 using namespace yazpp_1;
  31
  32 struct SocketManager::SocketEntry {
  33     ISocketObserver *observer;
  34     int fd;
  35     unsigned mask;
  36     int timeout;
  37     int timeout_this;
  38     time_t last_activity;
  39     SocketEntry *next;
  40 };
  41
  42 struct SocketManager::SocketEvent {
  43     ISocketObserver *observer;
  44     int event;
  45     SocketEvent *next;          // front in queue
  46     SocketEvent *prev;          // back in queue
  47 };
  48
  49 struct SocketManager::Rep {
  50     SocketEntry *observers;       // all registered observers
  51     SocketEvent *queue_front;
  52     SocketEvent *queue_back;
  53     int log;
  54 };
  55
  56 SocketManager::SocketEntry **SocketManager::lookupObserver(
  57     ISocketObserver *observer)
  58 {
  59     SocketEntry **se;
  60
  61     for (se = &m_p->observers; *se; se = &(*se)->next)
  62         if ((*se)->observer == observer)
  63             break;
  64     return se;
  65 }
  66
  67 int SocketManager::getNumberOfObservers()
  68 {
  69     int i = 0;
  70     SocketEntry *se;
  71     for (se = m_p->observers; se; se = se->next, i++)
  72         ;
  73     return i;
  74 }
  75
  76 void SocketManager::addObserver(int fd, ISocketObserver *observer)
  77 {
  78     SocketEntry *se;
  79
  80     se = *lookupObserver(observer);
  81     if (!se)
  82     {
  83         se = new SocketEntry;
  84         se->next= m_p->observers;
  85         m_p->observers = se;
  86         se->observer = observer;
  87     }
  88     se->fd = fd;
  89     se->mask = 0;
  90     se->last_activity = 0;
  91     se->timeout = -1;
  92 }
  93
  94 void SocketManager::deleteObserver(ISocketObserver *observer)
  95 {
  96     SocketEntry **se = lookupObserver(observer);
  97     if (*se)
  98     {
  99         removeEvent(observer);
 100         SocketEntry *se_tmp = *se;
 101         *se = (*se)->next;
 102         delete se_tmp;
 103     }
 104 }
 105
 106 void SocketManager::deleteObservers()
 107 {
 108     SocketEntry *se = m_p->observers;
 109
 110     while (se)
 111     {
 112         SocketEntry *se_next = se->next;
 113         delete se;
 114         se = se_next;
 115     }
 116     m_p->observers = 0;
 117 }
 118
 119 void SocketManager::maskObserver(ISocketObserver *observer, int mask)
 120 {
 121     SocketEntry *se;
 122
 123     yaz_log(m_p->log, "obs=%p read=%d write=%d except=%d", observer,
 124                     mask & SOCKET_OBSERVE_READ,
 125                     mask & SOCKET_OBSERVE_WRITE,
 126                     mask & SOCKET_OBSERVE_EXCEPT);
 127
 128     se = *lookupObserver(observer);
 129     if (se)
 130         se->mask = mask;
 131 }
 132
 133 void SocketManager::timeoutObserver(ISocketObserver *observer,
 134                                         int timeout)
 135 {
 136     SocketEntry *se;
 137
 138     se = *lookupObserver(observer);
 139     if (se)
 140         se->timeout = timeout;
 141 }
 142
 143 void SocketManager::inspect_poll_result(int res, struct yaz_poll_fd *fds,
 144                                         int no_fds, int timeout)
 145
 146 {
 147     yaz_log(m_p->log, "yaz_poll returned res=%d", res);
 148     time_t now = time(0);
 149     int i;
 150     int no_put_events = 0;
 151     int no_lost_observers = 0;
 152
 153     for (i = 0; i < no_fds; i++)
 154     {
 155         SocketEntry *p;
 156         for (p = m_p->observers; p; p = p->next)
 157             if (p->fd == fds[i].fd)
 158                 break;
 159         if (!p)
 160         {
 161             // m_p->observers list changed since poll started
 162             no_lost_observers++;
 163             continue;
 164         }
 165
 166         enum yaz_poll_mask output_mask = fds[i].output_mask;
 167
 168         int mask = 0;
 169         if (output_mask & yaz_poll_read)
 170             mask |= SOCKET_OBSERVE_READ;
 171
 172         if (output_mask & yaz_poll_write)
 173             mask |= SOCKET_OBSERVE_WRITE;
 174
 175         if (output_mask & yaz_poll_except)
 176             mask |= SOCKET_OBSERVE_EXCEPT;
 177
 178         if (mask)
 179         {
 180             SocketEvent *event = new SocketEvent;
 181             p->last_activity = now;
 182             event->observer = p->observer;
 183             event->event = mask;
 184             putEvent(event);
 185             no_put_events++;
 186             yaz_log(m_p->log, "putEvent I/O mask=%d", mask);
 187         }
 188         else if (res == 0 && p->timeout_this == timeout)
 189         {
 190             SocketEvent *event = new SocketEvent;
 191             assert(p->last_activity);
 192             yaz_log(m_p->log, "putEvent timeout fd=%d, now = %ld "
 193                     "last_activity=%ld timeout=%d",
 194                     p->fd, now, p->last_activity, p->timeout);
 195             p->last_activity = now;
 196             event->observer = p->observer;
 197             event->event = SOCKET_OBSERVE_TIMEOUT;
 198             putEvent(event);
 199             no_put_events++;
 200
 201         }
 202     }
 203     SocketEvent *event = getEvent();
 204     if (event)
 205     {
 206         event->observer->socketNotify(event->event);
 207         delete event;
 208     }
 209     else
 210     {
 211         if (no_lost_observers == 0)
 212         {
 213             // bug #2035
 214             yaz_log(YLOG_WARN, "unhandled socket event. yaz_poll returned %d",
 215                     res);
 216             yaz_log(YLOG_WARN, "no_put_events=%d no_fds=%d i=%d timeout=%d",
 217                     no_put_events, no_fds, i, timeout);
 218         }
 219     }
 220 }
 221
 222 int SocketManager::processEvent()
 223 {
 224     SocketEntry *p;
 225     SocketEvent *event = getEvent();
 226     int timeout = -1;
 227     yaz_log(m_p->log, "SocketManager::processEvent manager=%p", this);
 228     if (event)
 229     {
 230         event->observer->socketNotify(event->event);
 231         delete event;
 232         return 1;
 233     }
 234
 235     int res;
 236     time_t now = time(0);
 237     int i;
 238     int no_fds = 0;
 239     for (p = m_p->observers; p; p = p->next)
 240         no_fds++;
 241
 242     if (!no_fds)
 243         return 0;
 244     struct yaz_poll_fd *fds = new yaz_poll_fd [no_fds];
 245     for (i = 0, p = m_p->observers; p; p = p->next, i++)
 246     {
 247         fds[i].fd = p->fd;
 248         int input_mask = 0;
 249         if (p->mask & SOCKET_OBSERVE_READ)
 250             input_mask += yaz_poll_read;
 251         if (p->mask & SOCKET_OBSERVE_WRITE)
 252             input_mask += yaz_poll_write;
 253         if (p->mask & SOCKET_OBSERVE_EXCEPT)
 254             input_mask += yaz_poll_except;
 255         if (p->timeout > 0 ||
 256             (p->timeout == 0 && (p->mask & SOCKET_OBSERVE_WRITE) == 0))
 257         {
 258             int timeout_this;
 259             timeout_this = p->timeout;
 260             if (p->last_activity)
 261                 timeout_this -= now - p->last_activity;
 262             else
 263                 p->last_activity = now;
 264             if (timeout_this < 0 || timeout_this > 2147483646)
 265                 timeout_this = 0;
 266             if (timeout == -1 || timeout_this < timeout)
 267                 timeout = timeout_this;
 268             p->timeout_this = timeout_this;
 269             yaz_log(m_p->log, "SocketManager::select timeout_this=%d",
 270                     p->timeout_this);
 271         }
 272         else
 273             p->timeout_this = -1;
 274         fds[i].input_mask = (enum yaz_poll_mask) input_mask;
 275     }
 276
 277     int pass = 0;
 278     while ((res = yaz_poll(fds, no_fds, timeout, 0)) < 0 && pass < 10)
 279     {
 280         if (errno == EINTR)
 281         {
 282             delete [] fds;
 283             return 1;
 284         }
 285         yaz_log(YLOG_ERRNO|YLOG_WARN, "yaz_poll");
 286         yaz_log(YLOG_WARN, "errno=%d timeout=%d", errno, timeout);
 287     }
 288
 289     if (res >= 0)
 290         inspect_poll_result(res, fds, no_fds, timeout);
 291
 292     delete [] fds;
 293     return res >= 0 ? 1 : -1;
 294 }
 295
 296 //    n p    n p  ......   n p    n p
 297 //   front                        back
 298
 299 void SocketManager::putEvent(SocketEvent *event)
 300 {
 301     // put in back of queue
 302     if (m_p->queue_back)
 303     {
 304         m_p->queue_back->prev = event;
 305         assert(m_p->queue_front);
 306     }
 307     else
 308     {
 309         assert(!m_p->queue_front);
 310         m_p->queue_front = event;
 311     }
 312     event->next = m_p->queue_back;
 313     event->prev = 0;
 314     m_p->queue_back = event;
 315 }
 316
 317 SocketManager::SocketEvent *SocketManager::getEvent()
 318 {
 319     // get from front of queue
 320     SocketEvent *event = m_p->queue_front;
 321     if (!event)
 322         return 0;
 323     assert(m_p->queue_back);
 324     m_p->queue_front = event->prev;
 325     if (m_p->queue_front)
 326     {
 327         assert(m_p->queue_back);
 328         m_p->queue_front->next = 0;
 329     }
 330     else
 331         m_p->queue_back = 0;
 332     return event;
 333 }
 334
 335 void SocketManager::removeEvent(ISocketObserver *observer)
 336 {
 337     SocketEvent *ev = m_p->queue_back;
 338     while (ev)
 339     {
 340         SocketEvent *ev_next = ev->next;
 341         if (observer == ev->observer)
 342         {
 343             if (ev->prev)
 344                 ev->prev->next = ev->next;
 345             else
 346                 m_p->queue_back = ev->next;
 347             if (ev->next)
 348                 ev->next->prev = ev->prev;
 349             else
 350                 m_p->queue_front = ev->prev;
 351             delete ev;
 352         }
 353         ev = ev_next;
 354     }
 355 }
 356
 357 SocketManager::SocketManager()
 358 {
 359     m_p = new Rep;
 360     m_p->observers = 0;
 361     m_p->queue_front = 0;
 362     m_p->queue_back = 0;
 363     m_p->log = YLOG_DEBUG;
 364 }
 365
 366 SocketManager::~SocketManager()
 367 {
 368     deleteObservers();
 369     delete m_p;
 370 }
 371 /*
 372  * Local variables:
 373  * c-basic-offset: 4
 374  * c-file-style: "Stroustrup"
 375  * indent-tabs-mode: nil
 376  * End:
 377  * vim: shiftwidth=4 tabstop=8 expandtab
 378  */
 379