src/common/fdwatch_kqueue.c

/*
  * Abstraction API/layer for the various ways PvPGN can inspect sockets state
  * 2003 (C) dizzy@roedu.net
  *
  * Code is based on the ideas found in thttpd project.
  *
  * *BSD kqueue(2) based backend
  *
  * This program is free software; you can redistribute it and/or
  * modify it under the terms of the GNU General Public License
  * as published by the Free Software Foundation; either version 2
  * of the License, or (at your option) any later version.
  *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
  *
  * You should have received a copy of the GNU General Public License
  * along with this program; if not, write to the Free Software
  * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
  */

#include "common/setup_before.h"
#ifdef STDC_HEADERS
# include <stdlib.h>
#else
# ifdef HAVE_MALLOC_H
#  include <malloc.h>
# endif
#endif
#ifdef HAVE_STRING_H
# include <string.h>
#else
# ifdef HAVE_STRINGS_H
#  include <strings.h>
# endif
#endif
#ifdef HAVE_SYS_TYPES_H
# include <sys/types.h>
#endif
#ifdef HAVE_SYS_EVENT_H
# include <sys/event.h>
#endif
#ifdef HAVE_SYS_TIME_H
# include <sys/time.h>
#endif
#include "fdwatch.h"
#include "common/eventlog.h"
#include "common/xalloc.h"
#include "common/setup_after.h"

#ifdef HAVE_KQUEUE
static int sr;
static int kq;
static struct kevent *kqchanges = NULL;         /* changes to make to kqueue */
static struct kevent *kqevents = NULL;          /* events to investigate */
/* r/w indices from idx to the kqchanges index where the change is stored */
static int *_rridx, *_wridx;
static unsigned nochanges;

static int fdw_kqueue_init(int nfds);
static int fdw_kqueue_close(void);
static int fdw_kqueue_add_fd(int idx, t_fdwatch_type rw);
static int fdw_kqueue_del_fd(int idx);
static int fdw_kqueue_watch(long timeout_msecs);
static void fdw_kqueue_handle(void);

t_fdw_backend fdw_kqueue = {
    fdw_kqueue_init,
    fdw_kqueue_close,
    fdw_kqueue_add_fd,
    fdw_kqueue_del_fd,
    fdw_kqueue_watch,
    fdw_kqueue_handle
};

static int fdw_kqueue_init(int nfds)
{
    int i;

    if ((kq = kqueue()) == -1)
        return -1;
    kqevents = (struct kevent *) xmalloc(sizeof(struct kevent) * nfds);
    kqchanges = (struct kevent *) xmalloc(sizeof(struct kevent) * nfds * 2);
    _rridx = (int *) xmalloc(sizeof(int) * nfds);
    _wridx = (int *) xmalloc(sizeof(int) * nfds);

    memset(kqchanges, 0, sizeof(struct kevent) * nfds);
    for (i = 0; i < nfds; i++)
    {
        _rridx[i] = -1;
        _wridx[i] = -1;
    }
    sr = 0;
    nochanges = 0;

    eventlog(eventlog_level_info, __FUNCTION__, "fdwatch kqueue() based layer initialized (max %d sockets)", nfds);
    return 0;
}

static int fdw_kqueue_close(void)
{
    if (_rridx) { xfree((void *) _rridx); _rridx = NULL; }
    if (_wridx) { xfree((void *) _wridx); _wridx = NULL; }
    if (kqchanges) { xfree((void *) kqchanges); kqchanges = NULL; }
    if (kqevents) { xfree((void *) kqevents); kqevents = NULL; }
    sr = 0;
    nochanges = 0;

    return 0;
}

