Raise an error in Task.Supervisor.async when reaching max_childrenal/max-children-task-sup

Related to #7786.

Before this commit, when reaching the max children under a 
Task.Supervisor, we failed with a MatchError. Now, we're handling the 
"{:error, :max_children}" that we get from the DynamicSupervisor and 
raising a more meaningful error message. We're raising instead of 
returning an error tuple because async/2,4 are supposed to return the 
task directly.

2 files changed, 102 insertions, 68 deletions

diff --git a/lib/elixir/lib/task/supervisor.ex b/lib/elixir/lib/task/supervisor.ex
index 1b61057e4..ba0b3f86f 100644
--- a/lib/elixir/lib/task/supervisor.ex
+++ b/lib/elixir/lib/task/supervisor.ex
@@ -421,11 +421,21 @@ defmodule Task.Supervisor do
     owner = self()
     args = [owner, :monitor, get_info(owner), {module, fun, args}]
     shutdown = options[:shutdown]
-    {:ok, pid} = start_child_with_spec(supervisor, args, :temporary, shutdown)
-    if link_type == :link, do: Process.link(pid)
-    ref = Process.monitor(pid)
-    send(pid, {owner, ref})
-    %Task{pid: pid, ref: ref, owner: owner}
+
+    case start_child_with_spec(supervisor, args, :temporary, shutdown) do
+      {:ok, pid} ->
+        if link_type == :link, do: Process.link(pid)
+        ref = Process.monitor(pid)
+        send(pid, {owner, ref})
+        %Task{pid: pid, ref: ref, owner: owner}
+
+      {:error, :max_children} ->
+        raise """
+        reached the maximum number of tasks for this task supervisor. The maximum number \
+        of tasks that are allowed to run at the same time under this supervisor can be \
+        configured with the :max_children option passed to Task.Supervisor.start_link/1\
+        """
+    end
   end
 
   defp build_stream(supervisor, link_type, enumerable, fun, options) do
diff --git a/lib/elixir/test/elixir/task/supervisor_test.exs b/lib/elixir/test/elixir/task/supervisor_test.exs
index 86acc5566..8fb908a6e 100644
--- a/lib/elixir/test/elixir/task/supervisor_test.exs
+++ b/lib/elixir/test/elixir/task/supervisor_test.exs
@@ -61,34 +61,57 @@ defmodule Task.SupervisorTest do
              %{active: 0, specs: 0, supervisors: 0, workers: 0}
   end
 
-  test "async/1", config do
-    parent = self()
-    fun = fn -> wait_and_send(parent, :done) end
-    task = Task.Supervisor.async(config[:supervisor], fun)
-    assert Task.Supervisor.children(config[:supervisor]) == [task.pid]
+  describe "async/1" do
+    test "spawns tasks under the supervisor", config do
+      parent = self()
+      fun = fn -> wait_and_send(parent, :done) end
+      task = Task.Supervisor.async(config[:supervisor], fun)
+      assert Task.Supervisor.children(config[:supervisor]) == [task.pid]
+
+      # Assert the struct
+      assert task.__struct__ == Task
+      assert is_pid(task.pid)
+      assert is_reference(task.ref)
+
+      # Assert the link
+      {:links, links} = Process.info(self(), :links)
+      assert task.pid in links
+
+      receive do: (:ready -> :ok)
+
+      # Assert the initial call
+      {:name, fun_name} = Function.info(fun, :name)
+      assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(task.pid)
+
+      # Run the task
+      send(task.pid, true)
+
+      # Assert response and monitoring messages
+      ref = task.ref
+      assert_receive {^ref, :done}
+      assert_receive {:DOWN, ^ref, _, _, :normal}
+    end
 
-    # Assert the struct
-    assert task.__struct__ == Task
-    assert is_pid(task.pid)
-    assert is_reference(task.ref)
+    test "with custom shutdown", config do
+      Process.flag(:trap_exit, true)
+      parent = self()
 
-    # Assert the link
-    {:links, links} = Process.info(self(), :links)
-    assert task.pid in links
+      fun = fn -> wait_and_send(parent, :done) end
+      %{pid: pid} = Task.Supervisor.async(config[:supervisor], fun, shutdown: :brutal_kill)
 
-    receive do: (:ready -> :ok)
+      Process.exit(config[:supervisor], :shutdown)
+      assert_receive {:DOWN, _, _, ^pid, :killed}
+    end
 
-    # Assert the initial call
-    {:name, fun_name} = Function.info(fun, :name)
-    assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(task.pid)
+    test "raises when :max_children is reached" do
+      {:ok, sup} = Task.Supervisor.start_link(max_children: 1)
 
