Skip to content

Commit 75b8186

Browse files
pitr-chpitr-ch
committed
Update examples
1 parent 334e526 commit 75b8186

File tree

3 files changed

+104
-54
lines changed

3 files changed

+104
-54
lines changed

‎examples/init.rb

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,3 +3,5 @@
33
def do_stuff
44
:stuff
55
end
6+
7+
Concurrent.use_stdlib_logger Logger::DEBUG

‎examples/promises.in.rb

Lines changed: 34 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@
55

66
### Simple asynchronous task
77

8-
future = future { sleep 0.1; 1 + 1 } # evaluation starts immediately
8+
future = future(0.1) { |duration| sleep duration; :result } # evaluation starts immediately
99
future.completed?
1010
# block until evaluated
1111
future.value
@@ -21,6 +21,7 @@
2121
# re-raising
2222
raise future rescue $!
2323

24+
2425
### Direct creation of completed futures
2526

2627
succeeded_future(Object.new)
@@ -120,7 +121,7 @@
120121
### Completable Future and Event
121122

122123
future = completable_future
123-
event = event()
124+
event = completable_event()
124125

125126
# These threads will be blocked until the future and event is completed
126127
t1 = Thread.new { future.value } #
@@ -205,17 +206,40 @@
205206

206207

207208
# periodic task
208-
DONE = Concurrent::AtomicBoolean.new false
209-
210-
def schedule_job
211-
schedule(1) { do_stuff }.
212-
rescue { |e| StandardError === e ? report_error(e) : raise(e) }.
213-
then { schedule_job unless DONE.true? }
209+
def schedule_job(interval, &job)
210+
# schedule the first execution and chain restart og the job
211+
Concurrent.schedule(interval, &job).chain do |success, continue, reason|
212+
if success
213+
schedule_job(interval, &job) if continue
214+
else
215+
# handle error
216+
p reason
217+
# retry
218+
schedule_job(interval, &job)
219+
end
220+
end
214221
end
215222

216-
schedule_job
217-
DONE.make_true
223+
queue = Queue.new
224+
count = 0
225+
226+
schedule_job 0.05 do
227+
queue.push count
228+
count += 1
229+
# to continue scheduling return true, false will end the task
230+
if count < 4
231+
# to continue scheduling return true
232+
true
233+
else
234+
queue.push nil
235+
# to end the task return false
236+
false
237+
end
238+
end
218239

240+
# read the queue
241+
arr, v = [], nil; arr << v while (v = queue.pop) #
242+
arr
219243

220244
# How to limit processing where there are limited resources?
221245
# By creating an actor managing the resource

‎examples/promises.out.rb

Lines changed: 68 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -5,30 +5,31 @@
55

66
### Simple asynchronous task
77

8-
future = future { sleep 0.1; 1 + 1 } # evaluation starts immediately
9-
# => <#Concurrent::Promises::Future:0x7fc5cc1e5340 pending blocks:[]>
8+
future = future(0.1) { |duration| sleep duration; :result } # evaluation starts immediately
9+
# => <#Concurrent::Promises::Future:0x7fa602198e30 pending blocks:[]>
1010
future.completed? # => false
1111
# block until evaluated
12-
future.value # => 2
12+
future.value # => :result
1313
future.completed? # => true
1414

1515

1616
### Failing asynchronous task
1717

1818
future = future { raise 'Boom' }
19-
# => <#Concurrent::Promises::Future:0x7fc5cc1dc808 pending blocks:[]>
19+
# => <#Concurrent::Promises::Future:0x7fa602188a58 pending blocks:[]>
2020
future.value # => nil
2121
future.value! rescue $! # => #<RuntimeError: Boom>
2222
future.reason # => #<RuntimeError: Boom>
2323
# re-raising
2424
raise future rescue $! # => #<RuntimeError: Boom>
2525

26+
2627
### Direct creation of completed futures
2728

