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
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
|
class Ractor
# Create a new Ractor with args and a block.
# args are passed via incoming channel.
# A block (Proc) will be isolated (can't access to outer variables)
#
# A ractor has default two channels:
# an incoming channel and an outgoing channel.
#
# Other ractors send objects to the ractor via the incoming channel and
# the ractor receives them.
# The ractor send objects via the outgoing channel and other ractors can
# receive them.
#
# The result of the block is sent via the outgoing channel
# and other
#
# r = Ractor.new do
# Ractor.receive # receive via r's mailbox => 1
# Ractor.receive # receive via r's mailbox => 2
# Ractor.yield 3 # yield a message (3) and wait for taking by another ractor.
# 'ok' # the return value will be yielded.
# # and r's incoming/outgoing ports are closed automatically.
# end
# r.send 1 # send a message (1) into r's mailbox.
# r << 2 # << is an alias of `send`.
# p r.take # take a message from r's outgoing port => 3
# p r.take # => 'ok'
# p r.take # raise Ractor::ClosedError
#
# other options:
# name: Ractor's name
#
def self.new(*args, name: nil, &block)
b = block # TODO: builtin bug
raise ArgumentError, "must be called with a block" unless block
loc = caller_locations(1, 1).first
loc = "#{loc.path}:#{loc.lineno}"
__builtin_ractor_create(loc, name, args, b)
end
# return current Ractor
def self.current
__builtin_cexpr! %q{
rb_ec_ractor_ptr(ec)->self
}
end
def self.count
__builtin_cexpr! %q{
ULONG2NUM(GET_VM()->ractor.cnt);
}
end
# Multiplex multiple Ractor communications.
#
# r, obj = Ractor.select(r1, r2)
# #=> wait for taking from r1 or r2
# # returned obj is a taken object from Ractor r
#
# r, obj = Ractor.select(r1, r2, Ractor.current)
# #=> wait for taking from r1 or r2
# # or receive from incoming queue
# # If receive is succeed, then obj is received value
# # and r is :receive (Ractor.current)
#
# r, obj = Ractor.select(r1, r2, Ractor.current, yield_value: obj)
# #=> wait for taking from r1 or r2
# # or receive from incoming queue
# # or yield (Ractor.yield) obj
# # If yield is succeed, then obj is nil
# # and r is :yield
#
def self.select(*ractors, yield_value: yield_unspecified = true, move: false)
raise ArgumentError, 'specify at least one ractor or `yield_value`' if yield_unspecified && ractors.empty?
__builtin_cstmt! %q{
const VALUE *rs = RARRAY_CONST_PTR_TRANSIENT(ractors);
VALUE rv;
VALUE v = ractor_select(ec, rs, RARRAY_LENINT(ractors),
yield_unspecified == Qtrue ? Qundef : yield_value,
(bool)RTEST(move) ? true : false, &rv);
return rb_ary_new_from_args(2, rv, v);
}
end
# Receive an incoming message from Ractor's incoming queue.
def self.receive
__builtin_cexpr! %q{
ractor_receive(ec, rb_ec_ractor_ptr(ec))
}
end
class << self
alias recv receive
end
private def receive
__builtin_cexpr! %q{
// TODO: check current actor
ractor_receive(ec, RACTOR_PTR(self))
}
end
alias recv receive
# Send a message to a Ractor's incoming queue.
#
# # Example:
# r = Ractor.new do
# p Ractor.receive #=> 'ok'
# end
# r.send 'ok' # send to r's incoming queue.
def send(obj, move: false)
__builtin_cexpr! %q{
ractor_send(ec, RACTOR_PTR(self), obj, move)
}
end
alias << send
# yield a message to the ractor's outgoing port.
def self.yield(obj, move: false)
__builtin_cexpr! %q{
ractor_yield(ec, rb_ec_ractor_ptr(ec), obj, move)
}
end
# Take a message from ractor's outgoing port.
#
# Example:
# r = Ractor.new{ 'oK' }
# p r.take #=> 'ok'
def take
__builtin_cexpr! %q{
ractor_take(ec, RACTOR_PTR(self))
}
end
def inspect
loc = __builtin_cexpr! %q{ RACTOR_PTR(self)->loc }
name = __builtin_cexpr! %q{ RACTOR_PTR(self)->name }
id = __builtin_cexpr! %q{ INT2FIX(RACTOR_PTR(self)->id) }
status = __builtin_cexpr! %q{
rb_str_new2(ractor_status_str(RACTOR_PTR(self)->status_))
}
"#<Ractor:##{id}#{name ? ' '+name : ''}#{loc ? " " + loc : ''} #{status}>"
end
def name
__builtin_cexpr! %q{ RACTOR_PTR(self)->name }
end
class RemoteError
attr_reader :ractor
end
# Closes the incoming port and returns its previous state.
def close_incoming
__builtin_cexpr! %q{
ractor_close_incoming(ec, RACTOR_PTR(self));
}
end
# Closes the outgoing port and returns its previous state.
def close_outgoing
__builtin_cexpr! %q{
ractor_close_outgoing(ec, RACTOR_PTR(self));
}
end
# utility method
def self.shareable? obj
__builtin_cexpr! %q{
rb_ractor_shareable_p(obj) ? Qtrue : Qfalse;
}
end
def self.make_shareable obj
__builtin_cexpr! %q{
rb_ractor_make_shareable(obj);
}
end
end
|
