Pimpl SocketManager class
[yazpp-moved-to-github.git] / 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