2829
succeeded_future(Object.new)
29-
# => <#Concurrent::Promises::Future:0x7fc5cc1c6030 success blocks:[]>
30+
# => <#Concurrent::Promises::Future:0x7fa60217bf10 success blocks:[]>
3031
failed_future(StandardError.new("boom"))
31-
# => <#Concurrent::Promises::Future:0x7fc5cc1c50b8 failed blocks:[]>
32+
# => <#Concurrent::Promises::Future:0x7fa60217aa48 failed blocks:[]>
3233

3334
### Chaining of futures
3435

@@ -68,7 +69,7 @@
6869
# => 3
6970

7071
failing_zip = succeeded_future(1) & failed_future(StandardError.new('boom'))
71-
# => <#Concurrent::Promises::Future:0x7fc5cc11ec90 failed blocks:[]>
72+
# => <#Concurrent::Promises::Future:0x7fa602129300 failed blocks:[]>
7273
failing_zip.result # => [false, [1, nil], [nil, #<StandardError: boom>]]
7374
failing_zip.then { |v| 'never happens' }.result # => [false, [1, nil], [nil, #<StandardError: boom>]]
7475
failing_zip.rescue { |a, b| (a || b).message }.value
@@ -81,28 +82,28 @@
8182

8283
# will not evaluate until asked by #value or other method requiring completion
8384
future = delay { 'lazy' }
84-
# => <#Concurrent::Promises::Future:0x7fc5cc0ff660 pending blocks:[]>
85+
# => <#Concurrent::Promises::Future:0x7fa60211a490 pending blocks:[]>
8586
sleep 0.1
8687
future.completed? # => false
8788
future.value # => "lazy"
8889

8990
# propagates trough chain allowing whole or partial lazy chains
9091

9192
head = delay { 1 }
92-
# => <#Concurrent::Promises::Future:0x7fc5cc0fc938 pending blocks:[]>
93+
# => <#Concurrent::Promises::Future:0x7fa602113898 pending blocks:[]>
9394
branch1 = head.then(&:succ)
94-
# => <#Concurrent::Promises::Future:0x7fc5cc0df068 pending blocks:[]>
95+
# => <#Concurrent::Promises::Future:0x7fa602112a60 pending blocks:[]>
9596
branch2 = head.delay.then(&:succ)
96-
# => <#Concurrent::Promises::Future:0x7fc5cc0dd178 pending blocks:[]>
97+
# => <#Concurrent::Promises::Future:0x7fa602111188 pending blocks:[]>
9798
join = branch1 & branch2
98-
# => <#Concurrent::Promises::Future:0x7fc5cc0dc430 pending blocks:[]>
99+
# => <#Concurrent::Promises::Future:0x7fa6021102b0 pending blocks:[]>
99100

100101
sleep 0.1 # nothing will complete # => 0
101102
[head, branch1, branch2, join].map(&:completed?) # => [false, false, false, false]
102103

103104
branch1.value # => 2
104105
sleep 0.1 # forces only head to complete, branch 2 stays incomplete
105-
# => 0
106+
# => 1
106107
[head, branch1, branch2, join].map(&:completed?) # => [true, true, false, false]
107108

108109
join.value # => [2, 2]
@@ -125,14 +126,14 @@
125126

126127
# it'll be executed after 0.1 seconds
127128
scheduled = schedule(0.1) { 1 }
128-
# => <#Concurrent::Promises::Future:0x7fc5caaae028 pending blocks:[]>
129+
# => <#Concurrent::Promises::Future:0x7fa60288a810 pending blocks:[]>
129130

130131
scheduled.completed? # => false
131132
scheduled.value # available after 0.1sec # => 1
132133

133134
# and in chain
134135
scheduled = delay { 1 }.schedule(0.1).then(&:succ)
135-
# => <#Concurrent::Promises::Future:0x7fc5caa9f2d0 pending blocks:[]>
136+
# => <#Concurrent::Promises::Future:0x7fa603101638 pending blocks:[]>
136137
# will not be scheduled until value is requested
137138
sleep 0.1
138139
scheduled.value # returns after another 0.1sec # => 2
@@ -141,36 +142,36 @@
141142
### Completable Future and Event
142143

