github
/
ruby
зеркало из https://github.com/github/ruby.git
1
0
Форкнуть 0
Код Задачи Пакеты Проекты Релизы Вики Активность

Extract rinda

This commit is contained in:
Hiroshi SHIBATA 2024-01-19 18:11:29 +09:00
Родитель 0f315216bb
Коммит 411cda2d5c
7 изменённых файлов: 1 добавлений и 2574 удалений
Показать статистику Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

1
gems/bundled_gems
Просмотреть файл

@ -28,3 +28,4 @@ bigdecimal 3.1.6 https://github.com/ruby/bigdecimal
observer 0.1.2 https://github.com/ruby/observer
abbrev 0.1.2 https://github.com/ruby/abbrev
resolv-replace 0.1.1 https://github.com/ruby/resolv-replace
rinda 0.2.0 https://github.com/ruby/rinda

35
lib/rinda/rinda.gemspec
Просмотреть файл

@ -1,35 +0,0 @@
name = File.basename(__FILE__, ".gemspec")
version = ["lib/rinda", "."].find do |dir|
break File.foreach(File.join(__dir__, dir, "#{name}.rb")) do |line|
/^\s*VERSION\s*=\s*"(.*)"/ =~ line and break $1
end rescue nil
end
Gem::Specification.new do |spec|
spec.name = name
spec.version = version
spec.authors = ["Masatoshi SEKI"]
spec.email = ["seki@ruby-lang.org"]
spec.summary = %q{The Linda distributed computing paradigm in Ruby.}
spec.description = %q{The Linda distributed computing paradigm in Ruby.}
spec.homepage = "https://github.com/ruby/rinda"
spec.required_ruby_version = Gem::Requirement.new(">= 2.3.0")
spec.licenses = ["Ruby", "BSD-2-Clause"]
spec.metadata["homepage_uri"] = spec.homepage
spec.metadata["source_code_uri"] = spec.homepage
# Specify which files should be added to the gem when it is released.
# The `git ls-files -z` loads the files in the RubyGem that have been added into git.
spec.files = Dir.chdir(File.expand_path('..', __FILE__)) do
`git ls-files -z`.split("\x0").reject { |f| f.match(%r{^(test|spec|features)/}) }
end
spec.bindir = "exe"
spec.executables = spec.files.grep(%r{^exe/}) { |f| File.basename(f) }
spec.require_paths = ["lib"]
spec.add_dependency "drb"
spec.add_dependency "ipaddr"
spec.add_dependency "forwardable"
end

329
lib/rinda/rinda.rb
Просмотреть файл

@ -1,329 +0,0 @@
# frozen_string_literal: false
require 'drb/drb'
##
# A module to implement the Linda distributed computing paradigm in Ruby.
#
# Rinda is part of DRb (dRuby).
#
# == Example(s)
#
# See the sample/drb/ directory in the Ruby distribution, from 1.8.2 onwards.
#
#--
# TODO
# == Introduction to Linda/rinda?
#
# == Why is this library separate from DRb?
module Rinda
VERSION = "0.2.0"
##
# Rinda error base class
class RindaError < RuntimeError; end
##
# Raised when a hash-based tuple has an invalid key.
class InvalidHashTupleKey < RindaError; end
##
# Raised when trying to use a canceled tuple.
class RequestCanceledError < ThreadError; end
##
# Raised when trying to use an expired tuple.
class RequestExpiredError < ThreadError; end
##
# A tuple is the elementary object in Rinda programming.
# Tuples may be matched against templates if the tuple and
# the template are the same size.
class Tuple
##
# Creates a new Tuple from +ary_or_hash+ which must be an Array or Hash.
def initialize(ary_or_hash)
if hash?(ary_or_hash)
init_with_hash(ary_or_hash)
else
init_with_ary(ary_or_hash)
end
end
##
# The number of elements in the tuple.
def size
@tuple.size
end
##
# Accessor method for elements of the tuple.
def [](k)
@tuple[k]
end
##
# Fetches item +k+ from the tuple.
def fetch(k)
@tuple.fetch(k)
end
##
# Iterate through the tuple, yielding the index or key, and the
# value, thus ensuring arrays are iterated similarly to hashes.
def each # FIXME
if Hash === @tuple
@tuple.each { |k, v| yield(k, v) }
else
@tuple.each_with_index { |v, k| yield(k, v) }
end
end
##
# Return the tuple itself
def value
@tuple
end
private
def hash?(ary_or_hash)
ary_or_hash.respond_to?(:keys)
end
##
# Munges +ary+ into a valid Tuple.
def init_with_ary(ary)
@tuple = Array.new(ary.size)
@tuple.size.times do |i|
@tuple[i] = ary[i]
end
end
##
# Ensures +hash+ is a valid Tuple.
def init_with_hash(hash)
@tuple = Hash.new
hash.each do |k, v|
raise InvalidHashTupleKey unless String === k
@tuple[k] = v
end
end
end
##
# Templates are used to match tuples in Rinda.
class Template < Tuple
##
# Matches this template against +tuple+. The +tuple+ must be the same
# size as the template. An element with a +nil+ value in a template acts
# as a wildcard, matching any value in the corresponding position in the
# tuple. Elements of the template match the +tuple+ if the are #== or
# #===.
#
# Template.new([:foo, 5]).match Tuple.new([:foo, 5]) # => true
# Template.new([:foo, nil]).match Tuple.new([:foo, 5]) # => true
# Template.new([String]).match Tuple.new(['hello']) # => true
#
# Template.new([:foo]).match Tuple.new([:foo, 5]) # => false
# Template.new([:foo, 6]).match Tuple.new([:foo, 5]) # => false
# Template.new([:foo, nil]).match Tuple.new([:foo]) # => false
# Template.new([:foo, 6]).match Tuple.new([:foo]) # => false
def match(tuple)
return false unless tuple.respond_to?(:size)
return false unless tuple.respond_to?(:fetch)
return false unless self.size == tuple.size
each do |k, v|
begin
it = tuple.fetch(k)
rescue
return false
end
next if v.nil?
next if v == it
next if v === it
return false
end
return true
end
##
# Alias for #match.
def ===(tuple)
match(tuple)
end
end
##
# <i>Documentation?</i>
class DRbObjectTemplate
##
# Creates a new DRbObjectTemplate that will match against +uri+ and +ref+.
def initialize(uri=nil, ref=nil)
@drb_uri = uri
@drb_ref = ref
end
##
# This DRbObjectTemplate matches +ro+ if the remote object's drburi and
# drbref are the same. +nil+ is used as a wildcard.
def ===(ro)
return true if super(ro)
unless @drb_uri.nil?
return false unless (@drb_uri === ro.__drburi rescue false)
end
unless @drb_ref.nil?
return false unless (@drb_ref === ro.__drbref rescue false)
end
true
end
end
##
# TupleSpaceProxy allows a remote Tuplespace to appear as local.
class TupleSpaceProxy
##
# A Port ensures that a moved tuple arrives properly at its destination
# and does not get lost.
#
# See https://bugs.ruby-lang.org/issues/8125
class Port # :nodoc:
attr_reader :value
def self.deliver
port = new
begin
yield(port)
ensure
port.close
end
port.value
end
def initialize
@open = true
@value = nil
end
##
# Don't let the DRb thread push to it when remote sends tuple
def close
@open = false
end
##
# Stores +value+ and ensure it does not get marshaled multiple times.
def push value
raise 'port closed' unless @open
@value = value
nil # avoid Marshal
end
end
##
# Creates a new TupleSpaceProxy to wrap +ts+.
def initialize(ts)
@ts = ts
end
##
# Adds +tuple+ to the proxied TupleSpace. See TupleSpace#write.
def write(tuple, sec=nil)
@ts.write(tuple, sec)
end
##
# Takes +tuple+ from the proxied TupleSpace. See TupleSpace#take.
def take(tuple, sec=nil, &block)
Port.deliver do |port|
@ts.move(DRbObject.new(port), tuple, sec, &block)
end
end
##
# Reads +tuple+ from the proxied TupleSpace. See TupleSpace#read.
def read(tuple, sec=nil, &block)
@ts.read(tuple, sec, &block)
end
##
# Reads all tuples matching +tuple+ from the proxied TupleSpace. See
# TupleSpace#read_all.
def read_all(tuple)
@ts.read_all(tuple)
end
##
# Registers for notifications of event +ev+ on the proxied TupleSpace.
# See TupleSpace#notify
def notify(ev, tuple, sec=nil)
@ts.notify(ev, tuple, sec)
end
end
##
# An SimpleRenewer allows a TupleSpace to check if a TupleEntry is still
# alive.
class SimpleRenewer
include DRbUndumped
##
# Creates a new SimpleRenewer that keeps an object alive for another +sec+
# seconds.
def initialize(sec=180)
@sec = sec
end
##
# Called by the TupleSpace to check if the object is still alive.
def renew
@sec
end
end
end

