Update docs and fix exception handling on the final batch

Author: richiejp

base/asyncmap.jl

@@ -11,14 +11,13 @@ Uses multiple concurrent tasks to map `f` over a collection (or multiple
 equal length collections). For multiple collection arguments, `f` is
 applied elementwise.
 
-`ntasks` specifies the number of tasks to run concurrently.
-Depending on the length of the collections, if `ntasks` is unspecified,
-up to 100 tasks will be used for concurrent mapping.
+`ntasks` specifies the maximum number of tasks permitted to run
+concurrently. the default is 100.
 
-`ntasks` can also be specified as a zero-arg function. In this case, the
-number of tasks to run in parallel is checked before processing every element and a new
-task started if the value of `ntasks_func` is less than the current number
-of tasks.
+`ntasks` can also be specified as a zero-arg function. This is called before
+each task is scheduled (each iteration) to be used as the new
+maximum. Changing the maximum will not effect tasks that have already been
+scheduled, but it may effect the scheduling of new tasks.
 
 If `batch_size` is specified, the collection is processed in batch mode. `f` must
 then be a function that must accept a `Vector` of argument tuples and must
@@ -54,50 +53,62 @@ julia> result = asyncmap(1:2) do x
 The following examples highlight execution in different tasks by returning
 the `objectid` of the tasks in which the mapping function is executed.
 
-First, with `ntasks` undefined, each element is processed in a different task.
+First, with `ntasks` undefined, all three iterations can be started at once.
 ```
-julia> tskoid() = objectid(current_task());
-
-julia> asyncmap(x->tskoid(), 1:5)
-5-element Array{UInt64,1}:
- 0x6e15e66c75c75853
- 0x440f8819a1baa682
- 0x9fb3eeadd0c83985
- 0xebd3e35fe90d4050
- 0x29efc93edce2b961
-
-julia> length(unique(asyncmap(x->tskoid(), 1:5)))
-5
+julia> asyncmap(1:3) do i
+         println("before sleep \$i")
+         sleep(1/i)
+         println("after sleep \$i")
+         i
+       end
+before sleep 1
+before sleep 2
+before sleep 3
+after sleep 3
+after sleep 2
+after sleep 1
+3-element Array{Int64,1}:
+ 1
+ 2
+ 3
 ```
 
-With `ntasks=2` all elements are processed in 2 tasks.
+With `ntasks=2` the third iteration must wait until the second
+completes. However it can still start and finish before the first completes.
 ```
-julia> asyncmap(x->tskoid(), 1:5; ntasks=2)
-5-element Array{UInt64,1}:
- 0x027ab1680df7ae94
- 0xa23d2f80cd7cf157
- 0x027ab1680df7ae94
- 0xa23d2f80cd7cf157
- 0x027ab1680df7ae94
-
-julia> length(unique(asyncmap(x->tskoid(), 1:5; ntasks=2)))
-2
+julia> asyncmap(1:3; ntasks=2) do i
+         println("before sleep \$i")
+         sleep(1/i)
+         println("after sleep \$i")
+         i
+       end
+before sleep 1
+before sleep 2
+after sleep 2
+before sleep 3
+after sleep 3
+after sleep 1
+3-element Array{Int64,1}:
+ 1
+ 2
+ 3
 ```
 
 With `batch_size` defined, the mapping function needs to be changed to accept an array
 of argument tuples and return an array of results. `map` is used in the modified mapping
 function to achieve this.
 ```
-julia> batch_func(input) = map(x->string("args_tuple: ", x, ", element_val: ", x[1], ", task: ", tskoid()), input)
-batch_func (generic function with 1 method)
-
-julia> asyncmap(batch_func, 1:5; ntasks=2, batch_size=2)
+julia> asyncmap(1:5; ntasks=2, batch_size=2) do input
+         map(input) do x
+           string("args_tuple: ", x, ", element_val: ", x[1], ", ", current_task())
+         end
+       end
 5-element Array{String,1}:
- "args_tuple: (1,), element_val: 1, task: 9118321258196414413"
- "args_tuple: (2,), element_val: 2, task: 4904288162898683522"
- "args_tuple: (3,), element_val: 3, task: 9118321258196414413"
- "args_tuple: (4,), element_val: 4, task: 4904288162898683522"
- "args_tuple: (5,), element_val: 5, task: 9118321258196414413"
+ "args_tuple: 1, element_val: 1, Task (runnable) @0x00007f6629271390"
+ "args_tuple: 2, element_val: 2, Task (runnable) @0x00007f6629271390"
+ "args_tuple: 3, element_val: 3, Task (runnable) @0x00007f6629271600"
+ "args_tuple: 4, element_val: 4, Task (runnable) @0x00007f6629271600"
+ "args_tuple: 5, element_val: 5, Task (runnable) @0x00007f6629271870"
 ```
 
 !!! note
@@ -208,7 +219,7 @@ function do_asyncmap(f, c::Tuple, ntasks::Function, batch_size)
 
     new_batch()
     res = try
-        map(c...) do x
+        rs = map(c...) do x
             r = Ref{Any}(undef)
 
             push!(batch_in, x)
@@ -221,10 +232,13 @@ function do_asyncmap(f, c::Tuple, ntasks::Function, batch_size)
 
             r
         end
+
+        do_asyncmap_task!(exec_batch, s, batch_in, batch_out)
+
+        rs
     catch ex
         ex isa AbortMapException || rethrow()
     end
-    do_asyncmap_task!(exec_batch, s, batch_in, batch_out)
 
     wait_done(s)