143144
future = completable_future
144-
# => <#Concurrent::Promises::CompletableFuture:0x7fc5caa8eae8 pending blocks:[]>
145-
event = event()
146-
# => <#Concurrent::Promises::CompletableEvent:0x7fc5caa8d648 pending blocks:[]>
145+
# => <#Concurrent::Promises::CompletableFuture:0x7fa6030e0ac8 pending blocks:[]>
146+
event = completable_event()
147+
# => <#Concurrent::Promises::CompletableEvent:0x7fa6030e0a78 pending blocks:[]>
147148

148149
# These threads will be blocked until the future and event is completed
149150
t1 = Thread.new { future.value }
150151
t2 = Thread.new { event.wait }
151152

152153
future.success 1
153-
# => <#Concurrent::Promises::CompletableFuture:0x7fc5caa8eae8 success blocks:[]>
154+
# => <#Concurrent::Promises::CompletableFuture:0x7fa6030e0ac8 success blocks:[]>
154155
future.success 1 rescue $!
155156
# => #<Concurrent::MultipleAssignmentError: Future can be completed only once. Current result is [true, 1, nil], trying to set [true, 1, nil]>
156157
future.try_success 2 # => false
157158
event.complete
158-
# => <#Concurrent::Promises::CompletableEvent:0x7fc5caa8d648 completed blocks:[]>
159+
# => <#Concurrent::Promises::CompletableEvent:0x7fa6030e0a78 success blocks:[]>
159160

160161
# The threads can be joined now
161162
[t1, t2].each &:join
162163

163164

164165
### Callbacks
165166

166-
queue = Queue.new # => #<Thread::Queue:0x007fc5caa770c8>
167+
queue = Queue.new # => #<Thread::Queue:0x007fa6030ab5f8>
167168
future = delay { 1 + 1 }
168-
# => <#Concurrent::Promises::Future:0x7fc5caa754d0 pending blocks:[]>
169+
# => <#Concurrent::Promises::Future:0x7fa6030aa888 pending blocks:[]>
169170

170171
future.on_success { queue << 1 } # evaluated asynchronously
171-
# => <#Concurrent::Promises::Future:0x7fc5caa754d0 pending blocks:[]>
172+
# => <#Concurrent::Promises::Future:0x7fa6030aa888 pending blocks:[]>
172173
future.on_success! { queue << 2 } # evaluated on completing thread
173-
# => <#Concurrent::Promises::Future:0x7fc5caa754d0 pending blocks:[]>
174+
# => <#Concurrent::Promises::Future:0x7fa6030aa888 pending blocks:[]>
174175

175176
queue.empty? # => true
176177
future.value # => 2
@@ -188,15 +189,15 @@
188189
# executed on executor for blocking and long operations
189190
then_on(:io) { File.read __FILE__ }.
190191
wait
191-
# => <#Concurrent::Promises::Future:0x7fc5cb010b10 success blocks:[]>
192+
# => <#Concurrent::Promises::Future:0x7fa60307abb0 success blocks:[]>
192193

193194

194195
### Interoperability with actors
195196

196197
actor = Concurrent::Actor::Utils::AdHoc.spawn :square do
197198
-> v { v ** 2 }
198199
end
199-
# => #<Concurrent::Actor::Reference:0x7fc5caa35268 /square (Concurrent::Actor::Utils::AdHoc)>
200+
# => #<Concurrent::Actor::Reference:0x7fa601ab8ea8 /square (Concurrent::Actor::Utils::AdHoc)>
200201

201202

202203
future { 2 }.
@@ -210,51 +211,74 @@
210211
### Interoperability with channels
211212

