1
1
### Simple asynchronous task
2
2
3
3
future = Concurrent . future { sleep 0.1 ; 1 + 1 } # evaluation starts immediately
4
- # => <#Concurrent::Edge::Future:0x7fd41a956d70 pending blocks:[]>
4
+ # => <#Concurrent::Edge::Future:0x7fc6218f2318 pending blocks:[]>
5
5
future . completed? # => false
6
6
# block until evaluated
7
7
future . value # => 2
11
11
### Failing asynchronous task
12
12
13
13
future = Concurrent . future { raise 'Boom' }
14
- # => <#Concurrent::Edge::Future:0x7fd41a946c18 failed blocks:[]>
14
+ # => <#Concurrent::Edge::Future:0x7fc6218eae38 pending blocks:[]>
15
15
future . value # => nil
16
16
future . value! rescue $! # => #<RuntimeError: Boom>
17
17
future . reason # => #<RuntimeError: Boom>
41
41
Concurrent . future { 1 } . then ( &:succ ) . rescue { |e | e . message } . then ( &:succ ) . value
42
42
# => 3
43
43
44
+ failing_zip = Concurrent . completed_future ( 1 ) & Concurrent . future { raise 'boom' }
45
+ # => <#Concurrent::Edge::Future:0x7fc6218b0f08 pending blocks:[]>
46
+ failing_zip . result # => [false, [1, nil], [nil, #<RuntimeError: boom>]]
47
+ failing_zip . then { |v | 'never happens' } . result # => [false, [1, nil], [nil, #<RuntimeError: boom>]]
48
+ failing_zip . rescue { |a , b | ( a || b ) . message } . value
49
+ # => "boom"
50
+ failing_zip . chain { |success , values , reasons | [ success , values . compact , reasons . compactß ] } . value
51
+ # => nil
44
52
45
53
### Delay
46
54
47
55
# will not evaluate until asked by #value or other method requiring completion
48
56
future = Concurrent . delay { 'lazy' }
49
- # => <#Concurrent::Edge::Future:0x7fd41a8f6ec0 pending blocks:[]>
57
+ # => <#Concurrent::Edge::Future:0x7fc6218a37e0 pending blocks:[]>
50
58
sleep 0.1
51
59
future . completed? # => false
52
60
future . value # => "lazy"
53
61
54
62
# propagates trough chain allowing whole or partial lazy chains
55
63
56
64
head = Concurrent . delay { 1 }
57
- # => <#Concurrent::Edge::Future:0x7fd41a8ee1a8 pending blocks:[]>
65
+ # => <#Concurrent::Edge::Future:0x7fc6218a0720 pending blocks:[]>
58
66
branch1 = head . then ( &:succ )
59
- # => <#Concurrent::Edge::Future:0x7fd41a8ed438 pending blocks:[]>
67
+ # => <#Concurrent::Edge::Future:0x7fc6212c7b50 pending blocks:[]>
60
68
branch2 = head . delay . then ( &:succ )
61
- # => <#Concurrent::Edge::Future:0x7fd41a8e7588 pending blocks:[]>
69
+ # => <#Concurrent::Edge::Future:0x7fc6212c6098 pending blocks:[]>
62
70
join = branch1 & branch2
63
- # => <#Concurrent::Edge::Future:0x7fd41a8e4ec8 pending blocks:[]>
71
+ # => <#Concurrent::Edge::Future:0x7fc6212c4f40 pending blocks:[]>
64
72
65
73
sleep 0.1 # nothing will complete # => 0
66
74
[ head , branch1 , branch2 , join ] . map ( &:completed? ) # => [false, false, false, false]
88
96
### Schedule
89
97
90
98
scheduled = Concurrent . schedule ( 0.1 ) { 1 }
91
- # => <#Concurrent::Edge::Future:0x7fd41a8a4468 pending blocks:[]>
99
+ # => <#Concurrent::Edge::Future:0x7fc62128c550 pending blocks:[]>
92
100
93
101
scheduled . completed? # => false
94
102
scheduled . value # available after 0.1sec # => 1
95
103
96
104
# and in chain
97
105
scheduled = Concurrent . delay { 1 } . schedule ( 0.1 ) . then ( &:succ )
98
- # => <#Concurrent::Edge::Future:0x7fd41a895828 pending blocks:[]>
106
+ # => <#Concurrent::Edge::Future:0x7fc6228bcdc0 pending blocks:[]>
99
107
# will not be scheduled until value is requested
100
108
sleep 0.1
101
109
scheduled . value # returns after another 0.1sec # => 2
104
112
### Completable Future and Event
105
113
106
114
future = Concurrent . future
107
- # => <#Concurrent::Edge::CompletableFuture:0x7fd41a87c648 pending blocks:[]>
115
+ # => <#Concurrent::Edge::CompletableFuture:0x7fc623083720 pending blocks:[]>
108
116
event = Concurrent . event
109
- # => <#Concurrent::Edge::CompletableEvent:0x7fd41a8770d0 pending blocks:[]>
117
+ # => <#Concurrent::Edge::CompletableEvent:0x7fc623081100 pending blocks:[]>
110
118
# Don't forget to keep the reference, `Concurrent.future.then { |v| v }` is incompletable
111
119
112
120
# will be blocked until completed
113
121
t1 = Thread . new { future . value }
114
122
t2 = Thread . new { event . wait }
115
123
116
124
future . success 1
117
- # => <#Concurrent::Edge::CompletableFuture:0x7fd41a87c648 success blocks:[]>
125
+ # => <#Concurrent::Edge::CompletableFuture:0x7fc623083720 success blocks:[]>
118
126
future . success 1 rescue $!
119
127
# => #<Concurrent::MultipleAssignmentError: Future can be completed only once. Current result is [true, 1, nil], trying to set [true, 1, nil]>
120
128
future . try_success 2 # => false
121
129
event . complete
122
- # => <#Concurrent::Edge::CompletableEvent:0x7fd41a8770d0 completed blocks:[]>
130
+ # => <#Concurrent::Edge::CompletableEvent:0x7fc623081100 completed blocks:[]>
123
131
124
132
[ t1 , t2 ] . each &:join
125
133
126
134
127
135
### Callbacks
128
136
129
- queue = Queue . new # => #<Thread::Queue:0x007fd41b0543e8 >
137
+ queue = Queue . new # => #<Thread::Queue:0x007fc62127df00 >
130
138
future = Concurrent . delay { 1 + 1 }
131
- # => <#Concurrent::Edge::Future:0x7fd41b04f028 pending blocks:[]>
139
+ # => <#Concurrent::Edge::Future:0x7fc62127c060 pending blocks:[]>
132
140
133
141
future . on_success { queue << 1 } # evaluated asynchronously
134
- # => <#Concurrent::Edge::Future:0x7fd41b04f028 pending blocks:[]>
142
+ # => <#Concurrent::Edge::Future:0x7fc62127c060 pending blocks:[]>
135
143
future . on_success! { queue << 2 } # evaluated on completing thread
136
- # => <#Concurrent::Edge::Future:0x7fd41b04f028 pending blocks:[]>
144
+ # => <#Concurrent::Edge::Future:0x7fc62127c060 pending blocks:[]>
137
145
138
146
queue . empty? # => true
139
147
future . value # => 2
144
152
### Thread-pools
145
153
146
154
Concurrent . future ( :fast ) { 2 } . then ( :io ) { File . read __FILE__ } . wait
147
- # => <#Concurrent::Edge::Future:0x7fd41a857550 success blocks:[]>
155
+ # => <#Concurrent::Edge::Future:0x7fc62125d9a8 success blocks:[]>
148
156
149
157
150
158
### Interoperability with actors
151
159
152
160
actor = Concurrent ::Actor ::Utils ::AdHoc . spawn :square do
153
161
-> v { v ** 2 }
154
162
end
155
- # => #<Concurrent::Actor::Reference:0x7fd41c03b5b8 /square (Concurrent::Actor::Utils::AdHoc)>
163
+ # => #<Concurrent::Actor::Reference:0x7fc621234a30 /square (Concurrent::Actor::Utils::AdHoc)>
156
164
157
165
Concurrent .
158
166
future { 2 } .
165
173
166
174
### Interoperability with channels
167
175
168
- ch1 = Concurrent ::Edge ::Channel . new # => #<Concurrent::Edge::Channel:0x007fd41a297458 >
169
- ch2 = Concurrent ::Edge ::Channel . new # => #<Concurrent::Edge::Channel:0x007fd41a296940 >
176
+ ch1 = Concurrent ::Edge ::Channel . new # => #<Concurrent::Edge::Channel:0x007fc621205460 >
177
+ ch2 = Concurrent ::Edge ::Channel . new # => #<Concurrent::Edge::Channel:0x007fc6212041c8 >
170
178
171
179
result = Concurrent . select ( ch1 , ch2 )
172
- # => <#Concurrent::Edge::CompletableFuture:0x7fd41a295c98 pending blocks:[]>
180
+ # => <#Concurrent::Edge::CompletableFuture:0x7fc6211fe5c0 pending blocks:[]>
173
181
ch1 . push 1 # => nil
174
182
result . value!
175
- # => [1, #<Concurrent::Edge::Channel:0x007fd41a297458 >]
183
+ # => [1, #<Concurrent::Edge::Channel:0x007fc621205460 >]
176
184
177
185
Concurrent .
178
186
future { 1 +1 } .
179
187
then_push ( ch1 )
180
- # => <#Concurrent::Edge::Future:0x7fd41a284010 pending blocks:[]>
188
+ # => <#Concurrent::Edge::Future:0x7fc6211f7338 pending blocks:[]>
181
189
result = Concurrent .
182
190
future { '%02d' } .
183
191
then_select ( ch1 , ch2 ) .
184
192
then { |format , ( value , channel ) | format format , value }
185
- # => <#Concurrent::Edge::Future:0x7fd41a25d938 pending blocks:[]>
193
+ # => <#Concurrent::Edge::Future:0x7fc6211ec668 pending blocks:[]>
186
194
result . value! # => "02"
187
195
188
196
189
197
### Common use-cases Examples
190
198
191
199
# simple background processing
192
200
Concurrent . future { do_stuff }
193
- # => <#Concurrent::Edge::Future:0x7fd41a23d520 pending blocks:[]>
201
+ # => <#Concurrent::Edge::Future:0x7fc6211df170 pending blocks:[]>
194
202
195
203
# parallel background processing
196
204
jobs = 10 . times . map { |i | Concurrent . future { i } }
@@ -207,7 +215,7 @@ def schedule_job
207
215
end # => :schedule_job
208
216
209
217
schedule_job
210
- # => <#Concurrent::Edge::Future:0x7fd41a2245e8 pending blocks:[]>
218
+ # => <#Concurrent::Edge::Future:0x7fc62119c140 pending blocks:[]>
211
219
@end = true # => true
212
220
213
221
@@ -220,7 +228,7 @@ def schedule_job
220
228
data [ message ]
221
229
end
222
230
end
223
- # => #<Concurrent::Actor::Reference:0x7fd41a206228 /db (Concurrent::Actor::Utils::AdHoc)>
231
+ # => #<Concurrent::Actor::Reference:0x7fc62117f568 /db (Concurrent::Actor::Utils::AdHoc)>
224
232
225
233
concurrent_jobs = 11 . times . map do |v |
226
234
Concurrent .
@@ -250,7 +258,7 @@ def schedule_job
250
258
end
251
259
end
252
260
end
253
- # => #<Concurrent::Actor::Reference:0x7fd41a0530c0 /DB-pool (Concurrent::Actor::Utils::Pool)>
261
+ # => #<Concurrent::Actor::Reference:0x7fc6218969f0 /DB-pool (Concurrent::Actor::Utils::Pool)>
254
262
255
263
concurrent_jobs = 11 . times . map do |v |
256
264
Concurrent .
0 commit comments