1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
|
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
#include <ruby.h>
#include <sys/event.h>
#include <sys/time.h>
#include <errno.h>
#ifdef HAVE_RB_WAIT_FOR_SINGLE_FD
#include <ruby/io.h>
#endif
static VALUE mGod;
static VALUE cKQueueHandler;
static VALUE cEventHandler;
static ID proc_exit;
static ID proc_fork;
static ID m_call;
static ID m_size;
static ID m_deregister;
static int kq;
int num_events;
#define NUM_EVENTS FIX2INT(rb_funcall(rb_cv_get(cEventHandler, "@@actions"), m_size, 0))
VALUE
kqh_event_mask(VALUE klass, VALUE sym)
{
ID id = SYM2ID(sym);
if (proc_exit == id) {
return UINT2NUM(NOTE_EXIT);
} else if (proc_fork == id) {
return UINT2NUM(NOTE_FORK);
} else {
rb_raise(rb_eNotImpError, "Event `%s` not implemented", rb_id2name(id));
}
return Qnil;
}
VALUE
kqh_monitor_process(VALUE klass, VALUE pid, VALUE mask)
{
struct kevent new_event;
ID event;
(void)event; //!< Silence warning about unused var, should be removed?
u_int fflags = NUM2UINT(mask);
EV_SET(&new_event, FIX2UINT(pid), EVFILT_PROC,
EV_ADD | EV_ENABLE, fflags, 0, 0);
if (-1 == kevent(kq, &new_event, 1, NULL, 0, NULL)) {
rb_raise(rb_eStandardError, "%s", strerror(errno));
}
num_events = NUM_EVENTS;
return Qnil;
}
VALUE
kqh_handle_events()
{
int nevents, i, num_to_fetch;
struct kevent *events;
#ifdef HAVE_RB_WAIT_FOR_SINGLE_FD
rb_wait_for_single_fd(kq, RB_WAITFD_IN, NULL);
#else
fd_set read_set;
FD_ZERO(&read_set);
FD_SET(kq, &read_set);
// Don't actually run this method until we've got an event
rb_thread_select(kq + 1, &read_set, NULL, NULL, NULL);
#endif
// Grabbing num_events once for thread safety
num_to_fetch = num_events;
events = (struct kevent*)malloc(num_to_fetch * sizeof(struct kevent));
if (NULL == events) {
rb_raise(rb_eStandardError, "%s", strerror(errno));
}
nevents = kevent(kq, NULL, 0, events, num_to_fetch, NULL);
if (-1 == nevents) {
free(events);
rb_raise(rb_eStandardError, "%s", strerror(errno));
} else {
for (i = 0; i < nevents; i++) {
if (events[i].fflags & NOTE_EXIT) {
rb_funcall(cEventHandler, m_call, 2, INT2NUM(events[i].ident), ID2SYM(proc_exit));
} else if (events[i].fflags & NOTE_FORK) {
rb_funcall(cEventHandler, m_call, 2, INT2NUM(events[i].ident), ID2SYM(proc_fork));
}
}
}
free(events);
return INT2FIX(nevents);
}
void
Init_kqueue_handler_ext()
{
kq = kqueue();
if (kq == -1) {
rb_raise(rb_eStandardError, "kqueue initilization failed");
}
proc_exit = rb_intern("proc_exit");
proc_fork = rb_intern("proc_fork");
m_call = rb_intern("call");
m_size = rb_intern("size");
m_deregister = rb_intern("deregister");
mGod = rb_const_get(rb_cObject, rb_intern("God"));
cEventHandler = rb_const_get(mGod, rb_intern("EventHandler"));
cKQueueHandler = rb_define_class_under(mGod, "KQueueHandler", rb_cObject);
rb_define_singleton_method(cKQueueHandler, "monitor_process", kqh_monitor_process, 2);
rb_define_singleton_method(cKQueueHandler, "handle_events", kqh_handle_events, 0);
rb_define_singleton_method(cKQueueHandler, "event_mask", kqh_event_mask, 1);
}
#endif
|