212213
ch1 = Concurrent::Channel.new
213-
# => #<Concurrent::Channel:0x007fc5ca9f4448 @buffer=#<Concurrent::Channel::Buffer::Unbuffered:0x007fc5ca9f41f0 @__lock__=#<Mutex:0x007fc5ca9f4128>, @__condition__=#<Thread::ConditionVariable:0x007fc5ca9f4088>, @closed=false, @size=0, @capacity=1, @buffer=nil, @putting=[], @taking=[]>, @validator=#<Proc:0x007fc5cc0ce970@/Users/pitr/Workspace/public/concurrent-ruby/lib/concurrent/channel.rb:28 (lambda)>>
214+
# => #<Concurrent::Channel:0x007fa601a6a2d0 @buffer=#<Concurrent::Channel::Buffer::Unbuffered:0x007fa601a6a230 @__lock__=#<Mutex:0x007fa601a6a168>, @__condition__=#<Thread::ConditionVariable:0x007fa601a6a140>, @closed=false, @size=0, @capacity=1, @buffer=nil, @putting=[], @taking=[]>, @validator=#<Proc:0x007fa6028b12d0@/Users/pitr/Workspace/public/concurrent-ruby/lib/concurrent/channel.rb:28 (lambda)>>
214215
ch2 = Concurrent::Channel.new
215-
# => #<Concurrent::Channel:0x007fc5cc89a528 @buffer=#<Concurrent::Channel::Buffer::Unbuffered:0x007fc5cc89a2a8 @__lock__=#<Mutex:0x007fc5cc89a140>, @__condition__=#<Thread::ConditionVariable:0x007fc5cc89a078>, @closed=false, @size=0, @capacity=1, @buffer=nil, @putting=[], @taking=[]>, @validator=#<Proc:0x007fc5cc0ce970@/Users/pitr/Workspace/public/concurrent-ruby/lib/concurrent/channel.rb:28 (lambda)>>
216+
# => #<Concurrent::Channel:0x007fa601a69380 @buffer=#<Concurrent::Channel::Buffer::Unbuffered:0x007fa601a69308 @__lock__=#<Mutex:0x007fa601a69268>, @__condition__=#<Thread::ConditionVariable:0x007fa601a69218>, @closed=false, @size=0, @capacity=1, @buffer=nil, @putting=[], @taking=[]>, @validator=#<Proc:0x007fa6028b12d0@/Users/pitr/Workspace/public/concurrent-ruby/lib/concurrent/channel.rb:28 (lambda)>>
216217

217218
result = select(ch1, ch2)
218-
# => <#Concurrent::Promises::Future:0x7fc5cc892e40 pending blocks:[]>
219+
# => <#Concurrent::Promises::Future:0x7fa601a62d50 pending blocks:[]>
219220
ch1.put 1 # => true
220221
result.value!
221-
# => [1, #<Concurrent::Channel:0x007fc5ca9f4448 @buffer=#<Concurrent::Channel::Buffer::Unbuffered:0x007fc5ca9f41f0 @__lock__=#<Mutex:0x007fc5ca9f4128>, @__condition__=#<Thread::ConditionVariable:0x007fc5ca9f4088>, @closed=false, @size=0, @capacity=1, @buffer=nil, @putting=[], @taking=[]>, @validator=#<Proc:0x007fc5cc0ce970@/Users/pitr/Workspace/public/concurrent-ruby/lib/concurrent/channel.rb:28 (lambda)>>]
222+
# => [1, #<Concurrent::Channel:0x007fa601a6a2d0 @buffer=#<Concurrent::Channel::Buffer::Unbuffered:0x007fa601a6a230 @__lock__=#<Mutex:0x007fa601a6a168>, @__condition__=#<Thread::ConditionVariable:0x007fa601a6a140>, @closed=false, @size=0, @capacity=1, @buffer=nil, @putting=[], @taking=[]>, @validator=#<Proc:0x007fa6028b12d0@/Users/pitr/Workspace/public/concurrent-ruby/lib/concurrent/channel.rb:28 (lambda)>>]
222223

