| Class | Rinda::TupleSpace |
| In: |
lib/rinda/tuplespace.rb
|
| Parent: | Object |
Creates a new TupleSpace. period is used to control how often to look for dead tuples after modifications to the TupleSpace.
If no dead tuples are found period seconds after the last modification, the TupleSpace will stop looking for dead tuples.
# File lib/rinda/tuplespace.rb, line 392
392: def initialize(period=60)
393: super()
394: @bag = TupleBag.new
395: @read_waiter = TupleBag.new
396: @take_waiter = TupleBag.new
397: @notify_waiter = TupleBag.new
398: @period = period
399: @keeper = nil
400: end
Moves tuple to port.
# File lib/rinda/tuplespace.rb, line 439
439: def move(port, tuple, sec=nil)
440: template = WaitTemplateEntry.new(self, tuple, sec)
441: yield(template) if block_given?
442: synchronize do
443: entry = @bag.find(template)
444: if entry
445: port.push(entry.value) if port
446: @bag.delete(entry)
447: notify_event('take', entry.value)
448: return entry.value
449: end
450: raise RequestExpiredError if template.expired?
451:
452: begin
453: @take_waiter.push(template)
454: start_keeper if template.expires
455: while true
456: raise RequestCanceledError if template.canceled?
457: raise RequestExpiredError if template.expired?
458: entry = @bag.find(template)
459: if entry
460: port.push(entry.value) if port
461: @bag.delete(entry)
462: notify_event('take', entry.value)
463: return entry.value
464: end
465: template.wait
466: end
467: ensure
468: @take_waiter.delete(template)
469: end
470: end
471: end
Registers for notifications of event. Returns a NotifyTemplateEntry. See NotifyTemplateEntry for examples of how to listen for notifications.
event can be:
| ‘write’: | A tuple was added |
| ‘take’: | A tuple was taken or moved |
| ‘delete’: | A tuple was lost after being overwritten or expiring |
The TupleSpace will also notify you of the ‘close’ event when the NotifyTemplateEntry has expired.
# File lib/rinda/tuplespace.rb, line 522
522: def notify(event, tuple, sec=nil)
523: template = NotifyTemplateEntry.new(self, event, tuple, sec)
524: synchronize do
525: @notify_waiter.push(template)
526: end
527: template
528: end
Reads tuple, but does not remove it.
# File lib/rinda/tuplespace.rb, line 476
476: def read(tuple, sec=nil)
477: template = WaitTemplateEntry.new(self, tuple, sec)
478: yield(template) if block_given?
479: synchronize do
480: entry = @bag.find(template)
481: return entry.value if entry
482: raise RequestExpiredError if template.expired?
483:
484: begin
485: @read_waiter.push(template)
486: start_keeper if template.expires
487: template.wait
488: raise RequestCanceledError if template.canceled?
489: raise RequestExpiredError if template.expired?
490: return template.found
491: ensure
492: @read_waiter.delete(template)
493: end
494: end
495: end
Returns all tuples matching tuple. Does not remove the found tuples.
# File lib/rinda/tuplespace.rb, line 500
500: def read_all(tuple)
501: template = WaitTemplateEntry.new(self, tuple, nil)
502: synchronize do
503: entry = @bag.find_all(template)
504: entry.collect do |e|
505: e.value
506: end
507: end
508: end
Removes tuple
# File lib/rinda/tuplespace.rb, line 432
432: def take(tuple, sec=nil, &block)
433: move(nil, tuple, sec, &block)
434: end
Adds tuple
# File lib/rinda/tuplespace.rb, line 405
405: def write(tuple, sec=nil)
406: entry = TupleEntry.new(tuple, sec)
407: synchronize do
408: if entry.expired?
409: @read_waiter.find_all_template(entry).each do |template|
410: template.read(tuple)
411: end
412: notify_event('write', entry.value)
413: notify_event('delete', entry.value)
414: else
415: @bag.push(entry)
416: start_keeper if entry.expires
417: @read_waiter.find_all_template(entry).each do |template|
418: template.read(tuple)
419: end
420: @take_waiter.find_all_template(entry).each do |template|
421: template.signal
422: end
423: notify_event('write', entry.value)
424: end
425: end
426: entry
427: end
Removes dead tuples.
# File lib/rinda/tuplespace.rb, line 535
535: def keep_clean
536: synchronize do
537: @read_waiter.delete_unless_alive.each do |e|
538: e.signal
539: end
540: @take_waiter.delete_unless_alive.each do |e|
541: e.signal
542: end
543: @notify_waiter.delete_unless_alive.each do |e|
544: e.notify(['close'])
545: end
546: @bag.delete_unless_alive.each do |e|
547: notify_event('delete', e.value)
548: end
549: end
550: end
Checks the tuplespace to see if it needs cleaning.
# File lib/rinda/tuplespace.rb, line 582
582: def need_keeper?
583: return true if @bag.has_expires?
584: return true if @read_waiter.has_expires?
585: return true if @take_waiter.has_expires?
586: return true if @notify_waiter.has_expires?
587: end
Notifies all registered listeners for event of a status change of tuple.
# File lib/rinda/tuplespace.rb, line 556
556: def notify_event(event, tuple)
557: ev = [event, tuple]
558: @notify_waiter.find_all_template(ev).each do |template|
559: template.notify(ev)
560: end
561: end
Creates a thread that scans the tuplespace for expired tuples.
# File lib/rinda/tuplespace.rb, line 566
566: def start_keeper
567: return if @keeper && @keeper.alive?
568: @keeper = Thread.new do
569: while true
570: synchronize do
571: break unless need_keeper?
572: keep_clean
573: end
574: sleep(@period)
575: end
576: end
577: end