484
lib/rinda/ring.rb
Просмотреть файл

@ -1,484 +0,0 @@
# frozen_string_literal: false
#
# Note: Rinda::Ring API is unstable.
#
require 'drb/drb'
require_relative 'rinda'
require 'ipaddr'
module Rinda
##
# The default port Ring discovery will use.
Ring_PORT = 7647
##
# A RingServer allows a Rinda::TupleSpace to be located via UDP broadcasts.
# Default service location uses the following steps:
#
# 1. A RingServer begins listening on the network broadcast UDP address.
# 2. A RingFinger sends a UDP packet containing the DRb URI where it will
# listen for a reply.
# 3. The RingServer receives the UDP packet and connects back to the
# provided DRb URI with the DRb service.
#
# A RingServer requires a TupleSpace:
#
# ts = Rinda::TupleSpace.new
# rs = Rinda::RingServer.new
#
# RingServer can also listen on multicast addresses for announcements. This
# allows multiple RingServers to run on the same host. To use network
# broadcast and multicast:
#
# ts = Rinda::TupleSpace.new
# rs = Rinda::RingServer.new ts, %w[Socket::INADDR_ANY, 239.0.0.1 ff02::1]
class RingServer
include DRbUndumped
##
# Special renewer for the RingServer to allow shutdown
class Renewer # :nodoc:
include DRbUndumped
##
# Set to false to shutdown future requests using this Renewer
attr_writer :renew
def initialize # :nodoc:
@renew = true
end
def renew # :nodoc:
@renew ? 1 : true
end
end
##
# Advertises +ts+ on the given +addresses+ at +port+.
#
# If +addresses+ is omitted only the UDP broadcast address is used.
#
# +addresses+ can contain multiple addresses. If a multicast address is
# given in +addresses+ then the RingServer will listen for multicast
# queries.
#
# If you use IPv4 multicast you may need to set an address of the inbound
# interface which joins a multicast group.
#
# ts = Rinda::TupleSpace.new
# rs = Rinda::RingServer.new(ts, [['239.0.0.1', '9.5.1.1']])
#
# You can set addresses as an Array Object. The first element of the
# Array is a multicast address and the second is an inbound interface
# address. If the second is omitted then '0.0.0.0' is used.
#
# If you use IPv6 multicast you may need to set both the local interface
# address and the inbound interface index:
#
# rs = Rinda::RingServer.new(ts, [['ff02::1', '::1', 1]])
#
# The first element is a multicast address and the second is an inbound
# interface address. The third is an inbound interface index.
#
# At this time there is no easy way to get an interface index by name.
#
# If the second is omitted then '::1' is used.
# If the third is omitted then 0 (default interface) is used.
def initialize(ts, addresses=[Socket::INADDR_ANY], port=Ring_PORT)
@port = port
if Integer === addresses then
addresses, @port = [Socket::INADDR_ANY], addresses
end
@renewer = Renewer.new
@ts = ts
@sockets = []
addresses.each do |address|
if Array === address
make_socket(*address)
else
make_socket(address)
end
end
@w_services = write_services
@r_service = reply_service
end
##
# Creates a socket at +address+
#
# If +address+ is multicast address then +interface_address+ and
# +multicast_interface+ can be set as optional.
#
# A created socket is bound to +interface_address+. If you use IPv4
# multicast then the interface of +interface_address+ is used as the
# inbound interface. If +interface_address+ is omitted or nil then
# '0.0.0.0' or '::1' is used.
#
# If you use IPv6 multicast then +multicast_interface+ is used as the
# inbound interface. +multicast_interface+ is a network interface index.
# If +multicast_interface+ is omitted then 0 (default interface) is used.
def make_socket(address, interface_address=nil, multicast_interface=0)
addrinfo = Addrinfo.udp(address, @port)
socket = Socket.new(addrinfo.pfamily, addrinfo.socktype,
addrinfo.protocol)
if addrinfo.ipv4_multicast? or addrinfo.ipv6_multicast? then
if Socket.const_defined?(:SO_REUSEPORT) then
socket.setsockopt(:SOCKET, :SO_REUSEPORT, true)
else
socket.setsockopt(:SOCKET, :SO_REUSEADDR, true)
end
if addrinfo.ipv4_multicast? then
interface_address = '0.0.0.0' if interface_address.nil?
socket.bind(Addrinfo.udp(interface_address, @port))
mreq = IPAddr.new(addrinfo.ip_address).hton +
IPAddr.new(interface_address).hton
socket.setsockopt(:IPPROTO_IP, :IP_ADD_MEMBERSHIP, mreq)
else
interface_address = '::1' if interface_address.nil?
socket.bind(Addrinfo.udp(interface_address, @port))
mreq = IPAddr.new(addrinfo.ip_address).hton +
[multicast_interface].pack('I')
socket.setsockopt(:IPPROTO_IPV6, :IPV6_JOIN_GROUP, mreq)
end
else
socket.bind(addrinfo)
end
socket
rescue
socket = socket.close if socket
raise
ensure
@sockets << socket if socket
end
##
# Creates threads that pick up UDP packets and passes them to do_write for
# decoding.
def write_services
@sockets.map do |s|
Thread.new(s) do |socket|
loop do
msg = socket.recv(1024)
do_write(msg)
end
end
end
end
##
# Extracts the response URI from +msg+ and adds it to TupleSpace where it
# will be picked up by +reply_service+ for notification.
def do_write(msg)
Thread.new do
begin
tuple, sec = Marshal.load(msg)
@ts.write(tuple, sec)
rescue
end
end
end
##
# Creates a thread that notifies waiting clients from the TupleSpace.
def reply_service
Thread.new do
loop do
do_reply
end
end
end
##
# Pulls lookup tuples out of the TupleSpace and sends their DRb object the
# address of the local TupleSpace.
def do_reply
tuple = @ts.take([:lookup_ring, nil], @renewer)
Thread.new { tuple[1].call(@ts) rescue nil}
rescue
end
##
# Shuts down the RingServer
def shutdown
@renewer.renew = false
@w_services.each do |thread|
thread.kill
thread.join
end
@sockets.each do |socket|
socket.close
end
@r_service.kill
@r_service.join
end
end
##
# RingFinger is used by RingServer clients to discover the RingServer's
# TupleSpace. Typically, all a client needs to do is call
# RingFinger.primary to retrieve the remote TupleSpace, which it can then
# begin using.
#
# To find the first available remote TupleSpace:
#
# Rinda::RingFinger.primary
#
# To create a RingFinger that broadcasts to a custom list:
#
# rf = Rinda::RingFinger.new ['localhost', '192.0.2.1']
# rf.primary
#
# Rinda::RingFinger also understands multicast addresses and sets them up
# properly. This allows you to run multiple RingServers on the same host:
#
# rf = Rinda::RingFinger.new ['239.0.0.1']
# rf.primary
#
# You can set the hop count (or TTL) for multicast searches using
# #multicast_hops.
#
# If you use IPv6 multicast you may need to set both an address and the
# outbound interface index:
#
# rf = Rinda::RingFinger.new ['ff02::1']
# rf.multicast_interface = 1
# rf.primary
#
# At this time there is no easy way to get an interface index by name.
class RingFinger
@@broadcast_list = ['<broadcast>', 'localhost']
@@finger = nil
##
# Creates a singleton RingFinger and looks for a RingServer. Returns the
# created RingFinger.
def self.finger
unless @@finger
@@finger = self.new
@@finger.lookup_ring_any
end
@@finger
end
##
# Returns the first advertised TupleSpace.
def self.primary
finger.primary
end
##
# Contains all discovered TupleSpaces except for the primary.
def self.to_a
finger.to_a
end
##
# The list of addresses where RingFinger will send query packets.
attr_accessor :broadcast_list
##
# Maximum number of hops for sent multicast packets (if using a multicast
# address in the broadcast list). The default is 1 (same as UDP
# broadcast).
attr_accessor :multicast_hops
##
# The interface index to send IPv6 multicast packets from.
attr_accessor :multicast_interface
##
# The port that RingFinger will send query packets to.
attr_accessor :port
##
# Contain the first advertised TupleSpace after lookup_ring_any is called.
attr_accessor :primary
##
# Creates a new RingFinger that will look for RingServers at +port+ on
# the addresses in +broadcast_list+.
#
# If +broadcast_list+ contains a multicast address then multicast queries
# will be made using the given multicast_hops and multicast_interface.
def initialize(broadcast_list=@@broadcast_list, port=Ring_PORT)
@broadcast_list = broadcast_list || ['localhost']
@port = port
@primary = nil
@rings = []
@multicast_hops = 1
@multicast_interface = 0
end
##
# Contains all discovered TupleSpaces except for the primary.
def to_a
@rings
end
##
# Iterates over all discovered TupleSpaces starting with the primary.
def each
lookup_ring_any unless @primary
return unless @primary
yield(@primary)
@rings.each { |x| yield(x) }
end
##
# Looks up RingServers waiting +timeout+ seconds. RingServers will be
# given +block+ as a callback, which will be called with the remote
# TupleSpace.
def lookup_ring(timeout=5, &block)
return lookup_ring_any(timeout) unless block_given?
msg = Marshal.dump([[:lookup_ring, DRbObject.new(block)], timeout])
@broadcast_list.each do |it|
send_message(it, msg)
end
sleep(timeout)
end
##
# Returns the first found remote TupleSpace. Any further recovered
# TupleSpaces can be found by calling +to_a+.
def lookup_ring_any(timeout=5)
queue = Thread::Queue.new
Thread.new do
self.lookup_ring(timeout) do |ts|
queue.push(ts)
end
queue.push(nil)
end
@primary = queue.pop
raise('RingNotFound') if @primary.nil?
Thread.new do
while it = queue.pop
@rings.push(it)
end
end
@primary
end
##
# Creates a socket for +address+ with the appropriate multicast options
# for multicast addresses.
def make_socket(address) # :nodoc:
addrinfo = Addrinfo.udp(address, @port)
soc = Socket.new(addrinfo.pfamily, addrinfo.socktype, addrinfo.protocol)
begin
if addrinfo.ipv4_multicast? then
soc.setsockopt(Socket::Option.ipv4_multicast_loop(1))
soc.setsockopt(Socket::Option.ipv4_multicast_ttl(@multicast_hops))
elsif addrinfo.ipv6_multicast? then
soc.setsockopt(:IPPROTO_IPV6, :IPV6_MULTICAST_LOOP, true)
soc.setsockopt(:IPPROTO_IPV6, :IPV6_MULTICAST_HOPS,
[@multicast_hops].pack('I'))
soc.setsockopt(:IPPROTO_IPV6, :IPV6_MULTICAST_IF,
[@multicast_interface].pack('I'))
else
soc.setsockopt(:SOL_SOCKET, :SO_BROADCAST, true)
end
soc.connect(addrinfo)
rescue Exception
soc.close
raise
end
soc
end
def send_message(address, message) # :nodoc:
soc = make_socket(address)
soc.send(message, 0)
rescue
nil
ensure
soc.close if soc
end
end
##
# RingProvider uses a RingServer advertised TupleSpace as a name service.
# TupleSpace clients can register themselves with the remote TupleSpace and
# look up other provided services via the remote TupleSpace.
#
# Services are registered with a tuple of the format [:name, klass,
# DRbObject, description].
class RingProvider
##
# Creates a RingProvider that will provide a +klass+ service running on
# +front+, with a +description+. +renewer+ is optional.
def initialize(klass, front, desc, renewer = nil)
@tuple = [:name, klass, front, desc]
@renewer = renewer || Rinda::SimpleRenewer.new
end
##
# Advertises this service on the primary remote TupleSpace.
def provide
ts = Rinda::RingFinger.primary
ts.write(@tuple, @renewer)
end
end
end