223224

224225
future { 1+1 }.
225226
then_put(ch1)
226-
# => <#Concurrent::Promises::Future:0x7fc5cc87b920 pending blocks:[]>
227+
# => <#Concurrent::Promises::Future:0x7fa601a51910 pending blocks:[]>
227228
result = future { '%02d' }.
228229
then_select(ch1, ch2).
229230
then { |format, (value, channel)| format format, value }
230-
# => <#Concurrent::Promises::Future:0x7fc5cc862f60 pending blocks:[]>
231+
# => <#Concurrent::Promises::Future:0x7fa601a2b008 pending blocks:[]>
231232
result.value! # => "02"
232233

233234

234235
### Common use-cases Examples
235236

236237
# simple background processing
237238
future { do_stuff }
238-
# => <#Concurrent::Promises::Future:0x7fc5cc080518 pending blocks:[]>
239+
# => <#Concurrent::Promises::Future:0x7fa601a1b478 pending blocks:[]>
239240

240241
# parallel background processing
241242
jobs = 10.times.map { |i| future { i } }
242243
zip(*jobs).value # => [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
243244

244245

245246
# periodic task
246-
DONE = Concurrent::AtomicBoolean.new false # => #<Concurrent::AtomicBoolean:0x007fc5cc079a38>
247-
248-
def schedule_job
249-
schedule(1) { do_stuff }.
250-
rescue { |e| StandardError === e ? report_error(e) : raise(e) }.
251-
then { schedule_job unless DONE.true? }
247+
def schedule_job(interval, &job)
248+
# schedule the first execution and chain restart og the job
249+
Concurrent.schedule(interval, &job).chain do |success, continue, reason|
250+
if success
251+
schedule_job(interval, &job) if continue
252+
else
253+
# handle error
254+
p reason
255+
# retry
256+
schedule_job(interval, &job)
257+
end
258+
end
252259
end # => :schedule_job
253260

254-
schedule_job
255-
# => <#Concurrent::Promises::Future:0x7fc5ca9949d0 pending blocks:[]>
256-
DONE.make_true # => true
261+
queue = Queue.new # => #<Thread::Queue:0x007fa6020d2cf8>
262+
count = 0 # => 0
263+
264+
schedule_job 0.05 do
265+
queue.push count
266+
count += 1
267+
# to continue scheduling return true, false will end the task
268+
if count < 4
269+
# to continue scheduling return true
270+
true
271+
else
272+
queue.push nil
273+
# to end the task return false
274+
false
275+
end
276+
end
277+
# => <#Concurrent::Promises::Future:0x7fa6020c23d0 pending blocks:[]>
257278

279+
# read the queue
280+
arr, v = [], nil; arr << v while (v = queue.pop)
281+
arr # => [0, 1, 2, 3]
258282

259283
# How to limit processing where there are limited resources?
260284
# By creating an actor managing the resource
@@ -265,7 +289,7 @@ def schedule_job
265289
data[message]
266290
end
267291
end
268-
# => #<Concurrent::Actor::Reference:0x7fc5cc843318 /db (Concurrent::Actor::Utils::AdHoc)>
292+
# => #<Concurrent::Actor::Reference:0x7fa6019b8788 /db (Concurrent::Actor::Utils::AdHoc)>
269293

270294
concurrent_jobs = 11.times.map do |v|
271295

@@ -295,7 +319,7 @@ def schedule_job
295319
end
296320
end
297321
end
298-
# => #<Concurrent::Actor::Reference:0x7fc5ca89a160 /DB-pool (Concurrent::Actor::Utils::Pool)>
322+
# => #<Concurrent::Actor::Reference:0x7fa60284b278 /DB-pool (Concurrent::Actor::Utils::Pool)>
299323

300324
concurrent_jobs = 11.times.map do |v|
301325

0 commit comments

Comments
 (0)