@@ -53,28 +53,6 @@ FutureRegistry <- local({
5353 on.exit(mdebugf_pop())
5454 }
5555
56- futuresDB <- db [[where ]]
57- drop <- integer(0L )
58-
59- on.exit({
60- # # Clean out collected futures
61- if (length(drop ) > 0 ) {
62- if (debug ) {
63- mdebug_push(" Remove collected futures ..." )
64- mdebugf(" Indices of futures to drop: [n=%d] %s" , length(drop ), commaq(drop ))
65- }
66- futuresDB <- futuresDB [- drop ]
67- db [[where ]] <<- futuresDB
68- if (debug ) mdebug_pop()
69- } else {
70- if (debug ) mdebug(" Indices of futures to drop: [n=0] <none>" )
71- }
72- }, add = TRUE )
73-
74- if (debug ) {
75- on.exit(mdebugf_pop(), add = TRUE )
76- }
77-
7856 for (ii in seq_along(futures )) {
7957 future <- futures [[ii ]]
8058
@@ -118,31 +96,10 @@ FutureRegistry <- local({
11896
11997 # # (b) Make sure future is removed from registry, unless
12098 # # already done via above value() call.
121- idx <- indexOf(futuresDB , future = future )
99+ idx <- indexOf(db [[ where ]] , future = future )
122100 if (! is.na(idx )) {
123- drop <- c(drop , idx )
124-
125- # # Update backend
126- backend <- future [[" backend" ]]
127- if (! is.null(backend )) {
128- # # Update counters
129- counters <- backend [[" counters" ]]
130- counters [" finished" ] <- counters [" finished" ] + 1L
131- backend [[" counters" ]] <- counters
132-
133- # # Update total runtime
134- result <- future [[" result" ]]
135- if (inherits(result , " FutureResult" )) {
136- times <- result [c(" finished" , " started" )]
137- if (length(times ) == 2L ) {
138- dt <- times [[" finished" ]] - times [[" started" ]]
139- runtime <- backend [[" runtime" ]]
140- runtime <- runtime + dt
141- backend [[" runtime" ]] <- runtime
142- }
143- }
144- }
145- } # # if (!is.na(idx))
101+ FutureRegistry(where , action = " remove" , future = future , earlySignal = FALSE , debug = debug )
102+ }
146103
147104 # # (c) Collect only the first resolved future?
148105 if (firstOnly ) {
@@ -235,11 +192,14 @@ FutureRegistry <- local({
235192 }
236193 } else if (action == " collect-first" ) {
237194 collectValues(where , futures = futures , firstOnly = TRUE , debug = debug )
195+ futures <- db [[where ]]
238196 } else if (action == " collect-all" ) {
239197 collectValues(where , futures = futures , firstOnly = FALSE , debug = debug )
198+ futures <- db [[where ]]
240199 earlySignal <- FALSE # # Avoid second round of result collection
241200 } else if (action == " reset" ) {
242- db [[where ]] <<- list ()
201+ futures <- list ()
202+ db [[where ]] <<- futures
243203 if (debug ) mdebug(" Erased registry" )
244204 } else if (action == " list" ) {
245205 if (debug ) mdebug(" Listing all futures" )
@@ -263,6 +223,7 @@ FutureRegistry <- local({
263223 # # Collect values, which will trigger signaling during
264224 # # calls to resolved().
265225 collectValues(where , futures = futures [idxs ], firstOnly = FALSE )
226+ futures <- db [[where ]]
266227 }
267228 if (debug ) mdebugf(" Early signaling of %d future candidates ... done" , length(futures ))
268229 }
0 commit comments