static int fdw_kqueue_add_fd(int idx, t_fdwatch_type rw)
{
    static int ridx;
    t_fdwatch_fd *cfd;

/*    eventlog(eventlog_level_trace, __FUNCTION__, "called fd: %d rw: %d", fd, rw); */

    cfd = fdw_fds + idx;
    /* adding read event filter */
    if (!(fdw_rw(cfd) & fdwatch_type_read) && rw & fdwatch_type_read)
    {
        if (_rridx[idx] >= 0 && _rridx[idx] < nochanges && kqchanges[_rridx[idx]].ident == fdw_fd(cfd))
        {
            ridx = _rridx[idx];
/*          eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (read) fd on %d", ridx); */
        } else {
            ridx = nochanges++;
            _rridx[idx] = ridx;
/*          eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (read) fd on %d", ridx); */
        }
        EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_READ, EV_ADD, 0, 0, (void*)idx);
    } 
    else if (fdw_rw(cfd) & fdwatch_type_read && !( rw & fdwatch_type_read))
    {
        if (_rridx[idx] >= 0 && _rridx[idx] < nochanges && kqchanges[_rridx[idx]].ident == fdw_fd(cfd))
        {
            ridx = _rridx[idx];
/*          eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (read) fd on %d", ridx); */
        } else {
            ridx = nochanges++;
            _rridx[idx] = ridx;
/*          eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (read) fd on %d", ridx); */
        }
        EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_READ, EV_DELETE, 0, 0, (void*)idx);
    }

    /* adding write event filter */
    if (!(fdw_rw(cfd) & fdwatch_type_write) && rw & fdwatch_type_write)
    {
        if (_wridx[idx] >= 0 && _wridx[idx] < nochanges && kqchanges[_wridx[idx]].ident == fdw_fd(cfd))
        {
            ridx = _wridx[idx];
/*          eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (write) fd on %d", ridx); */
        } else {
            ridx = nochanges++;
            _wridx[idx] = ridx;
/*          eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (write) fd on %d", ridx); */
        }
        EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_WRITE, EV_ADD, 0, 0, (void*)idx);
    }
    else if (fdw_rw(cfd) & fdwatch_type_write && !(rw & fdwatch_type_write))
    {
        if (_wridx[idx] >= 0 && _wridx[idx] < nochanges && kqchanges[_wridx[idx]].ident == fdw_fd(cfd))
        {
            ridx = _wridx[idx];
/*          eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (write) fd on %d", ridx); */
        } else {
            ridx = nochanges++;
            _wridx[idx] = ridx;
/*          eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (write) fd on %d", ridx); */
        }
        EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_WRITE, EV_DELETE, 0, 0, (void*)idx);
    }

    return 0;
}

