pazpar2  1.13.0
eventl.c
Go to the documentation of this file.
1 /* This file is part of Pazpar2.
2  Copyright (C) Index Data
3 
4  Pazpar2 is free software; you can redistribute it and/or modify it under
5  the terms of the GNU General Public License as published by the Free
6  Software Foundation; either version 2, or (at your option) any later
7  version.
8 
9  Pazpar2 is distributed in the hope that it will be useful, but WITHOUT ANY
10  WARRANTY; without even the implied warranty of MERCHANTABILITY or
11  FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12  for more details.
13 
14  You should have received a copy of the GNU General Public License
15  along with this program; if not, write to the Free Software
16  Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
17 
18  */
19 
20 /*
21  * Based on ParaZ - a simple tool for harvesting performance data for
22  * parallel operations using Z39.50.
23  * Copyright (C) Index Data ApS
24  * See LICENSE file for details.
25  */
26 
27 /*
28  * Based on revision YAZ' server/eventl.c 1.29.
29  */
30 
31 #if HAVE_CONFIG_H
32 #include <config.h>
33 #endif
34 
35 #include <math.h>
36 #include <stdio.h>
37 #include <assert.h>
38 
39 #if HAVE_UNISTD_H
40 #include <unistd.h>
41 #endif
42 
43 #include <stdlib.h>
44 #include <errno.h>
45 #include <string.h>
46 
47 #include <yaz/yconfig.h>
48 #include <yaz/log.h>
49 #include <yaz/comstack.h>
50 #include <yaz/xmalloc.h>
51 #include <yaz/mutex.h>
52 #include <yaz/poll.h>
53 #if HAVE_GETRLIMIT
54 #include <sys/resource.h>
55 #endif
56 #include "eventl.h"
57 #include "sel_thread.h"
58 
59 #if 1
60 static YAZ_MUTEX g_mutex = 0;
61 static int no_iochans = 0;
62 static int no_iochans_total = 0;
63 
64 static int iochan_use(int delta)
65 {
66  int iochans;
67  if (!g_mutex)
68  yaz_mutex_create(&g_mutex);
69  yaz_mutex_enter(g_mutex);
70  no_iochans += delta;
71  if (delta > 0)
72  no_iochans_total += delta;
73  iochans = no_iochans;
74  yaz_mutex_leave(g_mutex);
75  yaz_log(YLOG_DEBUG, "%s iochans=%d",
76  delta == 0 ? "" : (delta > 0 ? "INC" : "DEC"), iochans);
77  return iochans;
78 }
79 
80 int iochans_count(void)
81 {
82  return iochan_use(0);
83 }
84 
86 {
87  int total = 0;
88  if (!g_mutex)
89  return 0;
90  yaz_mutex_enter(g_mutex);
91  total = no_iochans_total;
92  yaz_mutex_leave(g_mutex);
93  return total;
94 }
95 #else
96 #define iochan_use(x)
97 #define iochans_count(x) 0
98 #define iochans_count_total(x) 0
99 #endif
100 
101 struct iochan_man_s {
104  int sel_fd;
107  YAZ_MUTEX iochan_mutex;
108  int size_fds;
109  int limit_fd;
110  struct yaz_poll_fd *fds;
111 };
112 
114 {
115  iochan_man_t man = xmalloc(sizeof(*man));
116  man->channel_list = 0;
117  man->sel_thread = 0; /* can't create sel_thread yet because we may fork */
118  man->sel_fd = -1;
119  man->no_threads = no_threads;
120  man->log_level = yaz_log_module_level("iochan");
121  man->iochan_mutex = 0;
122  man->size_fds = 0;
123  man->fds = 0;
124  man->limit_fd = 0;
125 #if HAVE_GETRLIMIT
126  {
127  struct rlimit limit_data;
128  getrlimit(RLIMIT_NOFILE, &limit_data);
129  yaz_log(YLOG_LOG, "getrlimit NOFILE cur=%ld max=%ld",
130  (long) limit_data.rlim_cur, (long) limit_data.rlim_max);
131  man->limit_fd = limit_data.rlim_cur - 200;
132  }
133 #endif
134  if (max_sockets)
135  man->limit_fd = max_sockets;
136  yaz_log(YLOG_LOG, "iochan threads %d limit fd %d", no_threads,
137  man->limit_fd);
138  yaz_mutex_create(&man->iochan_mutex);
139  return man;
140 }
141 
143 {
144  IOCHAN next = chan->next;
145  if (chan->name)
146  xfree(chan->name);
147  xfree(chan);
148  iochan_use(-1);
149  return next;
150 }
151 
153 {
154  if (*mp)
155  {
156  IOCHAN c;
157  if ((*mp)->sel_thread)
158  sel_thread_destroy((*mp)->sel_thread);
159 
160  yaz_mutex_enter((*mp)->iochan_mutex);
161  c = (*mp)->channel_list;
162  (*mp)->channel_list = NULL;
163  yaz_mutex_leave((*mp)->iochan_mutex);
164  while (c)
165  c = iochan_destroy_real(c);
166  yaz_mutex_destroy(&(*mp)->iochan_mutex);
167  xfree((*mp)->fds);
168  xfree(*mp);
169  *mp = 0;
170  }
171 }
172 
173 int iochan_add(iochan_man_t man, IOCHAN chan, int slack)
174 {
175  int r = 0, no_fds = 0;
176  IOCHAN p;
177 
178  yaz_log(man->log_level, "iochan_add : chan=%p channel list=%p", chan,
179  man->channel_list);
180  yaz_mutex_enter(man->iochan_mutex);
181  for (p = man->channel_list; p; p = p->next)
182  {
183  if (p->fd >= 0)
184  no_fds++;
185  }
186  if (slack >= 0 && man->limit_fd > 0 && no_fds >= man->limit_fd - slack)
187  {
188  r = -1;
189  yaz_log(YLOG_WARN, "max channels %d in use", no_fds);
190  }
191  else
192  {
193  chan->man = man;
194  chan->next = man->channel_list;
195  man->channel_list = chan;
196  }
197  yaz_mutex_leave(man->iochan_mutex);
198  return r;
199 }
200 
202 {
203  if (chan->man)
204  chan->destroyed = 1;
205  else
206  iochan_destroy_real(chan);
207 }
208 
209 IOCHAN iochan_create(int fd, IOC_CALLBACK cb, int flags, const char *name)
210 {
211  IOCHAN new_iochan;
212 
213  if (!(new_iochan = (IOCHAN) xmalloc(sizeof(*new_iochan))))
214  return 0;
215  iochan_use(1);
216  new_iochan->destroyed = 0;
217  new_iochan->fd = fd;
218  new_iochan->flags = flags;
219  new_iochan->fun = cb;
220  new_iochan->last_event = new_iochan->max_idle = 0;
221  new_iochan->next = NULL;
222  new_iochan->man = 0;
223  new_iochan->thread_users = 0;
224  new_iochan->name = name ? xstrdup(name) : 0;
225  return new_iochan;
226 }
227 
228 static void work_handler(void *work_data)
229 {
230  IOCHAN p = work_data;
231 
232  yaz_log(p->man->log_level, "eventl: work begin chan=%p name=%s event=%d",
233  p, p->name ? p->name : "", p->this_event);
234 
235  if (!p->destroyed && (p->this_event & EVENT_TIMEOUT))
236  (*p->fun)(p, EVENT_TIMEOUT);
237  if (!p->destroyed && (p->this_event & EVENT_INPUT))
238  (*p->fun)(p, EVENT_INPUT);
239  if (!p->destroyed && (p->this_event & EVENT_OUTPUT))
240  (*p->fun)(p, EVENT_OUTPUT);
241  if (!p->destroyed && (p->this_event & EVENT_EXCEPT))
242  (*p->fun)(p, EVENT_EXCEPT);
243 
244  yaz_log(p->man->log_level, "eventl: work end chan=%p name=%s event=%d", p,
245  p->name ? p->name : "", p->this_event);
246 }
247 
248 static void run_fun(iochan_man_t man, IOCHAN p)
249 {
250  if (man->sel_thread)
251  {
252  yaz_log(man->log_level,
253  "eventl: work add chan=%p name=%s event=%d", p,
254  p->name ? p->name : "", p->this_event);
255  p->thread_users++;
256  sel_thread_add(man->sel_thread, p);
257  }
258  else
259  work_handler(p);
260 }
261 
262 static int event_loop(iochan_man_t man, IOCHAN *iochans)
263 {
264  do /* loop as long as there are active associations to process */
265  {
266  IOCHAN p, *nextp;
267  IOCHAN start;
268  IOCHAN inv_start;
269  int res;
270  struct yaz_poll_fd *fds;
271  int i, no_fds = 0;
272  int connection_fired = 0;
273  int tv_sec = 300;
274 
275  yaz_mutex_enter(man->iochan_mutex);
276  start = man->channel_list;
277  yaz_mutex_leave(man->iochan_mutex);
278  inv_start = start;
279  for (p = start; p; p = p->next)
280  no_fds++;
281  if (man->sel_fd != -1)
282  no_fds++;
283  if (no_fds > man->size_fds)
284  {
285  man->size_fds = no_fds * 2;
286  man->fds = xrealloc(man->fds, man->size_fds * sizeof(*man->fds));
287  }
288  fds = man->fds;
289  i = 0;
290  if (man->sel_fd != -1)
291  {
292  fds[i].fd = man->sel_fd;
293  fds[i].input_mask = yaz_poll_read;
294  fds[i].client_data = 0;
295  i++;
296  }
297  for (p = start; p; p = p->next, i++)
298  {
299  p->poll_offset = i;
300  fds[i].client_data = p;
301  fds[i].fd = -1;
302  fds[i].input_mask = 0;
303  if (p->thread_users > 0)
304  continue;
305  if (p->max_idle && p->max_idle < tv_sec)
306  tv_sec = p->max_idle;
307  if (p->fd < 0)
308  continue;
309  if (p->flags & EVENT_INPUT)
310  fds[i].input_mask |= yaz_poll_read;
311  if (p->flags & EVENT_OUTPUT)
312  fds[i].input_mask |= yaz_poll_write;
313  if (p->flags & EVENT_EXCEPT)
314  fds[i].input_mask |= yaz_poll_except;
315  if (fds[i].input_mask)
316  fds[i].fd = p->fd;
317  }
318  assert(i == no_fds);
319  yaz_log(man->log_level, "yaz_poll begin tv_sec=%d nofds=%d", tv_sec,
320  no_fds);
321  res = yaz_poll(fds, no_fds, tv_sec, 0);
322  yaz_log(man->log_level, "yaz_poll returned res=%d", res);
323  if (res < 0)
324  {
325  if (errno == EINTR)
326  continue;
327  else
328  {
329  yaz_log(YLOG_ERRNO | YLOG_WARN, "poll");
330  abort();
331  }
332  }
333  if (man->sel_fd != -1)
334  {
335  i = 0;
336  assert(fds[i].fd == man->sel_fd);
337  if (fds[i].output_mask)
338  {
339  IOCHAN chan;
340 
341  yaz_log(man->log_level, "eventl: sel input on sel_fd=%d",
342  man->sel_fd);
343  while ((chan = sel_thread_result(man->sel_thread)))
344  {
345  yaz_log(man->log_level,
346  "eventl: got thread result chan=%p name=%s", chan,
347  chan->name ? chan->name : "");
348  chan->thread_users--;
349  }
350  }
351  }
352  if (man->log_level)
353  {
354  int no = 0;
355  for (p = start; p; p = p->next)
356  no++;
357  yaz_log(man->log_level, "%d channels", no);
358  }
359  for (p = start; p; p = p->next)
360  {
361  time_t now = time(0);
362  i = p->poll_offset;
363 
364  if (p->destroyed)
365  {
366  yaz_log(man->log_level,
367  "eventl: skip destroyed chan=%p name=%s", p,
368  p->name ? p->name : "");
369  continue;
370  }
371  if (p->thread_users > 0)
372  {
373  yaz_log(man->log_level,
374  "eventl: skip chan=%p name=%s users=%d", p,
375  p->name ? p->name : "", p->thread_users);
376  continue;
377  }
378  p->this_event = 0;
379  if (p->max_idle && now - p->last_event > p->max_idle)
380  {
381  p->last_event = now;
383  }
384  if (fds[i].fd >= 0 && p->fd == fds[i].fd)
385  {
386  if (fds[i].output_mask & yaz_poll_read)
387  {
388  p->last_event = now;
389  p->this_event |= EVENT_INPUT;
390  }
391  if (fds[i].output_mask & yaz_poll_write)
392  {
393  p->last_event = now;
394  p->this_event |= EVENT_OUTPUT;
395  }
396  if (fds[i].output_mask & yaz_poll_except)
397  {
398  p->last_event = now;
399  p->this_event |= EVENT_EXCEPT;
400  }
401  }
402  /* only fire one Z39.50/SRU socket event.. except for timeout */
403  if (p->this_event)
404  {
405  if (!(p->this_event & EVENT_TIMEOUT) &&
406  !strcmp(p->name, "connection_socket"))
407  {
408  /* not a timeout and we have a Z39.50/SRU socket */
409  if (connection_fired == 0)
410  run_fun(man, p);
411  connection_fired++;
412  }
413  else
414  run_fun(man, p);
415  }
416  }
417 
418  assert(inv_start == start);
419  yaz_mutex_enter(man->iochan_mutex);
420  for (nextp = iochans; *nextp; )
421  {
422  IOCHAN p = *nextp;
423  if (p->destroyed && p->thread_users == 0)
424  *nextp = iochan_destroy_real(p);
425  else
426  nextp = &p->next;
427  }
428  yaz_mutex_leave(man->iochan_mutex);
429  } while (*iochans);
430  return 0;
431 }
432 
434 {
435  if (man->no_threads > 0 && !man->sel_thread)
436  {
437  man->sel_thread = sel_thread_create(work_handler, 0 /*work_destroy */,
438  &man->sel_fd, man->no_threads);
439  yaz_log(man->log_level, "iochan_man_events. Using %d threads",
440  man->no_threads);
441  }
442  event_loop(man, &man->channel_list);
443 }
444 
445 /*
446  * Local variables:
447  * c-basic-offset: 4
448  * c-file-style: "Stroustrup"
449  * indent-tabs-mode: nil
450  * End:
451  * vim: shiftwidth=4 tabstop=8 expandtab
452  */
453 
#define EVENT_INPUT
Definition: eventl.h:35
int iochans_count_total(void)
Definition: eventl.c:85
iochan_man_t man
Definition: eventl.h:47
void sel_thread_destroy(sel_thread_t p)
destorys select thread
Definition: sel_thread.c:182
static int no_iochans_total
Definition: eventl.c:62
void iochan_destroy(IOCHAN chan)
Definition: eventl.c:201
struct yaz_poll_fd * fds
Definition: eventl.c:110
#define EVENT_OUTPUT
Definition: eventl.h:36
IOCHAN channel_list
Definition: eventl.c:102
IOCHAN iochan_create(int fd, IOC_CALLBACK cb, int flags, const char *name)
Definition: eventl.c:209
time_t max_idle
Definition: eventl.h:43
int poll_offset
Definition: eventl.h:50
char * name
Definition: eventl.h:48
IOC_CALLBACK fun
Definition: eventl.h:39
static YAZ_MUTEX g_mutex
Definition: eventl.c:60
int this_event
Definition: eventl.h:44
#define EVENT_EXCEPT
Definition: eventl.h:37
int size_fds
Definition: eventl.c:108
char * name
static int event_loop(iochan_man_t man, IOCHAN *iochans)
Definition: eventl.c:262
void iochan_man_events(iochan_man_t man)
Definition: eventl.c:433
static int no_iochans
Definition: eventl.c:61
int iochan_add(iochan_man_t man, IOCHAN chan, int slack)
Definition: eventl.c:173
int limit_fd
Definition: eventl.c:109
struct iochan * next
Definition: eventl.h:49
iochan_man_t iochan_man_create(int no_threads, int max_sockets)
Definition: eventl.c:113
static void run_fun(iochan_man_t man, IOCHAN p)
Definition: eventl.c:248
sel_thread_t sel_thread
Definition: eventl.c:103
int fd
Definition: eventl.h:33
int flags
Definition: eventl.h:34
void sel_thread_add(sel_thread_t p, void *data)
adds work to be carried out in thread
Definition: sel_thread.c:205
int no_threads
Definition: eventl.c:105
#define EVENT_TIMEOUT
Definition: eventl.h:38
void(* IOC_CALLBACK)(struct iochan *i, int event)
Definition: eventl.h:27
sel_thread_t sel_thread_create(void(*work_handler)(void *work_data), void(*work_destroy)(void *work_data), int *read_fd, int no_of_threads)
creates select thread
Definition: sel_thread.c:129
void * sel_thread_result(sel_thread_t p)
gets result of work
Definition: sel_thread.c:230
Definition: eventl.h:31
int thread_users
Definition: eventl.h:45
void iochan_man_destroy(iochan_man_t *mp)
Definition: eventl.c:152
IOCHAN iochan_destroy_real(IOCHAN chan)
Definition: eventl.c:142
static int iochan_use(int delta)
Definition: eventl.c:64
time_t last_event
Definition: eventl.h:42
YAZ_MUTEX iochan_mutex
Definition: eventl.c:107
int sel_fd
Definition: eventl.c:104
int iochans_count(void)
Definition: eventl.c:80
int log_level
Definition: eventl.c:106
int destroyed
Definition: eventl.h:41
static void work_handler(void *work_data)
Definition: eventl.c:228