641
lib/rinda/tuplespace.rb
Просмотреть файл

@ -1,641 +0,0 @@
# frozen_string_literal: false
require 'monitor'
require 'drb/drb'
require_relative 'rinda'
require 'forwardable'
module Rinda
##
# A TupleEntry is a Tuple (i.e. a possible entry in some Tuplespace)
# together with expiry and cancellation data.
class TupleEntry
include DRbUndumped
attr_accessor :expires
##
# Creates a TupleEntry based on +ary+ with an optional renewer or expiry
# time +sec+.
#
# A renewer must implement the +renew+ method which returns a Numeric,
# nil, or true to indicate when the tuple has expired.
def initialize(ary, sec=nil)
@cancel = false
@expires = nil
@tuple = make_tuple(ary)
@renewer = nil
renew(sec)
end
##
# Marks this TupleEntry as canceled.
def cancel
@cancel = true
end
##
# A TupleEntry is dead when it is canceled or expired.
def alive?
!canceled? && !expired?
end
##
# Return the object which makes up the tuple itself: the Array
# or Hash.
def value; @tuple.value; end
##
# Returns the canceled status.
def canceled?; @cancel; end
##
# Has this tuple expired? (true/false).
#
# A tuple has expired when its expiry timer based on the +sec+ argument to
# #initialize runs out.
def expired?
return true unless @expires
return false if @expires > Time.now
return true if @renewer.nil?
renew(@renewer)
return true unless @expires
return @expires < Time.now
end
##
# Reset the expiry time according to +sec_or_renewer+.
#
# +nil+:: it is set to expire in the far future.
# +true+:: it has expired.
# Numeric:: it will expire in that many seconds.
#
# Otherwise the argument refers to some kind of renewer object
# which will reset its expiry time.
def renew(sec_or_renewer)
sec, @renewer = get_renewer(sec_or_renewer)
@expires = make_expires(sec)
end
##
# Returns an expiry Time based on +sec+ which can be one of:
# Numeric:: +sec+ seconds into the future
# +true+:: the expiry time is the start of 1970 (i.e. expired)
# +nil+:: it is Tue Jan 19 03:14:07 GMT Standard Time 2038 (i.e. when
# UNIX clocks will die)
def make_expires(sec=nil)
case sec
when Numeric
Time.now + sec
when true
Time.at(1)
when nil
Time.at(2**31-1)
end
end
##
# Retrieves +key+ from the tuple.
def [](key)
@tuple[key]
end
##
# Fetches +key+ from the tuple.
def fetch(key)
@tuple.fetch(key)
end
##
# The size of the tuple.
def size
@tuple.size
end
##
# Creates a Rinda::Tuple for +ary+.
def make_tuple(ary)
Rinda::Tuple.new(ary)
end
private
##
# Returns a valid argument to make_expires and the renewer or nil.
#
# Given +true+, +nil+, or Numeric, returns that value and +nil+ (no actual
# renewer). Otherwise it returns an expiry value from calling +it.renew+
# and the renewer.
def get_renewer(it)
case it
when Numeric, true, nil
return it, nil
else
begin
return it.renew, it
rescue Exception
return it, nil
end
end
end
end
##
# A TemplateEntry is a Template together with expiry and cancellation data.
class TemplateEntry < TupleEntry
##
# Matches this TemplateEntry against +tuple+. See Template#match for
# details on how a Template matches a Tuple.
def match(tuple)
@tuple.match(tuple)
end
alias === match
def make_tuple(ary) # :nodoc:
Rinda::Template.new(ary)
end
end
##
# <i>Documentation?</i>
class WaitTemplateEntry < TemplateEntry
attr_reader :found
def initialize(place, ary, expires=nil)
super(ary, expires)
@place = place
@cond = place.new_cond
@found = nil
end
def cancel
super
signal
end
def wait
@cond.wait
end
def read(tuple)
@found = tuple
signal
end
def signal
@place.synchronize do
@cond.signal
end
end
end
##
# A NotifyTemplateEntry is returned by TupleSpace#notify and is notified of
# TupleSpace changes. You may receive either your subscribed event or the
# 'close' event when iterating over notifications.
#
# See TupleSpace#notify_event for valid notification types.
#
# == Example
#
# ts = Rinda::TupleSpace.new
# observer = ts.notify 'write', [nil]
#
# Thread.start do
# observer.each { |t| p t }
# end
#
# 3.times { |i| ts.write [i] }
#
# Outputs:
#
# ['write', [0]]
# ['write', [1]]
# ['write', [2]]
class NotifyTemplateEntry < TemplateEntry
##
# Creates a new NotifyTemplateEntry that watches +place+ for +event+s that
# match +tuple+.
def initialize(place, event, tuple, expires=nil)
ary = [event, Rinda::Template.new(tuple)]
super(ary, expires)
@queue = Thread::Queue.new
@done = false
end
##
# Called by TupleSpace to notify this NotifyTemplateEntry of a new event.
def notify(ev)
@queue.push(ev)
end
##
# Retrieves a notification. Raises RequestExpiredError when this
# NotifyTemplateEntry expires.
def pop
raise RequestExpiredError if @done
it = @queue.pop
@done = true if it[0] == 'close'
return it
end
##
# Yields event/tuple pairs until this NotifyTemplateEntry expires.
def each # :yields: event, tuple
while !@done
it = pop
yield(it)
end
rescue
ensure
cancel
end
end
##
# TupleBag is an unordered collection of tuples. It is the basis
# of Tuplespace.
class TupleBag
class TupleBin
extend Forwardable
def_delegators '@bin', :find_all, :delete_if, :each, :empty?
def initialize
@bin = []
end
def add(tuple)
@bin.push(tuple)
end
def delete(tuple)
idx = @bin.rindex(tuple)
@bin.delete_at(idx) if idx
end
def find
@bin.reverse_each do |x|
return x if yield(x)
end
nil
end
end
def initialize # :nodoc:
@hash = {}
@enum = enum_for(:each_entry)
end
##
# +true+ if the TupleBag to see if it has any expired entries.
def has_expires?
@enum.find do |tuple|
tuple.expires
end
end
##
# Add +tuple+ to the TupleBag.
def push(tuple)
key = bin_key(tuple)
@hash[key] ||= TupleBin.new
@hash[key].add(tuple)
end
##
# Removes +tuple+ from the TupleBag.
def delete(tuple)
key = bin_key(tuple)
bin = @hash[key]
return nil unless bin
bin.delete(tuple)
@hash.delete(key) if bin.empty?
tuple
end
##
# Finds all live tuples that match +template+.
def find_all(template)
bin_for_find(template).find_all do |tuple|
tuple.alive? && template.match(tuple)
end
end
##
# Finds a live tuple that matches +template+.
def find(template)
bin_for_find(template).find do |tuple|
tuple.alive? && template.match(tuple)
end
end
##
# Finds all tuples in the TupleBag which when treated as templates, match
# +tuple+ and are alive.
def find_all_template(tuple)
@enum.find_all do |template|
template.alive? && template.match(tuple)
end
end
##
# Delete tuples which dead tuples from the TupleBag, returning the deleted
# tuples.
def delete_unless_alive
deleted = []
@hash.each do |key, bin|
bin.delete_if do |tuple|
if tuple.alive?
false
else
deleted.push(tuple)
true
end
end
end
deleted
end
private
def each_entry(&blk)
@hash.each do |k, v|
v.each(&blk)
end
end
def bin_key(tuple)
head = tuple[0]
if head.class == Symbol
return head
else
false
end
end
def bin_for_find(template)
key = bin_key(template)
key ? @hash.fetch(key, []) : @enum
end
end
##
# The Tuplespace manages access to the tuples it contains,
# ensuring mutual exclusion requirements are met.
#
# The +sec+ option for the write, take, move, read and notify methods may
# either be a number of seconds or a Renewer object.
class TupleSpace
include DRbUndumped
include MonitorMixin
##
# 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.
def initialize(period=60)
super()
@bag = TupleBag.new
@read_waiter = TupleBag.new
@take_waiter = TupleBag.new
@notify_waiter = TupleBag.new
@period = period
@keeper = nil
end
##
# Adds +tuple+
def write(tuple, sec=nil)
entry = create_entry(tuple, sec)
synchronize do
if entry.expired?
@read_waiter.find_all_template(entry).each do |template|
template.read(tuple)
end
notify_event('write', entry.value)
notify_event('delete', entry.value)
else
@bag.push(entry)
start_keeper if entry.expires
@read_waiter.find_all_template(entry).each do |template|
template.read(tuple)
end
@take_waiter.find_all_template(entry).each do |template|
template.signal
end
notify_event('write', entry.value)
end
end
entry
end
##
# Removes +tuple+
def take(tuple, sec=nil, &block)
move(nil, tuple, sec, &block)
end
##
# Moves +tuple+ to +port+.
def move(port, tuple, sec=nil)
template = WaitTemplateEntry.new(self, tuple, sec)
yield(template) if block_given?
synchronize do
entry = @bag.find(template)
if entry
port.push(entry.value) if port
@bag.delete(entry)
notify_event('take', entry.value)
return port ? nil : entry.value
end
raise RequestExpiredError if template.expired?
begin
@take_waiter.push(template)
start_keeper if template.expires
while true
raise RequestCanceledError if template.canceled?
raise RequestExpiredError if template.expired?
entry = @bag.find(template)
if entry
port.push(entry.value) if port
@bag.delete(entry)
notify_event('take', entry.value)
return port ? nil : entry.value
end
template.wait
end
ensure
@take_waiter.delete(template)
end
end
end
##
# Reads +tuple+, but does not remove it.
def read(tuple, sec=nil)
template = WaitTemplateEntry.new(self, tuple, sec)
yield(template) if block_given?
synchronize do
entry = @bag.find(template)
return entry.value if entry
raise RequestExpiredError if template.expired?
begin
@read_waiter.push(template)
start_keeper if template.expires
template.wait
raise RequestCanceledError if template.canceled?
raise RequestExpiredError if template.expired?
return template.found
ensure
@read_waiter.delete(template)
end
end
end
##
# Returns all tuples matching +tuple+. Does not remove the found tuples.
def read_all(tuple)
template = WaitTemplateEntry.new(self, tuple, nil)
synchronize do
entry = @bag.find_all(template)
entry.collect do |e|
e.value
end
end
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.
def notify(event, tuple, sec=nil)
template = NotifyTemplateEntry.new(self, event, tuple, sec)
synchronize do
@notify_waiter.push(template)
end
template
end
private
def create_entry(tuple, sec)
TupleEntry.new(tuple, sec)
end
##
# Removes dead tuples.
def keep_clean
synchronize do
@read_waiter.delete_unless_alive.each do |e|
e.signal
end
@take_waiter.delete_unless_alive.each do |e|
e.signal
end
@notify_waiter.delete_unless_alive.each do |e|
e.notify(['close'])
end
@bag.delete_unless_alive.each do |e|
notify_event('delete', e.value)
end
end
end
##
# Notifies all registered listeners for +event+ of a status change of
# +tuple+.
def notify_event(event, tuple)
ev = [event, tuple]
@notify_waiter.find_all_template(ev).each do |template|
template.notify(ev)
end
end
##
# Creates a thread that scans the tuplespace for expired tuples.
def start_keeper
return if @keeper && @keeper.alive?
@keeper = Thread.new do
while true
sleep(@period)
synchronize do
break unless need_keeper?
keep_clean
end
end
end
end
##
# Checks the tuplespace to see if it needs cleaning.
def need_keeper?
return true if @bag.has_expires?
return true if @read_waiter.has_expires?
return true if @take_waiter.has_expires?
return true if @notify_waiter.has_expires?
end
end
end