-    # Run the task
-    send(task.pid, true)
+      Task.Supervisor.async(sup, fn -> Process.sleep(:infinity) end)
 
-    # Assert response and monitoring messages
-    ref = task.ref
-    assert_receive {^ref, :done}
-    assert_receive {:DOWN, ^ref, _, _, :normal}
+      assert_raise RuntimeError, ~r/reached the maximum number of tasks/, fn ->
+        Task.Supervisor.async(sup, fn -> :ok end)
+      end
+    end
   end
 
   test "async/3", config do
@@ -104,45 +127,57 @@ defmodule Task.SupervisorTest do
     assert Task.await(task) == :done
   end
 
-  test "async/1 with custom shutdown", config do
-    Process.flag(:trap_exit, true)
-    parent = self()
+  describe "async_nolink/1" do
+    test "spawns a task under the supervisor without linking to the caller", config do
+      parent = self()
+      fun = fn -> wait_and_send(parent, :done) end
+      task = Task.Supervisor.async_nolink(config[:supervisor], fun)
+      assert Task.Supervisor.children(config[:supervisor]) == [task.pid]
 
-    fun = fn -> wait_and_send(parent, :done) end
-    %{pid: pid} = Task.Supervisor.async(config[:supervisor], fun, shutdown: :brutal_kill)
+      # Assert the struct
+      assert task.__struct__ == Task
+      assert is_pid(task.pid)
+      assert is_reference(task.ref)
 
-    Process.exit(config[:supervisor], :shutdown)
-    assert_receive {:DOWN, _, _, ^pid, :killed}
-  end
+      # Refute the link
+      {:links, links} = Process.info(self(), :links)
+      refute task.pid in links
 
-  test "async_nolink/1", config do
-    parent = self()
-    fun = fn -> wait_and_send(parent, :done) end
-    task = Task.Supervisor.async_nolink(config[:supervisor], fun)
-    assert Task.Supervisor.children(config[:supervisor]) == [task.pid]
+      receive do: (:ready -> :ok)
 
-    # Assert the struct
-    assert task.__struct__ == Task
-    assert is_pid(task.pid)
-    assert is_reference(task.ref)
+      # Assert the initial call
+      {:name, fun_name} = Function.info(fun, :name)
+      assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(task.pid)
 
-    # Refute the link
-    {:links, links} = Process.info(self(), :links)
-    refute task.pid in links
+      # Run the task
+      send(task.pid, true)
 
-    receive do: (:ready -> :ok)
+      # Assert response and monitoring messages
+      ref = task.ref
+      assert_receive {^ref, :done}
+      assert_receive {:DOWN, ^ref, _, _, :normal}
+    end
 
-    # Assert the initial call
-    {:name, fun_name} = Function.info(fun, :name)
-    assert {__MODULE__, fun_name, 0} === :proc_lib.translate_initial_call(task.pid)
+    test "with custom shutdown", config do
+      Process.flag(:trap_exit, true)
+      parent = self()
 
-    # Run the task
-    send(task.pid, true)
+      fun = fn -> wait_and_send(parent, :done) end
+      %{pid: pid} = Task.Supervisor.async_nolink(config[:supervisor], fun, shutdown: :brutal_kill)
+
+      Process.exit(config[:supervisor], :shutdown)
+      assert_receive {:DOWN, _, _, ^pid, :killed}
+    end
+
+    test "raises when :max_children is reached" do
+      {:ok, sup} = Task.Supervisor.start_link(max_children: 1)
+
+      Task.Supervisor.async_nolink(sup, fn -> Process.sleep(:infinity) end)
 
-    # Assert response and monitoring messages
-    ref = task.ref
-    assert_receive {^ref, :done}
-    assert_receive {:DOWN, ^ref, _, _, :normal}
+      assert_raise RuntimeError, ~r/reached the maximum number of tasks/, fn ->
+        Task.Supervisor.async_nolink(sup, fn -> :ok end)
+      end
+    end
   end
 
   test "async_nolink/3", config do
@@ -158,17 +193,6 @@ defmodule Task.SupervisorTest do
     assert Task.await(task) == :done
   end
 
-  test "async_nolink/1 with custom shutdown", config do
-    Process.flag(:trap_exit, true)
-    parent = self()
-
-    fun = fn -> wait_and_send(parent, :done) end
-    %{pid: pid} = Task.Supervisor.async_nolink(config[:supervisor], fun, shutdown: :brutal_kill)
-
-    Process.exit(config[:supervisor], :shutdown)
-    assert_receive {:DOWN, _, _, ^pid, :killed}
-  end
-
   test "start_child/1", config do
     parent = self()
     fun = fn -> wait_and_send(parent, :done) end