static int fdw_kqueue_del_fd(int idx)
{
    t_fdwatch_fd *cfd;

/*    eventlog(eventlog_level_trace, __FUNCTION__, "called fd: %d", fd); */
    if (sr > 0) 
        eventlog(eventlog_level_error, __FUNCTION__, "BUG: called while still handling sockets");

    cfd = fdw_fds + idx;
    /* the last event changes about this fd has not yet been sent to kernel */
    if (fdw_rw(cfd) & fdwatch_type_read &&
        nochanges && _rridx[idx] >= 0 && _rridx[idx] < nochanges && 
        kqchanges[_rridx[idx]].ident == fdw_fd(cfd))
    {
        nochanges--;
        if (_rridx[idx] < nochanges)
        {
            int oidx;

            oidx = (int)(kqchanges[nochanges].udata);
            if (kqchanges[nochanges].filter == EVFILT_READ && 
                _rridx[oidx] == nochanges)
            {
/*          eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
                _rridx[oidx] = _rridx[idx];
                memcpy(kqchanges + _rridx[idx], kqchanges + nochanges, sizeof(struct kevent));
            }

            if (kqchanges[nochanges].filter == EVFILT_WRITE &&
                _wridx[oidx] == nochanges)
            {
/*          eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
                _wridx[oidx] = _rridx[idx];
                memcpy(kqchanges + _rridx[idx], kqchanges + nochanges, sizeof(struct kevent));
            }
        }
        _rridx[idx] = -1;
    }

    if (fdw_rw(cfd) & fdwatch_type_write &&
        nochanges && _wridx[idx] >= 0 && _wridx[idx] < nochanges && 
        kqchanges[_wridx[idx]].ident == fdw_fd(cfd))
    {
        nochanges--;
        if (_wridx[idx] < nochanges)
        {
            int oidx;

            oidx = (int)(kqchanges[nochanges].udata);
            if (kqchanges[nochanges].filter == EVFILT_READ && 
                _rridx[oidx] == nochanges)
            {
/*          eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
                _rridx[oidx] = _wridx[idx];
                memcpy(kqchanges + _wridx[idx], kqchanges + nochanges, sizeof(struct kevent));
            }

            if (kqchanges[nochanges].filter == EVFILT_WRITE &&
                _wridx[oidx] == nochanges)
            {
/*          eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
                _wridx[oidx] = _wridx[idx];
                memcpy(kqchanges + _wridx[idx], kqchanges + nochanges, sizeof(struct kevent));
            }
        }
        _wridx[idx] = -1;
    }

/* here we presume the calling code does close() on the socket and if so
 * it is automatically removed from any kernel kqueues */

    return 0;
}

static int fdw_kqueue_watch(long timeout_msec)
{
    static struct timespec ts;

    ts.tv_sec = timeout_msec / 1000L;
    ts.tv_nsec = (timeout_msec % 1000L) * 1000000L;
    sr = kevent(kq, nochanges > 0 ? kqchanges : NULL, nochanges, kqevents, fdw_maxcons, &ts);
    nochanges = 0;
    return sr;
}

static void fdw_kqueue_handle(void)
{
    register unsigned i;
    t_fdwatch_fd *cfd;

/*    eventlog(eventlog_level_trace, __FUNCTION__, "called"); */
    for (i = 0; i < sr; i++)
    {
/*      eventlog(eventlog_level_trace, __FUNCTION__, "checking %d ident: %d read: %d write: %d", i, kqevents[i].ident, kqevents[i].filter & EVFILT_READ, kqevents[i].filter & EVFILT_WRITE); */
        cfd = fdw_fds + (int)kqevents[i].udata;
        if (fdw_rw(cfd) & fdwatch_type_read && kqevents[i].filter == EVFILT_READ)
            if (fdw_hnd(cfd) (fdw_data(cfd), fdwatch_type_read) == -2)
                continue;

        if (fdw_rw(cfd) & fdwatch_type_write && kqevents[i].filter == EVFILT_WRITE)
            fdw_hnd(cfd) (fdw_data(cfd), fdwatch_type_write);
    }
    sr = 0;
}

#endif                          /* HAVE_KQUEUE */
1	sysadm	1.1	/*
2			* Abstraction API/layer for the various ways PvPGN can inspect sockets state
3			* 2003 (C) dizzy@roedu.net
4			*
5			* Code is based on the ideas found in thttpd project.
6			*
7			* *BSD kqueue(2) based backend
8			*
9			* This program is free software; you can redistribute it and/or
10			* modify it under the terms of the GNU General Public License
11			* as published by the Free Software Foundation; either version 2
12			* of the License, or (at your option) any later version.
13			*
14			* This program is distributed in the hope that it will be useful,
15			* but WITHOUT ANY WARRANTY; without even the implied warranty of
16			* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17			* GNU General Public License for more details.
18			*
19			* You should have received a copy of the GNU General Public License
20			* along with this program; if not, write to the Free Software
21			* Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
22			*/
23
24			#include "common/setup_before.h"
25			#ifdef STDC_HEADERS
26			# include <stdlib.h>
27			#else
28			# ifdef HAVE_MALLOC_H
29			# include <malloc.h>
30			# endif
31			#endif
32			#ifdef HAVE_STRING_H
33			# include <string.h>
34			#else
35			# ifdef HAVE_STRINGS_H
36			# include <strings.h>
37			# endif
38			#endif
39			#ifdef HAVE_SYS_TYPES_H
40			# include <sys/types.h>
41			#endif
42			#ifdef HAVE_SYS_EVENT_H
43			# include <sys/event.h>
44			#endif
45			#ifdef HAVE_SYS_TIME_H
46			# include <sys/time.h>
47			#endif
48			#include "fdwatch.h"
49			#include "common/eventlog.h"
50			#include "common/xalloc.h"
51			#include "common/setup_after.h"
52
53			#ifdef HAVE_KQUEUE
54			static int sr;
55			static int kq;
56			static struct kevent kqchanges = NULL; / changes to make to kqueue */
57			static struct kevent kqevents = NULL; / events to investigate */
58			/* r/w indices from idx to the kqchanges index where the change is stored */
59			static int _rridx, _wridx;
60			static unsigned nochanges;
61
62			static int fdw_kqueue_init(int nfds);
63			static int fdw_kqueue_close(void);
64			static int fdw_kqueue_add_fd(int idx, t_fdwatch_type rw);
65			static int fdw_kqueue_del_fd(int idx);
66			static int fdw_kqueue_watch(long timeout_msecs);
67			static void fdw_kqueue_handle(void);
68
69			t_fdw_backend fdw_kqueue = {
70			fdw_kqueue_init,
71			fdw_kqueue_close,
72			fdw_kqueue_add_fd,
73			fdw_kqueue_del_fd,
74			fdw_kqueue_watch,
75			fdw_kqueue_handle
76			};
77
78			static int fdw_kqueue_init(int nfds)
79			{
80			int i;
81
82			if ((kq = kqueue()) == -1)
83			return -1;
84			kqevents = (struct kevent ) xmalloc(sizeof(struct kevent) nfds);
85			kqchanges = (struct kevent ) xmalloc(sizeof(struct kevent) nfds * 2);
86			_rridx = (int ) xmalloc(sizeof(int) nfds);
87			_wridx = (int ) xmalloc(sizeof(int) nfds);
88
89			memset(kqchanges, 0, sizeof(struct kevent) * nfds);
90			for (i = 0; i < nfds; i++)
91			{
92			_rridx[i] = -1;
93			_wridx[i] = -1;
94			}
95			sr = 0;
96			nochanges = 0;
97
98			eventlog(eventlog_level_info, __FUNCTION__, "fdwatch kqueue() based layer initialized (max %d sockets)", nfds);
99			return 0;
100			}
101
102			static int fdw_kqueue_close(void)
103			{
104			if (_rridx) { xfree((void *) _rridx); _rridx = NULL; }
105			if (_wridx) { xfree((void *) _wridx); _wridx = NULL; }
106			if (kqchanges) { xfree((void *) kqchanges); kqchanges = NULL; }
107			if (kqevents) { xfree((void *) kqevents); kqevents = NULL; }
108			sr = 0;
109			nochanges = 0;
110
111			return 0;
112			}
113
114			static int fdw_kqueue_add_fd(int idx, t_fdwatch_type rw)
115			{
116			static int ridx;
117			t_fdwatch_fd *cfd;
118
119			/* eventlog(eventlog_level_trace, __FUNCTION__, "called fd: %d rw: %d", fd, rw); */
120
121			cfd = fdw_fds + idx;
122			/* adding read event filter */
123			if (!(fdw_rw(cfd) & fdwatch_type_read) && rw & fdwatch_type_read)
124			{
125			if (_rridx[idx] >= 0 && _rridx[idx] < nochanges && kqchanges[_rridx[idx]].ident == fdw_fd(cfd))
126			{
127			ridx = _rridx[idx];
128			/* eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (read) fd on %d", ridx); */
129			} else {
130			ridx = nochanges++;
131			_rridx[idx] = ridx;
132			/* eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (read) fd on %d", ridx); */
133			}
134			EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_READ, EV_ADD, 0, 0, (void*)idx);
135			}
136			else if (fdw_rw(cfd) & fdwatch_type_read && !( rw & fdwatch_type_read))
137			{
138			if (_rridx[idx] >= 0 && _rridx[idx] < nochanges && kqchanges[_rridx[idx]].ident == fdw_fd(cfd))
139			{
140			ridx = _rridx[idx];
141			/* eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (read) fd on %d", ridx); */
142			} else {
143			ridx = nochanges++;
144			_rridx[idx] = ridx;
145			/* eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (read) fd on %d", ridx); */
146			}
147			EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_READ, EV_DELETE, 0, 0, (void*)idx);
148			}
149
150			/* adding write event filter */
151			if (!(fdw_rw(cfd) & fdwatch_type_write) && rw & fdwatch_type_write)
152			{
153			if (_wridx[idx] >= 0 && _wridx[idx] < nochanges && kqchanges[_wridx[idx]].ident == fdw_fd(cfd))
154			{
155			ridx = _wridx[idx];
156			/* eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (write) fd on %d", ridx); */
157			} else {
158			ridx = nochanges++;
159			_wridx[idx] = ridx;
160			/* eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (write) fd on %d", ridx); */
161			}
162			EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_WRITE, EV_ADD, 0, 0, (void*)idx);
163			}
164			else if (fdw_rw(cfd) & fdwatch_type_write && !(rw & fdwatch_type_write))
165			{
166			if (_wridx[idx] >= 0 && _wridx[idx] < nochanges && kqchanges[_wridx[idx]].ident == fdw_fd(cfd))
167			{
168			ridx = _wridx[idx];
169			/* eventlog(eventlog_level_trace, __FUNCTION__, "updating change event (write) fd on %d", ridx); */
170			} else {
171			ridx = nochanges++;
172			_wridx[idx] = ridx;
173			/* eventlog(eventlog_level_trace, __FUNCTION__, "adding new change event (write) fd on %d", ridx); */
174			}
175			EV_SET(kqchanges + ridx, fdw_fd(cfd), EVFILT_WRITE, EV_DELETE, 0, 0, (void*)idx);
176			}
177
178			return 0;
179			}
180
181			static int fdw_kqueue_del_fd(int idx)
182			{
183			t_fdwatch_fd *cfd;
184
185			/* eventlog(eventlog_level_trace, __FUNCTION__, "called fd: %d", fd); */
186			if (sr > 0)
187			eventlog(eventlog_level_error, __FUNCTION__, "BUG: called while still handling sockets");
188
189			cfd = fdw_fds + idx;
190			/* the last event changes about this fd has not yet been sent to kernel */
191			if (fdw_rw(cfd) & fdwatch_type_read &&
192			nochanges && _rridx[idx] >= 0 && _rridx[idx] < nochanges &&
193			kqchanges[_rridx[idx]].ident == fdw_fd(cfd))
194			{
195			nochanges--;
196			if (_rridx[idx] < nochanges)
197			{
198			int oidx;
199
200			oidx = (int)(kqchanges[nochanges].udata);
201			if (kqchanges[nochanges].filter == EVFILT_READ &&
202			_rridx[oidx] == nochanges)
203			{
204			/* eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
205			_rridx[oidx] = _rridx[idx];
206			memcpy(kqchanges + _rridx[idx], kqchanges + nochanges, sizeof(struct kevent));
207			}
208
209			if (kqchanges[nochanges].filter == EVFILT_WRITE &&
210			_wridx[oidx] == nochanges)
211			{
212			/* eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
213			_wridx[oidx] = _rridx[idx];
214			memcpy(kqchanges + _rridx[idx], kqchanges + nochanges, sizeof(struct kevent));
215			}
216			}
217			_rridx[idx] = -1;
218			}
219
220			if (fdw_rw(cfd) & fdwatch_type_write &&
221			nochanges && _wridx[idx] >= 0 && _wridx[idx] < nochanges &&
222			kqchanges[_wridx[idx]].ident == fdw_fd(cfd))
223			{
224			nochanges--;
225			if (_wridx[idx] < nochanges)
226			{
227			int oidx;
228
229			oidx = (int)(kqchanges[nochanges].udata);
230			if (kqchanges[nochanges].filter == EVFILT_READ &&
231			_rridx[oidx] == nochanges)
232			{
233			/* eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
234			_rridx[oidx] = _wridx[idx];
235			memcpy(kqchanges + _wridx[idx], kqchanges + nochanges, sizeof(struct kevent));
236			}
237
238			if (kqchanges[nochanges].filter == EVFILT_WRITE &&
239			_wridx[oidx] == nochanges)
240			{
241			/* eventlog(eventlog_level_trace, __FUNCTION__, "not last, moving %d", kqchanges[rnfds].ident); */
242			_wridx[oidx] = _wridx[idx];
243			memcpy(kqchanges + _wridx[idx], kqchanges + nochanges, sizeof(struct kevent));
244			}
245			}
246			_wridx[idx] = -1;
247			}
248
249			/* here we presume the calling code does close() on the socket and if so
250			* it is automatically removed from any kernel kqueues */
251
252			return 0;
253			}
254
255			static int fdw_kqueue_watch(long timeout_msec)
256			{
257			static struct timespec ts;
258
259			ts.tv_sec = timeout_msec / 1000L;
260			ts.tv_nsec = (timeout_msec % 1000L) * 1000000L;
261			sr = kevent(kq, nochanges > 0 ? kqchanges : NULL, nochanges, kqevents, fdw_maxcons, &ts);
262			nochanges = 0;
263			return sr;
264			}
265
266			static void fdw_kqueue_handle(void)
267			{
268			register unsigned i;
269			t_fdwatch_fd *cfd;
270
271			/* eventlog(eventlog_level_trace, __FUNCTION__, "called"); */
272			for (i = 0; i < sr; i++)
273			{
274			/* eventlog(eventlog_level_trace, __FUNCTION__, "checking %d ident: %d read: %d write: %d", i, kqevents[i].ident, kqevents[i].filter & EVFILT_READ, kqevents[i].filter & EVFILT_WRITE); */
275			cfd = fdw_fds + (int)kqevents[i].udata;
276			if (fdw_rw(cfd) & fdwatch_type_read && kqevents[i].filter == EVFILT_READ)
277			if (fdw_hnd(cfd) (fdw_data(cfd), fdwatch_type_read) == -2)
278			continue;
279
280			if (fdw_rw(cfd) & fdwatch_type_write && kqevents[i].filter == EVFILT_WRITE)
281			fdw_hnd(cfd) (fdw_data(cfd), fdwatch_type_write);
282			}
283			sr = 0;
284			}
285
286			#endif /* HAVE_KQUEUE */
webmaster@leafok.com	ViewVC Help
Powered by ViewVC 1.3.0-beta1