{"action":"created","issue":{"url":"https://api.github.com/repos/JuliaLang/julia/issues/14843","repository_url":"https://api.github.com/repos/JuliaLang/julia","labels_url":"https://api.github.com/repos/JuliaLang/julia/issues/14843/labels{/name}","comments_url":"https://api.github.com/repos/JuliaLang/julia/issues/14843/comments","events_url":"https://api.github.com/repos/JuliaLang/julia/issues/14843/events","html_url":"https://github.com/JuliaLang/julia/issues/14843","id":129634809,"number":14843,"title":"RFC: Simplifying and generalising pmap ","user":{"login":"samoconnor","id":8266283,"avatar_url":"https://avatars.githubusercontent.com/u/8266283?v=3","gravatar_id":"","url":"https://api.github.com/users/samoconnor","html_url":"https://github.com/samoconnor","followers_url":"https://api.github.com/users/samoconnor/followers","following_url":"https://api.github.com/users/samoconnor/following{/other_user}","gists_url":"https://api.github.com/users/samoconnor/gists{/gist_id}","starred_url":"https://api.github.com/users/samoconnor/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/samoconnor/subscriptions","organizations_url":"https://api.github.com/users/samoconnor/orgs","repos_url":"https://api.github.com/users/samoconnor/repos","events_url":"https://api.github.com/users/samoconnor/events{/privacy}","received_events_url":"https://api.github.com/users/samoconnor/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/JuliaLang/julia/labels/parallel","name":"parallel","color":"f7e102"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":7,"created_at":"2016-01-29T01:23:32Z","updated_at":"2016-02-10T23:17:59Z","closed_at":null,"body":"_This issue covers some of the same ground as https://github.com/JuliaLang/julia/pull/12943  (stale), https://github.com/JuliaLang/julia/issues/14515 and https://github.com/JuliaLang/julia/pull/14736. My intention is to present an overall `pmap` refactoring plan here for comment and submit new PRs if it is seen as useful._\r\n\r\n### `pmap` Features\r\n\r\nThe current `pmap` implementation has the following features:\r\n\r\n1. Using `@sync`/`@async` to run mapped function in parallel and collect results.\r\n2. Automatic allocation of idle workers to `remotecall_wait` calls.\r\n3. Not reusing workers believed to be faulty.\r\n4. Option to retry on error.\r\n5. Option to continue mapping despite errors (returning exceptions in the result array).\r\n\r\nThese are all useful features. However, it seems to me that `pmap` currently tries to do too much and that many of these features would be useful in contexts other than `pmap`:\r\n\r\n - \"1.\" is useful when using RPC mechanisms other than `remotecall` ( e.g. HTTP, `readstring(::Cmd)` or AWS Lambda).\r\n - \"2.\" and \"3.\" are probably useful to other users of the `remotecall` mechanism.\r\n - If \"4.\" and \"5.\" are useful with `pmap` they should also be useful with ordinary `map`.\r\n\r\n### Proposed Separation of Features\r\n\r\n#### `amap`\r\n\r\n[`amap`](https://github.com/samoconnor/AsyncMap.jl/blob/master/src/AsyncMap.jl#L22) adds `@sync`/`@async` to regular `map`.  (feature \"1.\")\r\n```julia\r\n# Run `::Cmd`s in parallel...\r\njulia> @time map(t->run(`sleep $t`), 1:3)\r\n6 seconds\r\njulia> @time amap(t->run(`sleep $t`), 1:3)\r\n3 seconds\r\n\r\n# Run downloads in parallel...\r\njulia> url_data = amap(download, urls))\r\n```\r\n\r\n#### `WorkerPool` and `remote`\r\n\r\nA `WorkerPool` keeps track of which workers are busy. `take!(default_worker_pool())` yeilds a worker that is not already busy doing `remotecall_wait`, waiting if needed (feature \"2\".). If there is a \r\nreliable way to identify \"faulty\" workers (feature \"3.\") then `worker` will not return fault workers. (Partial implementation in comments of: https://github.com/JuliaLang/julia/pull/14736)\r\n\r\n`remote` takes a function and returns a lambda that executes the function on a remote worker.\r\n\r\nUsing `amap` and `remote` together...\r\n\r\n```julia\r\npmap(f, c...) = amap(remote(f), c...)\r\n```\r\n\r\n```julia\r\nfunction remotecall_fetch(f, args...)\r\n    remotecall_fetch(f, default_worker_pool(), args...)\r\nend\r\nremote(f::Function) = (args...)->remotecall_fetch(f, args...)\r\n```\r\n\r\n\r\n#### `@catch`\r\n\r\n`@catch` takes a function can returns a lambda that catches any exceptions thrown by the function and returns the exception object. \r\n\r\n`pmap(f, v; err_stop=false)` can be replaced with `pmap(@catch(f), v)` (feature \"5.\")\r\n\r\n```julia\r\nmacro catch(f)\r\n    :((args...) -> try $f(args...) catch ex; ex end)\r\nend\r\n```\r\n\r\n\r\n#### `retry`\r\n\r\n`retry` takes a function and returns a lambda that retries the function if an error occurs (feature \"4.\").\r\n\r\n`pmap(f, v; err_retry=true)` can be replaced with `pmap(retry(f), v)`, or for more granular error handling `pmap(retry(f, e -> isa(e, NetworkTimeoutError)), v)`\r\n\r\n```julia\r\nfunction retry(f::Function, condition::Function=(e)->true, n::Integer=3)\r\n    (args...) -> @repeat n try\r\n        f(args...)\r\n    catch e\r\n        @retry if condition(e) end\r\n    end\r\nend\r\n```\r\n_(`@repeat` and `@retry` are implemented in https://github.com/samoconnor/Retry.jl)_\r\n\r\n### `pmap` Defaults\r\n\r\n `pmap` currently has some unexpected (at least to me) defaults.\r\n\r\n - The default behaviour is to re-execute the mapped function on error irrespective of the type of error. (i.e. the function is assumed to be idempotent).\r\n\r\n - When the function throws an error, the worker is removed from the pool irrespective of the type of error (i.e. it is assumed that all errors are due to a faulty worker).\r\n\r\n - By default, errors thrown by the mapped function are returned as part of the result array rather than being re-thrown. This is inconsistent with regular `map`.\r\n"},"comment":{"url":"https://api.github.com/repos/JuliaLang/julia/issues/comments/182631649","html_url":"https://github.com/JuliaLang/julia/issues/14843#issuecomment-182631649","issue_url":"https://api.github.com/repos/JuliaLang/julia/issues/14843","id":182631649,"user":{"login":"samoconnor","id":8266283,"avatar_url":"https://avatars.githubusercontent.com/u/8266283?v=3","gravatar_id":"","url":"https://api.github.com/users/samoconnor","html_url":"https://github.com/samoconnor","followers_url":"https://api.github.com/users/samoconnor/followers","following_url":"https://api.github.com/users/samoconnor/following{/other_user}","gists_url":"https://api.github.com/users/samoconnor/gists{/gist_id}","starred_url":"https://api.github.com/users/samoconnor/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/samoconnor/subscriptions","organizations_url":"https://api.github.com/users/samoconnor/orgs","repos_url":"https://api.github.com/users/samoconnor/repos","events_url":"https://api.github.com/users/samoconnor/events{/privacy}","received_events_url":"https://api.github.com/users/samoconnor/received_events","type":"User","site_admin":false},"created_at":"2016-02-10T23:17:59Z","updated_at":"2016-02-10T23:17:59Z","body":"@amitmurthy, In the interests of avoiding one-big-PR, I plan to start by submitting a PR adds just adds `amap` (and its underlying iterator version `imap`) without changing any existing code as follows. Do you agree with this approach?\r\n\r\n**amap(f, c...; async_max=100) → collection**\r\nTransform collection c by applying f to each element using at most 100 asynchronous tasks. For multiple collection arguments, apply f elementwise.\r\nNote: `amap(f, c...; async_max=1)` is equivalent to `map(f, c...)`.\r\n\r\n_Implementation:_\r\n```julia\r\namap(f, c...; kv...) = collect(imap(f, c...; kv...))\r\n```\r\n\r\n**imap(f, c...; async_max=100) → iterator**\r\n\r\nApply f to each element of c using at most 100 asynchronous tasks. For multiple collection arguments, apply f elementwise.\r\nNote: `collect(imap(f, c...; async_max=1))` is equivalent to `map(f, c...)`.\r\n\r\n_Implementation using [StreamMapItr](https://github.com/samoconnor/AsyncMap.jl/blob/master/src/mapitr.jl#L136):_\r\n\r\n```julia\r\nimap(f, c...; async_max=nothing) = StreamMapItr(f, c...; async_max=async_max)\r\n```\r\n\r\n\r\n**amap!(function, collection; async_max=100)**\r\nIn-place version of amap().\r\n\r\n**amap!(function, destination, collection...; async_max=100)**\r\nLike amap(), but stores the result in destination rather than a new collection. destination must be at least as large as the first collection.\r\n\r\n_Implementation using [AsyncMapItr](https://github.com/samoconnor/AsyncMap.jl/blob/master/src/mapitr.jl#L40):_\r\n\r\n```julia\r\nfunction amap!(f, c...; async_max=nothing)\r\n\r\n    destination = c[1]\r\n    if length(c) > 1\r\n        c = c[2:end]\r\n    end\r\n\r\n    for task in AsyncMapItr(f, destination, c..., async_max=max) end\r\n\r\n    return destination\r\nend\r\n```"}}