912
test/rinda/test_rinda.rb
Просмотреть файл

@ -1,912 +0,0 @@
# frozen_string_literal: false
require 'test/unit'
require 'envutil'
require 'drb/drb'
require 'drb/eq'
require 'rinda/ring'
require 'rinda/tuplespace'
require 'timeout'
require 'singleton'
module Rinda
class MockClock
include Singleton
class MyTS < Rinda::TupleSpace
def keeper_thread
nil
end
def stop_keeper
if @keeper
@keeper.kill
@keeper.join
@keeper = nil
end
end
end
def initialize
@now = 2
@reso = 1
@ts = nil
@inf = 2**31 - 1
end
def start_keeper
@now = 2
@reso = 1
@ts&.stop_keeper
@ts = MyTS.new
@ts.write([2, :now])
@inf = 2**31 - 1
end
def stop_keeper
@ts.stop_keeper
end
def now
@now.to_f
end
def at(n)
n
end
def _forward(n=nil)
now ,= @ts.take([nil, :now])
@now = now + n
@ts.write([@now, :now])
end
def forward(n)
while n > 0
_forward(@reso)
n -= @reso
Thread.pass
end
end
def rewind
@ts.take([nil, :now])
@ts.write([@inf, :now])
@ts.take([nil, :now])
@now = 2
@ts.write([2, :now])
end
def sleep(n=nil)
now ,= @ts.read([nil, :now])
@ts.read([(now + n)..@inf, :now])
0
end
end
module Time
def sleep(n)
@m.sleep(n)
end
module_function :sleep
def at(n)
n
end
module_function :at
def now
defined?(@m) && @m ? @m.now : 2
end
module_function :now
def rewind
@m.rewind
end
module_function :rewind
def forward(n)
@m.forward(n)
end
module_function :forward
@m = MockClock.instance
end
class TupleSpace
def sleep(n)
Kernel.sleep(n * 0.01)
end
end
module TupleSpaceTestModule
def setup
MockClock.instance.start_keeper
end
def teardown
MockClock.instance.stop_keeper
end
def sleep(n)
if Thread.current == Thread.main
Time.forward(n)
else
Time.sleep(n)
end
end
def thread_join(th)
while th.alive?
Kernel.sleep(0.1)
sleep(1)
end
th.value
end
def test_00_tuple
tuple = Rinda::TupleEntry.new([1,2,3])
assert(!tuple.canceled?)
assert(!tuple.expired?)
assert(tuple.alive?)
end
def test_00_template
tmpl = Rinda::Template.new([1,2,3])
assert_equal(3, tmpl.size)
assert_equal(3, tmpl[2])
assert(tmpl.match([1,2,3]))
assert(!tmpl.match([1,nil,3]))
tmpl = Rinda::Template.new([/^rinda/i, nil, :hello])
assert_equal(3, tmpl.size)
assert(tmpl.match(['Rinda', 2, :hello]))
assert(!tmpl.match(['Rinda', 2, Symbol]))
assert(!tmpl.match([1, 2, :hello]))
assert(tmpl.match([/^rinda/i, 2, :hello]))
tmpl = Rinda::Template.new([Symbol])
assert_equal(1, tmpl.size)
assert(tmpl.match([:hello]))
assert(tmpl.match([Symbol]))
assert(!tmpl.match(['Symbol']))
tmpl = Rinda::Template.new({"message"=>String, "name"=>String})
assert_equal(2, tmpl.size)
assert(tmpl.match({"message"=>"Hello", "name"=>"Foo"}))
assert(!tmpl.match({"message"=>"Hello", "name"=>"Foo", "1"=>2}))
assert(!tmpl.match({"message"=>"Hi", "name"=>"Foo", "age"=>1}))
assert(!tmpl.match({"message"=>"Hello", "no_name"=>"Foo"}))
assert_raise(Rinda::InvalidHashTupleKey) do
Rinda::Template.new({:message=>String, "name"=>String})
end
tmpl = Rinda::Template.new({"name"=>String})
assert_equal(1, tmpl.size)
assert(tmpl.match({"name"=>"Foo"}))
assert(!tmpl.match({"message"=>"Hello", "name"=>"Foo"}))
assert(!tmpl.match({"message"=>:symbol, "name"=>"Foo", "1"=>2}))
assert(!tmpl.match({"message"=>"Hi", "name"=>"Foo", "age"=>1}))
assert(!tmpl.match({"message"=>"Hello", "no_name"=>"Foo"}))
tmpl = Rinda::Template.new({"message"=>String, "name"=>String})
assert_equal(2, tmpl.size)
assert(tmpl.match({"message"=>"Hello", "name"=>"Foo"}))
assert(!tmpl.match({"message"=>"Hello", "name"=>"Foo", "1"=>2}))
assert(!tmpl.match({"message"=>"Hi", "name"=>"Foo", "age"=>1}))
assert(!tmpl.match({"message"=>"Hello", "no_name"=>"Foo"}))
tmpl = Rinda::Template.new({"message"=>String})
assert_equal(1, tmpl.size)
assert(tmpl.match({"message"=>"Hello"}))
assert(!tmpl.match({"message"=>"Hello", "name"=>"Foo"}))
assert(!tmpl.match({"message"=>"Hello", "name"=>"Foo", "1"=>2}))
assert(!tmpl.match({"message"=>"Hi", "name"=>"Foo", "age"=>1}))
assert(!tmpl.match({"message"=>"Hello", "no_name"=>"Foo"}))
tmpl = Rinda::Template.new({"message"=>String, "name"=>nil})
assert_equal(2, tmpl.size)
assert(tmpl.match({"message"=>"Hello", "name"=>"Foo"}))
assert(!tmpl.match({"message"=>"Hello", "name"=>"Foo", "1"=>2}))
assert(!tmpl.match({"message"=>"Hi", "name"=>"Foo", "age"=>1}))
assert(!tmpl.match({"message"=>"Hello", "no_name"=>"Foo"}))
assert_raise(Rinda::InvalidHashTupleKey) do
@ts.write({:message=>String, "name"=>String})
end
@ts.write([1, 2, 3])
assert_equal([1, 2, 3], @ts.take([1, 2, 3]))
@ts.write({'1'=>1, '2'=>2, '3'=>3})
assert_equal({'1'=>1, '2'=>2, '3'=>3}, @ts.take({'1'=>1, '2'=>2, '3'=>3}))
entry = @ts.write(['1'=>1, '2'=>2, '3'=>3])
assert_raise(Rinda::RequestExpiredError) do
assert_equal({'1'=>1, '2'=>2, '3'=>3}, @ts.read({'1'=>1}, 0))
end
entry.cancel
end
def test_00_DRbObject
ro = DRbObject.new(nil, "druby://host:1234")
tmpl = Rinda::DRbObjectTemplate.new
assert(tmpl === ro)
tmpl = Rinda::DRbObjectTemplate.new("druby://host:1234")
assert(tmpl === ro)
tmpl = Rinda::DRbObjectTemplate.new("druby://host:12345")
assert(!(tmpl === ro))
tmpl = Rinda::DRbObjectTemplate.new(/^druby:\/\/host:/)
assert(tmpl === ro)
ro = DRbObject.new_with(12345, 1234)
assert(!(tmpl === ro))
ro = DRbObject.new_with("druby://foo:12345", 1234)
assert(!(tmpl === ro))
tmpl = Rinda::DRbObjectTemplate.new(/^druby:\/\/(foo|bar):/)
assert(tmpl === ro)
ro = DRbObject.new_with("druby://bar:12345", 1234)
assert(tmpl === ro)
ro = DRbObject.new_with("druby://baz:12345", 1234)
assert(!(tmpl === ro))
end
def test_inp_rdp
assert_raise(Rinda::RequestExpiredError) do
@ts.take([:empty], 0)
end
assert_raise(Rinda::RequestExpiredError) do
@ts.read([:empty], 0)
end
end
def test_ruby_talk_264062
th = Thread.new {
assert_raise(Rinda::RequestExpiredError) do
@ts.take([:empty], 1)
end
}
sleep(10)
thread_join(th)
th = Thread.new {
assert_raise(Rinda::RequestExpiredError) do
@ts.read([:empty], 1)
end
}
sleep(10)
thread_join(th)
end
def test_symbol_tuple
@ts.write([:symbol, :symbol])
@ts.write(['string', :string])
assert_equal([[:symbol, :symbol]], @ts.read_all([:symbol, nil]))
assert_equal([[:symbol, :symbol]], @ts.read_all([Symbol, nil]))
assert_equal([], @ts.read_all([:nil, nil]))
end
def test_core_01
5.times do
@ts.write([:req, 2])
end
assert_equal([[:req, 2], [:req, 2], [:req, 2], [:req, 2], [:req, 2]],
@ts.read_all([nil, nil]))
taker = Thread.new(5) do |count|
s = 0
count.times do
tuple = @ts.take([:req, Integer])
assert_equal(2, tuple[1])
s += tuple[1]
end
@ts.write([:ans, s])
s
end
assert_equal(10, thread_join(taker))
assert_equal([:ans, 10], @ts.take([:ans, 10]))
assert_equal([], @ts.read_all([nil, nil]))
end
def test_core_02
taker = Thread.new(5) do |count|
s = 0
count.times do
tuple = @ts.take([:req, Integer])
assert_equal(2, tuple[1])
s += tuple[1]
end
@ts.write([:ans, s])
s
end
5.times do
@ts.write([:req, 2])
end
assert_equal(10, thread_join(taker))
assert_equal([:ans, 10], @ts.take([:ans, 10]))
assert_equal([], @ts.read_all([nil, nil]))
end
def test_core_03_notify
notify1 = @ts.notify(nil, [:req, Integer])
notify2 = @ts.notify(nil, {"message"=>String, "name"=>String})
5.times do
@ts.write([:req, 2])
end
5.times do
tuple = @ts.take([:req, Integer])
assert_equal(2, tuple[1])
end
5.times do
assert_equal(['write', [:req, 2]], notify1.pop)
end
5.times do
assert_equal(['take', [:req, 2]], notify1.pop)
end
@ts.write({"message"=>"first", "name"=>"3"})
@ts.write({"message"=>"second", "name"=>"1"})
@ts.write({"message"=>"third", "name"=>"0"})
@ts.take({"message"=>"third", "name"=>"0"})
@ts.take({"message"=>"first", "name"=>"3"})
assert_equal(["write", {"message"=>"first", "name"=>"3"}], notify2.pop)
assert_equal(["write", {"message"=>"second", "name"=>"1"}], notify2.pop)
assert_equal(["write", {"message"=>"third", "name"=>"0"}], notify2.pop)
assert_equal(["take", {"message"=>"third", "name"=>"0"}], notify2.pop)
assert_equal(["take", {"message"=>"first", "name"=>"3"}], notify2.pop)
end
def test_cancel_01
entry = @ts.write([:removeme, 1])
assert_equal([[:removeme, 1]], @ts.read_all([nil, nil]))
entry.cancel
assert_equal([], @ts.read_all([nil, nil]))
template = nil
taker = Thread.new do
assert_raise(Rinda::RequestCanceledError) do
@ts.take([:take, nil], 10) do |t|
template = t
Thread.new do
template.cancel
end
end
end
end
sleep(2)
thread_join(taker)
assert(template.canceled?)
@ts.write([:take, 1])
assert_equal([[:take, 1]], @ts.read_all([nil, nil]))
end
def test_cancel_02
omit 'this test is unstable with --jit-wait' if defined?(RubyVM::RJIT) && RubyVM::RJIT.enabled?
entry = @ts.write([:removeme, 1])
assert_equal([[:removeme, 1]], @ts.read_all([nil, nil]))
entry.cancel
assert_equal([], @ts.read_all([nil, nil]))
template = nil
reader = Thread.new do
assert_raise(Rinda::RequestCanceledError) do
@ts.read([:take, nil], 10) do |t|
template = t
Thread.new do
template.cancel
end
end
end
end
sleep(2)
thread_join(reader)
assert(template.canceled?)
@ts.write([:take, 1])
assert_equal([[:take, 1]], @ts.read_all([nil, nil]))
end
class SimpleRenewer
def initialize(sec, n = 1)
@sec = sec
@n = n
end
def renew
return -1 if @n <= 0
@n -= 1
return @sec
end
end
def test_00_renewer
tuple = Rinda::TupleEntry.new([1,2,3], true)
assert(!tuple.canceled?)
assert(tuple.expired?)
assert(!tuple.alive?)
tuple = Rinda::TupleEntry.new([1,2,3], 1)
assert(!tuple.canceled?)
assert(!tuple.expired?)
assert(tuple.alive?)
sleep(2)
assert(tuple.expired?)
assert(!tuple.alive?)
@renewer = SimpleRenewer.new(1,2)
tuple = Rinda::TupleEntry.new([1,2,3], @renewer)
assert(!tuple.canceled?)
assert(!tuple.expired?)
assert(tuple.alive?)
sleep(1)
assert(!tuple.canceled?)
assert(!tuple.expired?)
assert(tuple.alive?)
sleep(2)
assert(tuple.expired?)
assert(!tuple.alive?)
end
end
class TupleSpaceTest < Test::Unit::TestCase
include TupleSpaceTestModule
def setup
super
ThreadGroup.new.add(Thread.current)
@ts = Rinda::TupleSpace.new(1)
end
def teardown
# implementation-dependent
@ts.instance_eval{
if th = @keeper
th.kill
th.join
end
}
super
end
end
class TupleSpaceProxyTest < Test::Unit::TestCase
include TupleSpaceTestModule
def setup
if RUBY_PLATFORM.match?(/mingw/)
@omitted = true
omit 'This test seems to randomly hang on GitHub Actions MinGW'
end
super
ThreadGroup.new.add(Thread.current)
@ts_base = Rinda::TupleSpace.new(1)
@ts = Rinda::TupleSpaceProxy.new(@ts_base)
@server = DRb.start_service("druby://localhost:0")
end
def teardown
return if @omitted
@omitted = false
# implementation-dependent
@ts_base.instance_eval{
if th = @keeper
th.kill
th.join
end
}
@server.stop_service
DRb::DRbConn.stop_pool
super
end
def test_remote_array_and_hash
# Don't remove ary/hsh local variables.
# These are necessary to protect objects from GC.
ary = [1, 2, 3]
@ts.write(DRbObject.new(ary))
assert_equal([1, 2, 3], @ts.take([1, 2, 3], 0))
hsh = {'head' => 1, 'tail' => 2}
@ts.write(DRbObject.new(hsh))
assert_equal({'head' => 1, 'tail' => 2},
@ts.take({'head' => 1, 'tail' => 2}, 0))
end
def test_take_bug_8215
omit "this test randomly fails on mswin" if /mswin/ =~ RUBY_PLATFORM
service = DRb.start_service("druby://localhost:0", @ts_base)
uri = service.uri
args = [EnvUtil.rubybin, *%W[-rdrb/drb -rdrb/eq -rrinda/ring -rrinda/tuplespace -e]]
take = spawn(*args, <<-'end;', uri)
uri = ARGV[0]
DRb.start_service("druby://localhost:0")
ro = DRbObject.new_with_uri(uri)
ts = Rinda::TupleSpaceProxy.new(ro)
th = Thread.new do
ts.take([:test_take, nil])
rescue Interrupt
# Expected
end
Kernel.sleep(0.1)
th.raise(Interrupt) # causes loss of the taken tuple
ts.write([:barrier, :continue])
Kernel.sleep
end;
@ts_base.take([:barrier, :continue])
write = spawn(*args, <<-'end;', uri)
uri = ARGV[0]
DRb.start_service("druby://localhost:0")
ro = DRbObject.new_with_uri(uri)
ts = Rinda::TupleSpaceProxy.new(ro)
ts.write([:test_take, 42])
end;
status = Process.wait(write)
assert_equal([[:test_take, 42]], @ts_base.read_all([:test_take, nil]),
'[bug:8215] tuple lost')
ensure
service.stop_service if service
DRb::DRbConn.stop_pool
signal = /mswin|mingw/ =~ RUBY_PLATFORM ? "KILL" : "TERM"
Process.kill(signal, write) if write && status.nil?
Process.kill(signal, take) if take
Process.wait(write) if write && status.nil?
Process.wait(take) if take
end
end
module RingIPv4
def ipv4_mc(rf)
begin
v4mc = rf.make_socket('239.0.0.1')
rescue Errno::ENETUNREACH, Errno::ENOBUFS, Errno::ENODEV
omit 'IPv4 multicast not available'
end
begin
yield v4mc
ensure
v4mc.close
end
end
end
module RingIPv6
def prepare_ipv6(r)
begin
Socket.getifaddrs.each do |ifaddr|
next unless ifaddr.addr
next unless ifaddr.addr.ipv6_linklocal?
next if ifaddr.name[0, 2] == "lo"
r.multicast_interface = ifaddr.ifindex
return ifaddr
end
rescue NotImplementedError
# ifindex() function may not be implemented on Windows.
return if
Socket.ip_address_list.any? { |addrinfo| addrinfo.ipv6? && !addrinfo.ipv6_loopback? }
end
omit 'IPv6 not available'
end
def ipv6_mc(rf, hops = nil)
ifaddr = prepare_ipv6(rf)
rf.multicast_hops = hops if hops
begin
v6mc = rf.make_socket("ff02::1")
rescue Errno::EINVAL
# somehow Debian 6.0.7 needs ifname
v6mc = rf.make_socket("ff02::1%#{ifaddr.name}")
rescue Errno::EADDRNOTAVAIL
return # IPv6 address for multicast not available
rescue Errno::ENETDOWN
return # Network is down
rescue Errno::EHOSTUNREACH
return # Unreachable for some reason
end
begin
yield v6mc
ensure
v6mc.close
end
end
end
class TestRingServer < Test::Unit::TestCase
include RingIPv4
def setup
@port = Rinda::Ring_PORT
@ts = Rinda::TupleSpace.new
@rs = Rinda::RingServer.new(@ts, [], @port)
@server = DRb.start_service("druby://localhost:0")
end
def teardown
@rs.shutdown
# implementation-dependent
@ts.instance_eval{
if th = @keeper
th.kill
th.join
end
}
@server.stop_service
DRb::DRbConn.stop_pool
end
def test_do_reply
with_timeout(30) {_test_do_reply}
end
def _test_do_reply
called = nil
callback_orig = proc { |ts|
called = ts
}
callback = DRb::DRbObject.new callback_orig
@ts.write [:lookup_ring, callback]
@rs.do_reply
wait_for(30) {called}
assert_same @ts, called
end
def test_do_reply_local
omit 'timeout-based test becomes unstable with --jit-wait' if defined?(RubyVM::RJIT) && RubyVM::RJIT.enabled?
with_timeout(30) {_test_do_reply_local}
end
def _test_do_reply_local
called = nil
callback = proc { |ts|
called = ts
}
@ts.write [:lookup_ring, callback]
@rs.do_reply
wait_for(30) {called}
assert_same @ts, called
end
def test_make_socket_unicast
v4 = @rs.make_socket('127.0.0.1')
assert_equal('127.0.0.1', v4.local_address.ip_address)
assert_equal(@port, v4.local_address.ip_port)
end
def test_make_socket_ipv4_multicast
ipv4_mc(@rs) do |v4mc|
begin
if Socket.const_defined?(:SO_REUSEPORT) then
assert(v4mc.getsockopt(:SOCKET, :SO_REUSEPORT).bool)
else
assert(v4mc.getsockopt(:SOCKET, :SO_REUSEADDR).bool)
end
rescue TypeError
if /aix/ =~ RUBY_PLATFORM
omit "Known bug in getsockopt(2) on AIX"
end
raise $!
end
assert_equal('0.0.0.0', v4mc.local_address.ip_address)
assert_equal(@port, v4mc.local_address.ip_port)
end
end
def test_make_socket_ipv6_multicast
omit 'IPv6 not available' unless
Socket.ip_address_list.any? { |addrinfo| addrinfo.ipv6? && !addrinfo.ipv6_loopback? }
begin
v6mc = @rs.make_socket('ff02::1')
rescue Errno::EADDRNOTAVAIL
return # IPv6 address for multicast not available
rescue Errno::ENOBUFS => e
omit "Missing multicast support in OS: #{e.message}"
end
if Socket.const_defined?(:SO_REUSEPORT) then
assert v6mc.getsockopt(:SOCKET, :SO_REUSEPORT).bool
else
assert v6mc.getsockopt(:SOCKET, :SO_REUSEADDR).bool
end
assert_equal('::1', v6mc.local_address.ip_address)
assert_equal(@port, v6mc.local_address.ip_port)
end
def test_ring_server_ipv4_multicast
@rs.shutdown
begin
@rs = Rinda::RingServer.new(@ts, [['239.0.0.1', '0.0.0.0']], @port)
rescue Errno::ENOBUFS, Errno::ENODEV => e
omit "Missing multicast support in OS: #{e.message}"
end
v4mc = @rs.instance_variable_get('@sockets').first
begin
if Socket.const_defined?(:SO_REUSEPORT) then
assert(v4mc.getsockopt(:SOCKET, :SO_REUSEPORT).bool)
else
assert(v4mc.getsockopt(:SOCKET, :SO_REUSEADDR).bool)
end
rescue TypeError
if /aix/ =~ RUBY_PLATFORM
omit "Known bug in getsockopt(2) on AIX"
end
raise $!
end
assert_equal('0.0.0.0', v4mc.local_address.ip_address)
assert_equal(@port, v4mc.local_address.ip_port)
end
def test_ring_server_ipv6_multicast
omit 'IPv6 not available' unless
Socket.ip_address_list.any? { |addrinfo| addrinfo.ipv6? && !addrinfo.ipv6_loopback? }
@rs.shutdown
begin
@rs = Rinda::RingServer.new(@ts, [['ff02::1', '::1', 0]], @port)
rescue Errno::EADDRNOTAVAIL
return # IPv6 address for multicast not available
end
v6mc = @rs.instance_variable_get('@sockets').first
if Socket.const_defined?(:SO_REUSEPORT) then
assert v6mc.getsockopt(:SOCKET, :SO_REUSEPORT).bool
else
assert v6mc.getsockopt(:SOCKET, :SO_REUSEADDR).bool
end
assert_equal('::1', v6mc.local_address.ip_address)
assert_equal(@port, v6mc.local_address.ip_port)
end
def test_shutdown
@rs.shutdown
assert_nil(@rs.do_reply, 'otherwise should hang forever')
end
private
def with_timeout(n)
aoe = Thread.abort_on_exception
Thread.abort_on_exception = true
tl0 = Thread.list
tl = nil
th = Thread.new(Thread.current) do |mth|
sleep n
(tl = Thread.list - tl0).each {|t|t.raise(Timeout::Error)}
mth.raise(Timeout::Error)
end
tl0 << th
yield
rescue Timeout::Error => e
$stderr.puts "TestRingServer#with_timeout: timeout in #{n}s:"
$stderr.puts caller
if tl
bt = e.backtrace
tl.each do |t|
begin
t.value
rescue Timeout::Error => e
bt.unshift("")
bt[0, 0] = e.backtrace
end
end
end
raise Timeout::Error, "timeout", bt
ensure
if th
th.kill
th.join
end
Thread.abort_on_exception = aoe
end
def wait_for(n)
t = n + Process.clock_gettime(Process::CLOCK_MONOTONIC, :second)
until yield
if t < Process.clock_gettime(Process::CLOCK_MONOTONIC, :second)
flunk "timeout during waiting call"
end
sleep 0.1
end
end
end
class TestRingFinger < Test::Unit::TestCase
include RingIPv6
include RingIPv4
def setup
@rf = Rinda::RingFinger.new
end
def test_make_socket_unicast
v4 = @rf.make_socket('127.0.0.1')
assert(v4.getsockopt(:SOL_SOCKET, :SO_BROADCAST).bool)
rescue TypeError
if /aix/ =~ RUBY_PLATFORM
omit "Known bug in getsockopt(2) on AIX"
end
raise $!
ensure
v4.close if v4
end
def test_make_socket_ipv4_multicast
ipv4_mc(@rf) do |v4mc|
assert_equal(1, v4mc.getsockopt(:IPPROTO_IP, :IP_MULTICAST_LOOP).ipv4_multicast_loop)
assert_equal(1, v4mc.getsockopt(:IPPROTO_IP, :IP_MULTICAST_TTL).ipv4_multicast_ttl)
end
end
def test_make_socket_ipv6_multicast
ipv6_mc(@rf) do |v6mc|
assert_equal(1, v6mc.getsockopt(:IPPROTO_IPV6, :IPV6_MULTICAST_LOOP).int)
assert_equal(1, v6mc.getsockopt(:IPPROTO_IPV6, :IPV6_MULTICAST_HOPS).int)
end
end
def test_make_socket_ipv4_multicast_hops
@rf.multicast_hops = 2
ipv4_mc(@rf) do |v4mc|
assert_equal(2, v4mc.getsockopt(:IPPROTO_IP, :IP_MULTICAST_TTL).ipv4_multicast_ttl)
end
end
def test_make_socket_ipv6_multicast_hops
ipv6_mc(@rf, 2) do |v6mc|
assert_equal(2, v6mc.getsockopt(:IPPROTO_IPV6, :IPV6_MULTICAST_HOPS).int)
end
end
end
end

