{"action":"created","issue":{"url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/issues/493","labels_url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/issues/493/labels{/name}","comments_url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/issues/493/comments","events_url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/issues/493/events","html_url":"https://github.com/ruby-concurrency/concurrent-ruby/issues/493","id":125757588,"number":493,"title":"No good tools for long streams of tasks","user":{"login":"chrisseaton","id":341094,"avatar_url":"https://avatars.githubusercontent.com/u/341094?v=3","gravatar_id":"","url":"https://api.github.com/users/chrisseaton","html_url":"https://github.com/chrisseaton","followers_url":"https://api.github.com/users/chrisseaton/followers","following_url":"https://api.github.com/users/chrisseaton/following{/other_user}","gists_url":"https://api.github.com/users/chrisseaton/gists{/gist_id}","starred_url":"https://api.github.com/users/chrisseaton/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/chrisseaton/subscriptions","organizations_url":"https://api.github.com/users/chrisseaton/orgs","repos_url":"https://api.github.com/users/chrisseaton/repos","events_url":"https://api.github.com/users/chrisseaton/events{/privacy}","received_events_url":"https://api.github.com/users/chrisseaton/received_events","type":"User","site_admin":false},"labels":[{"url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/labels/enhance","name":"enhance","color":"84b6eb"}],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":13,"created_at":"2016-01-09T12:45:44Z","updated_at":"2016-01-15T06:27:41Z","closed_at":null,"body":"A couple of times now I've been speaking to someone and they have a problem like this.\r\n\r\n```ruby\r\npool = Concurrent::FixedThreadPool.new(Concurrent.processor_count)\r\n\r\n1.upto 2000000 do |i|\r\n  pool.post do\r\n    # work here\r\n  end\r\nend\r\n```\r\n\r\nor\r\n\r\n```ruby\r\nFile.open('foo').each do |line|\r\n  pool.post do\r\n    # work here\r\n  end\r\nend\r\n```\r\n\r\nIn both cases we're creating a huge number of tasks, and in the latter case we may not know how many tasks in advance.\r\n\r\nThe problem in both cases is that people create millions of tasks which can take up a lot of memory with the proc and the closure.\r\n\r\nI feel like we're missing two abstractions here.\r\n\r\nThe first is a basic parallel `#each` on a `Enumerable` with a `length`. We don't have that do we? It would need chunking (run `n` tasks in each task), and perhaps automatic chunking based on profiling (run 1 task, see how long it takes, think about how many tasks there are and set `n` based on that).\r\n\r\nThe second is something similar that works on an `Enumerator`, which doesn't have a length. Here chunking is harder as we don't know how many tasks there will be in advance. We may need some kind of work stealing here.\r\n\r\nI'd like to write the two examples above as:\r\n\r\n```ruby\r\n1.upto(2000000).parallel_each(pool) do\r\n  # work here\r\nend\r\n```\r\n\r\nor\r\n\r\n```ruby\r\nFile.open('foo').each.parallel_each(pool) do |line|\r\n  # work here\r\nend\r\n```\r\n\r\nIn both cases we'd only create as many tasks at a time as was reasonable (if the pool has `n` threads it may be `kn` tasks for some small constant `k`).\r\n\r\nA workaround in the mean time to just stop so many tasks being created and memory being blown may be to do this (ping @digininja this is relevant to you):\r\n\r\n```ruby\r\npool = Concurrent::FixedThreadPool.new(\r\n  Concurrent.processor_count,\r\n  max_queue: 10 * Concurrent.processor_count,\r\n  fallback_policy: :caller_runs)\r\n\r\n1.upto 2000000 do |i|\r\n  pool.post do\r\n    # work here\r\n  end\r\nend\r\n```\r\n\r\nThis will only create up to 10 times as many tasks as you have cores, with any other tasks being run immediately instead of being added to the pool. This means if you already have say 40 tasks in the pool, instead of creating a new task it will be run, and then by the time the loop gets around to finishing that the pool may be ready for new tasks, or may not be in which case the main thread runs that new task as well."},"comment":{"url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/issues/comments/171882102","html_url":"https://github.com/ruby-concurrency/concurrent-ruby/issues/493#issuecomment-171882102","issue_url":"https://api.github.com/repos/ruby-concurrency/concurrent-ruby/issues/493","id":171882102,"user":{"login":"jdantonio","id":173910,"avatar_url":"https://avatars.githubusercontent.com/u/173910?v=3","gravatar_id":"","url":"https://api.github.com/users/jdantonio","html_url":"https://github.com/jdantonio","followers_url":"https://api.github.com/users/jdantonio/followers","following_url":"https://api.github.com/users/jdantonio/following{/other_user}","gists_url":"https://api.github.com/users/jdantonio/gists{/gist_id}","starred_url":"https://api.github.com/users/jdantonio/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/jdantonio/subscriptions","organizations_url":"https://api.github.com/users/jdantonio/orgs","repos_url":"https://api.github.com/users/jdantonio/repos","events_url":"https://api.github.com/users/jdantonio/events{/privacy}","received_events_url":"https://api.github.com/users/jdantonio/received_events","type":"User","site_admin":false},"created_at":"2016-01-15T06:27:41Z","updated_at":"2016-01-15T06:27:41Z","body":"Excellent feedback and suggestions. Thank you everyone!\r\n"}}