Class: Kernel::Kqueue

Inherits:
Object
  • Object
show all
Extended by:
FFI::Library
Defined in:
lib/ktools/kqueue.rb

Defined Under Namespace

Classes: Kevent, Timespec

Constant Summary collapse

KQ_FILTERS = 
{
  :read => EVFILT_READ,
  :write => EVFILT_WRITE,
  :file => EVFILT_VNODE,
  :process => EVFILT_PROC,
  :signal => EVFILT_SIGNAL
}
KQ_FLAGS = 
{
  :add => EV_ADD,
  :enable => EV_ENABLE,
  :disable => EV_DISABLE,
  :delete => EV_DELETE,
  :oneshot => EV_ONESHOT,
  :clear => EV_CLEAR
}
KQ_FFLAGS = 
{
  :delete => NOTE_DELETE,
  :write => NOTE_WRITE,
  :extend => NOTE_EXTEND,
  :attrib => NOTE_ATTRIB,
  :link => NOTE_LINK,
  :rename => NOTE_RENAME,
  :revoke => NOTE_REVOKE,
  :exit => NOTE_EXIT,
  :fork => NOTE_FORK,
  :exec => NOTE_EXEC
}

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Kqueue

Creates a new kqueue event queue. Will raise an error if the operation fails.

Raises:

  • (SystemCallError) 


119
120
121
122
123
124
 
# File 'lib/ktools/kqueue.rb', line 119

def initialize
  @fds = {}
  @pids = {}
  @kqfd = kqueue
  raise SystemCallError.new("Error creating kqueue descriptor", get_errno) unless @kqfd > 0
end

Instance Method Details

#add(type, target, options = {}) ⇒ Object

Generic method for adding events. This simply calls the proper add_foo method specified by the type symbol. Example:

kq.add(:process, pid, :events => [:fork])
calls -> kq.add_process(pid, events => [:fork])



130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/ktools/kqueue.rb', line 130

def add(type, target, options={})
  case type
  when :socket
    add_socket(target, options)
  when :file
    add_file(target, options)
  when :process
    add_process(target, options)
  when :signal
    add_signal(target, options)
  else
    raise ArgumentError.new("Unknown event type #{type}")
  end
end

#add_file(file, options = {}) ⇒ Object

Add events on a file to the Kqueue. kqueue requires that a file actually be opened (to get a descriptor) before it can be monitored. You can pass a File object here, or a String of the pathname, in which case we’ll try to open the file for you. In either case, a File object will be returned as the :target in the event returned by #poll. If you want to keep track of it yourself, you can just pass the file descriptor number (and that’s what you’ll get back.)

Valid events here are as follows, using descriptions from the kqueue man pages:

  • :delete - “The unlink() system call was called on the file referenced by the descriptor.”

  • :write - “A write occurred on the file referenced by the descriptor.”

  • :extend - “The file referenced by the descriptor was extended.”

  • :attrib - “The file referenced by the descriptor had its attributes changed.”

  • :link - “The link count on the file changed.”

  • :rename - “The file referenced by the descriptor was renamed.”

  • :revoke - “Access to the file was revoked via revoke(2) or the underlying fileystem was unmounted.”

Example:

irb(main):001:0> require 'ktools'
=> true
irb(main):002:0> file = Tempfile.new("kqueue-test")
=> #<File:/tmp/kqueue-test20090417-602-evm5wc-0>
irb(main):003:0> kq = Kqueue.new
=> #<Kernel::Kqueue:0x4f0aec @kqfd=5, @fds={}>
irb(main):004:0> kq.add(:file, file, :events => [:write, :delete])
=> true
irb(main):005:0> kq.poll
=> []
irb(main):006:0> file.delete
=> #<File:/tmp/kqueue-test20090417-602-evm5wc-0>
irb(main):007:0> kq.poll
=> [{:type=>:file, :target=>#<File:/tmp/kqueue-test20090417-602-evm5wc-0>, :event=>:delete}]
irb(main):008:0> file.close and kq.close 
=> nil

Raises:

  • (ArgumentError) 


178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/ktools/kqueue.rb', line 178

def add_file(file, options={})
  fflags, flags = options.values_at :events, :flags
  raise ArgumentError.new("must specify which file events to watch for") unless fflags

  file = file.kind_of?(String) ? File.open(file, 'r') : file
  fdnum = file.respond_to?(:fileno) ? file.fileno : file

  k = Kevent.new
  flags = flags ? flags.inject(0){|m,i| m | KQ_FLAGS[i] } : EV_CLEAR
  fflags = fflags.inject(0){|m,i| m | KQ_FFLAGS[i] }
  ev_set(k, fdnum, EVFILT_VNODE, EV_ADD | flags, fflags, 0, nil)

  if kevent(@kqfd, k, 1, nil, 0, nil) == -1
    return false
  else
    @fds[fdnum] = {:target => file, :event => k}
    return true
  end
end

#add_process(pid, options = {}) ⇒ Object

Add events for a process. Takes a process id and and options hash. Supported events are:

  • :exit - The process has exited

  • :fork - The process has created a child process via fork(2) or similar call.

  • :exec - The process executed a new process via execve(2) or similar call.

  • :signal - The process was sent a signal. Status can be checked via waitpid(2) or similar call.

  • :reap - The process was reaped by the parent via wait(2) or similar call.\

Note: SIGNAL and REAP do not appear to exist in OSX older than Leopard.

Example:

irb(main):001:0> require 'ktools'
=> true
irb(main):002:0> kq = Kqueue.new
=> #<Kernel::Kqueue:0x14f55b4 @kqfd=4, @pids={}, @fds={}>
irb(main):003:0> fpid = fork{ sleep } 
=> 616
irb(main):004:0> kq.add(:process, fpid, :events => [:exit])
=> true
irb(main):005:0> Process.kill('TERM', fpid)
=> 1
irb(main):006:0> kq.poll.first
=> {:event=>:exit, :type=>:process, :target=>616}



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
 
# File 'lib/ktools/kqueue.rb', line 265

def add_process(pid, options={})
  flags, fflags = options.values_at :flags, :events
  flags = flags ? flags.inject(0){|m,i| m | KQ_FLAGS[i] } : EV_CLEAR
  fflags = fflags.inject(0){|m,i| m | KQ_FFLAGS[i] }

  k = Kevent.new
  ev_set(k, pid, EVFILT_PROC, EV_ADD | flags, fflags, 0, nil)

  if kevent(@kqfd, k, 1, nil, 0, nil) == -1
    return false
  else
    @pids[pid] = {:target => pid, :event => k}
    return true
  end
end

#add_socket(target, options = {}) ⇒ Object

Add events to a socket-style descriptor (socket or pipe). Your target can be either an IO object (socket, pipe), or a file descriptor number.

Supported events are:

  • :read - The descriptor has become readable.

  • :write - The descriptor has become writeable.

See the kqueue manpage for how behavior differs depending on the descriptor types. In general, you shouldn’t have to worry about it.

Example:

irb(main):001:0> require 'ktools'
=> true
irb(main):002:0> r, w = IO.pipe
=> [#<IO:0x4fa90c>, #<IO:0x4fa880>]
irb(main):003:0> kq = Kqueue.new
=> #<Kernel::Kqueue:0x4f43a4 @kqfd=6, @fds={}>
irb(main):004:0> kq.add(:socket, r, :events => [:read, :write])
=> true
irb(main):005:0> kq.poll
=> []
irb(main):006:0> w.write "foo"
=> 3
irb(main):007:0> kq.poll
=> [{:type=>:socket, :target=>#<IO:0x4fa90c>, :event=>:read}]
irb(main):008:0> [r, w, kq].each {|i| i.close}



225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/ktools/kqueue.rb', line 225

def add_socket(target, options={})
  filters, flags = options.values_at :events, :flags
  flags = flags ? flags.inject(0){|m,i| m | KQ_FLAGS[i] } : EV_CLEAR
  filters = filters ? filters.inject(0){|m,i| m | KQ_FILTERS[i] } : EVFILT_READ | EVFILT_WRITE
  fdnum = target.respond_to?(:fileno) ? target.fileno : target

  k = Kevent.new
  ev_set(k, fdnum, filters, EV_ADD | flags, 0, 0, nil)

  if kevent(@kqfd, k, 1, nil, 0, nil) == -1
    return false
  else
    @fds[fdnum] = {:target => target, :event => k}
    return true
  end
end

#close ⇒ Object

Close the kqueue descriptor. This essentially shuts down your kqueue and renders all active events on this kqueue removed.



383
384
385
 
# File 'lib/ktools/kqueue.rb', line 383

def close
  IO.for_fd(@kqfd).close
end

#delete(type, target) ⇒ Object

Stop generating events for the given type and event target, ie:

kq.delete(:process, 6244)



363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
 
# File 'lib/ktools/kqueue.rb', line 363

def delete(type, target)
  ident = target.respond_to?(:fileno) ? target.fileno : target
  container = case type
  when :socket
    @fds
  when :file
    @fds
  when :process
    @pids
  end
  h = container[ident]
  return false if h.nil?
  k = h[:event]
  ev_set(k, k[:ident], k[:filter], EV_DELETE, k[:fflags], 0, nil)
  kevent(@kqfd, k, 1, nil, 0, nil)
  container.delete(ident)
  return true
end

#poll(timeout = 0.0) ⇒ Object

Poll for an event. Pass an optional timeout float as number of seconds to wait for an event. Default is 0.0 (do not wait).

Using a timeout will block for the duration of the timeout. Under Ruby 1.9.1, we use rb_thread_blocking_region() under the hood to allow other threads to run during this call. Prior to 1.9 though, we do not have native threads and hence this call will block the whole interpreter (all threads) until it returns.

This call returns an array of hashes, similar to the following:

=> [{:type=>:socket, :target=>#<IO:0x4fa90c>, :event=>:read}]

  • :type - will be the type of event target, i.e. an event set with #add_file will have :type => :file

  • :target - the ‘target’ or ‘subject’ of the event. This can be a File, IO, process or signal number.

  • :event - the event that occurred on the target. This is one of the symbols you passed as :events => [:foo] when adding the event.



293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
 
# File 'lib/ktools/kqueue.rb', line 293

def poll(timeout=0.0)
  k = Kevent.new
  t = Timespec.new
  t[:tv_sec] = timeout.to_i
  t[:tv_nsec] = ((timeout - timeout.to_i) * 1e9).to_i

  case kevent(@kqfd, nil, 0, k, 1, t)
  when -1
    [errno]
  when 0
    []
  else
    [process_event(k)]
  end
end

#process_event(k) ⇒ Object

:nodoc:



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
 
# File 'lib/ktools/kqueue.rb', line 309

def process_event(k) #:nodoc:
  res = case k[:filter]
  when EVFILT_VNODE
    h = @fds[k[:ident]]
    return nil if h.nil?
    event = if k[:fflags] & NOTE_DELETE == NOTE_DELETE
      :delete
    elsif k[:fflags] & NOTE_WRITE == NOTE_WRITE
      :write
    elsif k[:fflags] & NOTE_EXTEND == NOTE_EXTEND
      :extend
    elsif k[:fflags] & NOTE_ATTRIB == NOTE_ATTRIB
      :attrib
    elsif k[:fflags] & NOTE_LINK == NOTE_LINK
      :link
    elsif k[:fflags] & NOTE_RENAME == NOTE_RENAME
      :rename
    elsif k[:fflags] & NOTE_REVOKE == NOTE_REVOKE
      :revoke
    end
    delete(:file, k[:ident]) if event == :delete || event == :revoke
    {:target => h[:target], :type => :file, :event => event}
  when EVFILT_READ
    h = @fds[k[:ident]]
    return nil if h.nil?
    {:target => h[:target], :type => :socket, :event => :read}
  when EVFILT_WRITE
    h = @fds[k[:ident]]
    return nil if h.nil?
    {:target => h[:target], :type => :socket, :event => :write}
  when EVFILT_PROC
    h = @pids[k[:ident]]
    return nil if h.nil?
    event = if k[:fflags] & NOTE_EXIT == NOTE_EXIT
      :exit
    elsif k[:fflags] & NOTE_FORK == NOTE_FORK
      :fork
    elsif k[:fflags] & NOTE_EXEC == NOTE_EXEC
      :exec
    elsif Kqueue.const_defined?("NOTE_SIGNAL") and k[:fflags] & NOTE_SIGNAL == NOTE_SIGNAL
      :signal
    elsif Kqueue.const_defined?("NOTE_REAP") and k[:fflags] & NOTE_REAP == NOTE_REAP
      :reap
    end
    delete(:process, k[:ident]) if event == :exit
    {:target => h[:target], :type => :process, :event => event}
  end

  delete(res[:type], res[:target]) if k[:flags] & EV_ONESHOT == EV_ONESHOT
  res
end