ModKqueue.cc
Go to the documentation of this file.
1 /*
2  * Copyright (C) 1996-2017 The Squid Software Foundation and contributors
3  *
4  * Squid software is distributed under GPLv2+ license and includes
5  * contributions from numerous individuals and organizations.
6  * Please see the COPYING and CONTRIBUTORS files for details.
7  */
8 
9 /* DEBUG: section 05 Socket Functions */
10 
11 /*
12  * This code was originally written by Benno Rice and hacked on quite
13  * a bit by Adrian. Adrian then took it to the hybrid-ircd project to use
14  * in their new IO subsystem. After a year of modifications and some
15  * rather interesting changes (event aggregation) its back in squid.
16  * Thanks to the ircd-hybrid guys.
17  */
18 
19 /*
20  * XXX Currently not implemented / supported by this module XXX
21  *
22  * - delay pools
23  * - deferred reads
24  * - flags.read_pending
25  *
26  * So, its not entirely useful in a production setup since if a read
27  * is meant to be deferred it isn't (we're not even throwing the event
28  * away here). Eventually the rest of the code will be rewritten
29  * so deferred reads aren't required.
30  * -- adrian
31  */
32 #include "squid.h"
33 
34 #if USE_KQUEUE
35 #include "comm/Loops.h"
36 #include "fde.h"
37 #include "globals.h"
38 #include "SquidTime.h"
39 #include "StatCounters.h"
40 #include "Store.h"
41 
42 #include <cerrno>
43 #if HAVE_SYS_EVENT_H
44 #include <sys/event.h>
45 #endif
46 
47 #define KE_LENGTH 128
48 
49 /* jlemon goofed up and didn't add EV_SET until fbsd 4.3 */
50 
51 #ifndef EV_SET
52 #define EV_SET(kevp, a, b, c, d, e, f) do { \
53  (kevp)->ident = (a); \
54  (kevp)->filter = (b); \
55  (kevp)->flags = (c); \
56  (kevp)->fflags = (d); \
57  (kevp)->data = (e); \
58  (kevp)->udata = (f); \
59 } while(0)
60 #endif
61 
62 static void kq_update_events(int, short, PF *);
63 static int kq;
64 
65 static struct timespec zero_timespec;
66 
67 static struct kevent *kqlst; /* kevent buffer */
68 static int kqmax; /* max structs to buffer */
69 static int kqoff; /* offset into the buffer */
70 static int max_poll_time = 1000;
71 
72 static void commKQueueRegisterWithCacheManager(void);
73 
74 /* XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX */
75 /* Private functions */
76 
77 void
78 kq_update_events(int fd, short filter, PF * handler)
79 {
80  PF *cur_handler;
81  int kep_flags;
82 
83  switch (filter) {
84 
85  case EVFILT_READ:
86  cur_handler = fd_table[fd].read_handler;
87  break;
88 
89  case EVFILT_WRITE:
90  cur_handler = fd_table[fd].write_handler;
91  break;
92 
93  default:
94  /* XXX bad! -- adrian */
95  return;
96  break;
97  }
98 
99  if ((cur_handler == NULL && handler != NULL)
100  || (cur_handler != NULL && handler == NULL)) {
101 
102  struct kevent *kep;
103 
104  kep = kqlst + kqoff;
105 
106  if (handler != NULL) {
107  kep_flags = (EV_ADD | EV_ONESHOT);
108  } else {
109  kep_flags = EV_DELETE;
110  }
111 
112  EV_SET(kep, (uintptr_t) fd, filter, kep_flags, 0, 0, 0);
113 
114  /* Check if we've used the last one. If we have then submit them all */
115  if (kqoff == kqmax - 1) {
116  int ret;
117 
118  ret = kevent(kq, kqlst, kqmax, NULL, 0, &zero_timespec);
119  /* jdc -- someone needs to do error checking... */
120 
121  if (ret == -1) {
122  perror("kq_update_events(): kevent()");
123  return;
124  }
125 
126  kqoff = 0;
127  } else {
128  ++kqoff;
129  }
130  }
131 }
132 
133 /* XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX */
134 /* Public functions */
135 
136 /*
137  * comm_select_init
138  *
139  * This is a needed exported function which will be called to initialise
140  * the network loop code.
141  */
142 void
144 {
145  kq = kqueue();
146 
147  if (kq < 0) {
148  fatal("comm_select_init: Couldn't open kqueue fd!\n");
149  }
150 
151  kqmax = getdtablesize();
152 
153  kqlst = (struct kevent *)xmalloc(sizeof(*kqlst) * kqmax);
154  zero_timespec.tv_sec = 0;
155  zero_timespec.tv_nsec = 0;
156 
158 }
159 
160 /*
161  * comm_setselect
162  *
163  * This is a needed exported function which will be called to register
164  * and deregister interest in a pending IO state for a given FD.
165  */
166 void
167 Comm::SetSelect(int fd, unsigned int type, PF * handler, void *client_data, time_t timeout)
168 {
169  fde *F = &fd_table[fd];
170  assert(fd >= 0);
171  assert(F->flags.open);
172  debugs(5, 5, HERE << "FD " << fd << ", type=" << type <<
173  ", handler=" << handler << ", client_data=" << client_data <<
174  ", timeout=" << timeout);
175 
176  if (type & COMM_SELECT_READ) {
177  if (F->flags.read_pending)
178  kq_update_events(fd, EVFILT_WRITE, handler);
179 
180  kq_update_events(fd, EVFILT_READ, handler);
181 
182  F->read_handler = handler;
183  F->read_data = client_data;
184  }
185 
186  if (type & COMM_SELECT_WRITE) {
187  kq_update_events(fd, EVFILT_WRITE, handler);
188  F->write_handler = handler;
189  F->write_data = client_data;
190  }
191 
192  if (timeout)
193  F->timeout = squid_curtime + timeout;
194 
195 }
196 
197 void
198 Comm::ResetSelect(int fd)
199 {
200  fde *F = &fd_table[fd];
201  if (F->read_handler) {
202  kq_update_events(fd, EVFILT_READ, (PF *)1);
203  }
204  if (F->write_handler) {
205  kq_update_events(fd, EVFILT_WRITE, (PF *)1);
206  }
207 }
208 
209 /*
210  * Check all connections for new connections and input data that is to be
211  * processed. Also check for connections with data queued and whether we can
212  * write it out.
213  */
214 
215 /*
216  * comm_select
217  *
218  * Called to do the new-style IO, courtesy of of squid (like most of this
219  * new IO code). This routine handles the stuff we've hidden in
220  * comm_setselect and fd_table[] and calls callbacks for IO ready
221  * events.
222  */
223 
225 Comm::DoSelect(int msec)
226 {
227  int num, i;
228 
229  static struct kevent ke[KE_LENGTH];
230 
231  struct timespec poll_time;
232 
233  if (msec > max_poll_time)
234  msec = max_poll_time;
235 
236  poll_time.tv_sec = msec / 1000;
237 
238  poll_time.tv_nsec = (msec % 1000) * 1000000;
239 
240  for (;;) {
241  num = kevent(kq, kqlst, kqoff, ke, KE_LENGTH, &poll_time);
243  kqoff = 0;
244 
245  if (num >= 0)
246  break;
247 
248  if (ignoreErrno(errno))
249  break;
250 
251  getCurrentTime();
252 
253  return Comm::COMM_ERROR;
254 
255  /* NOTREACHED */
256  }
257 
258  getCurrentTime();
259 
260  if (num == 0)
261  return Comm::OK; /* No error.. */
262 
263  for (i = 0; i < num; ++i) {
264  int fd = (int) ke[i].ident;
265  PF *hdl = NULL;
266  fde *F = &fd_table[fd];
267 
268  if (ke[i].flags & EV_ERROR) {
269  errno = ke[i].data;
270  /* XXX error == bad! -- adrian */
271  continue; /* XXX! */
272  }
273 
274  if (ke[i].filter == EVFILT_READ || F->flags.read_pending) {
275  if ((hdl = F->read_handler) != NULL) {
276  F->read_handler = NULL;
277  F->flags.read_pending = 0;
278  hdl(fd, F->read_data);
279  }
280  }
281 
282  if (ke[i].filter == EVFILT_WRITE) {
283  if ((hdl = F->write_handler) != NULL) {
284  F->write_handler = NULL;
285  hdl(fd, F->write_data);
286  }
287  }
288 
289  if (ke[i].filter != EVFILT_WRITE && ke[i].filter != EVFILT_READ) {
290  /* Bad! -- adrian */
291  debugs(5, DBG_IMPORTANT, "comm_select: kevent returned " << ke[i].filter << "!");
292  }
293  }
294 
295  return Comm::OK;
296 }
297 
298 void
300 {
301  max_poll_time = 10;
302 }
303 
304 static void
306 {
307 }
308 
309 #endif /* USE_KQUEUE */
310 
#define fd_table
Definition: fde.h:157
StatCounters statCounter
Definition: StatCounters.cc:12
#define assert(EX)
Definition: assert.h:17
void ResetSelect(int)
reset/undo/unregister the watch for an FD which was set by Comm::SetSelect()
Definition: ModDevPoll.cc:304
void SelectLoopInit(void)
Initialize the module on Squid startup.
Definition: ModDevPoll.cc:178
struct fde::_fde_flags flags
static int kqmax
Definition: ModKqueue.cc:68
int i
Definition: membanger.c:49
static struct kevent * kqlst
Definition: ModKqueue.cc:67
Definition: Flag.h:16
#define EV_SET(kevp, a, b, c, d, e, f)
Definition: ModKqueue.cc:52
static struct timespec zero_timespec
Definition: ModKqueue.cc:65
PF * write_handler
Definition: fde.h:128
time_t squid_curtime
Definition: stub_time.cc:17
static int kq
Definition: ModKqueue.cc:63
static void commKQueueRegisterWithCacheManager(void)
Definition: ModKqueue.cc:305
Comm::Flag DoSelect(int)
Do poll and trigger callback functions as appropriate.
Definition: ModDevPoll.cc:324
#define debugs(SECTION, LEVEL, CONTENT)
Definition: Debug.h:123
PF * read_handler
Definition: fde.h:126
#define DBG_IMPORTANT
Definition: Debug.h:45
void SetSelect(int, unsigned int, PF *, void *, time_t)
Mark an FD to be watched for its IO status.
Definition: ModDevPoll.cc:225
#define KE_LENGTH
Definition: ModKqueue.cc:47
static int max_poll_time
Definition: ModKqueue.cc:70
void * write_data
Definition: fde.h:129
void fatal(const char *message)
Definition: fatal.cc:39
void PF(int, void *)
Definition: forward.h:18
void * read_data
Definition: fde.h:127
static int kqoff
Definition: ModKqueue.cc:69
unsigned long int select_loops
Definition: StatCounters.h:118
static void handler(int signo)
Definition: purge.cc:860
static void kq_update_events(int, short, PF *)
Definition: ModKqueue.cc:78
std::ostream & HERE(std::ostream &s)
Definition: Debug.h:147
Flag
Definition: Flag.h:15
bool SIGHDLR int STUB void int
Definition: stub_tools.cc:68
int ignoreErrno(int ierrno)
Definition: comm.cc:1477
time_t getCurrentTime(void)
Get current time.
bool read_pending
Definition: fde.h:106
#define xmalloc
bool open
Definition: fde.h:96
Definition: fde.h:49
time_t timeout
Definition: fde.h:131
#define NULL
Definition: types.h:166
void QuickPollRequired(void)
Definition: ModDevPoll.cc:444

 

Introduction

Documentation

Support

Miscellaneous

Web Site Translations

Mirrors