173
test/rinda/test_tuplebag.rb
Просмотреть файл

@ -1,173 +0,0 @@
# frozen_string_literal: false
require 'test/unit'
require 'rinda/tuplespace'
class TestTupleBag < Test::Unit::TestCase
def setup
@tb = Rinda::TupleBag.new
end
def test_delete
assert_nothing_raised do
val = @tb.delete tup(:val, 1)
assert_equal nil, val
end
t = tup(:val, 1)
@tb.push t
val = @tb.delete t
assert_equal t, val
assert_equal [], @tb.find_all(tem(:val, 1))
t1 = tup(:val, 1)
t2 = tup(:val, 1)
@tb.push t1
@tb.push t2
val = @tb.delete t1
assert_equal t1, val
assert_equal [t2], @tb.find_all(tem(:val, 1))
end
def test_delete_unless_alive
assert_equal [], @tb.delete_unless_alive
t1 = tup(:val, nil)
t2 = tup(:val, nil)
@tb.push t1
@tb.push t2
assert_equal [], @tb.delete_unless_alive
t1.cancel
assert_equal [t1], @tb.delete_unless_alive, 'canceled'
t2.renew Object.new
assert_equal [t2], @tb.delete_unless_alive, 'expired'
end
def test_find
template = tem(:val, nil)
assert_equal nil, @tb.find(template)
t1 = tup(:other, 1)
@tb.push t1
assert_equal nil, @tb.find(template)
t2 = tup(:val, 1)
@tb.push t2
assert_equal t2, @tb.find(template)
t2.cancel
assert_equal nil, @tb.find(template), 'canceled'
t3 = tup(:val, 3)
@tb.push t3
assert_equal t3, @tb.find(template)
t3.renew Object.new
assert_equal nil, @tb.find(template), 'expired'
end
def test_find_all
template = tem(:val, nil)
t1 = tup(:other, 1)
@tb.push t1
assert_equal [], @tb.find_all(template)
t2 = tup(:val, 2)
t3 = tup(:val, 3)
@tb.push t2
@tb.push t3
assert_equal [t2, t3], @tb.find_all(template)
t2.cancel
assert_equal [t3], @tb.find_all(template), 'canceled'
t3.renew Object.new
assert_equal [], @tb.find_all(template), 'expired'
end
def test_find_all_template
tuple = tup(:val, 1)
t1 = tem(:other, nil)
@tb.push t1
assert_equal [], @tb.find_all_template(tuple)
t2 = tem(:val, nil)
t3 = tem(:val, nil)
@tb.push t2
@tb.push t3
assert_equal [t2, t3], @tb.find_all_template(tuple)
t2.cancel
assert_equal [t3], @tb.find_all_template(tuple), 'canceled'
t3.renew Object.new
assert_equal [], @tb.find_all_template(tuple), 'expired'
end
def test_has_expires_eh
assert !@tb.has_expires?
t = tup(:val, 1)
@tb.push t
assert @tb.has_expires?
t.renew Object.new
assert !@tb.has_expires?
end
def test_push
t = tup(:val, 1)
@tb.push t
assert_equal t, @tb.find(tem(:val, 1))
end
##
# Create a tuple with +ary+ for its contents
def tup(*ary)
Rinda::TupleEntry.new ary
end
##
# Create a template with +ary+ for its contents
def tem(*ary)
Rinda::TemplateEntry.new ary
end
end