Disallow streaming with max_concurrency=0 (#10966)

Author: aerosol

lib/elixir/lib/task/supervised.ex

@@ -177,6 +177,11 @@ defmodule Task.Supervised do
   def stream(enumerable, acc, reducer, mfa, options, spawn) do
     next = &Enumerable.reduce(enumerable, &1, fn x, acc -> {:suspend, [x | acc]} end)
     max_concurrency = Keyword.get(options, :max_concurrency, System.schedulers_online())
+
+    unless is_integer(max_concurrency) and max_concurrency > 0 do
+      raise ArgumentError, ":max_concurrency must be an integer greater than zero"
+    end
+
     ordered? = Keyword.get(options, :ordered, true)
     timeout = Keyword.get(options, :timeout, 5000)
     on_timeout = Keyword.get(options, :on_timeout, :exit)

lib/elixir/test/elixir/task_test.exs

@@ -717,6 +717,12 @@ defmodule TaskTest do
       [ok: :ok] = Task.async_stream([1], fn _ -> :ok end, timeout: :infinity) |> Enum.to_list()
     end
 
+    test "does not allow streaming with max_concurrency = 0" do
+      assert_raise ArgumentError, ":max_concurrency must be an integer greater than zero", fn ->
+        Task.async_stream([1], fn _ -> :ok end, max_concurrency: 0) |> Enum.to_list()
+      end
+    end
+
     test "streams with fake down messages on the inbox" do
       parent = self()