kaa.python

kaa.python OP

How will that help me to handle :DOWN or :exit?
I already have something like this.

0
Post #11
kaa.python

kaa.python OP

I don’t understand: no answer and before timeout? why would I want to return anything in such a case? I’d wait for timeout or answer

0
Post #12
NobbZ

NobbZ

Don’t just look at the function head but the body as well. The code you have shown elsewhere just stored the last from in the state, this is a map which is able to get the correct from for any ref. And the ref is exactly the thing, that is also used by Task.await/1 and Task.yield/1 to check if they got the correct message. The ref is also available in the :DOWN message.

Task.yield does return nil after the timeout (which is bad enough since it could have been a proper result of the computation in the task). The task will continue to run, and when you return the %Task{}, your receiver can of course ask for the result again using Task.yield/1/2. Another possibility were to simply return {:error, :timeout} from your GenServer when yield didn’t yield anything (but then one should try to kill or stop the Task.

Currently, I’m not even sure if a Task is the thing you want to go with. I do think, that it were so much easier, just to spawn a process and let it do the GenServer.reply(from, answer) stuff. In that process you could do anything you want regarding error- and timeout handling and don’t need to mess up with your genserver anymore.

0
Post #13
kaa.python

kaa.python OP

I need to use GenServer because I have a bunch of user with some data for which i need to poll an external website every N seconds. Data/state in GenServer can grow, become empty and grow again over time depending on a response of an external website. That is, it’s not a one-time operation, it has to work in background all the time.

0
Post #14
NobbZ

NobbZ

I’m not talking about repl,acing the GenServer, but about replacing the Task by a bare process

0
Post #15
kaa.python

kaa.python OP

Maybe. I’m asking for advice. How will it work in “do_work” function, will it get along with GenServer well?

0
Post #16
NobbZ

NobbZ

I’m not sure if I do understand your question correctly here. But outsourcing work from a GenServer into another process as soon as it will take longer than a couple of “cycles” is idiomatic and good practice.

This is done to keep the work the actual GenServer process has to do to a minimum so that it is able to process incomming calls and casts as fast as possible,

But as a rough concept it looks like this:

def handle_call{:do_stuff, from, status} do
  spawn(fn ->
    answer = do_stuff()
    GenServer.reply(from, answer)
  end)
  {:noreply, state}
end

Doing it this way, will remove the burden to handle any communication with that process from your GenServer. Assuming that HTTPoison does have timeouts, you can even handle them inside that spawned process and reply differently from the default when a timeout happens. Also you can try and catch there. You will still run into trouble when do_stuff or something inside of it does call exit!

1
Post #17
peerreynders

peerreynders

Basically ditch Task.await and simply grab the results in handle_info/2. If your tasks have a habit of abnormally exiting, trap exits and process the :EXIT messages accordingly.

defmodule Charlie do
  use GenServer

  defp new_task do
    max_time = 5000
    work =
      case {(:rand.uniform max_time), (:rand.uniform max_time)} do
        {work, panic} when work < panic ->
          fn ->
            Process.sleep work
            work # return the work time as the result
          end
        {_, panic} ->
          fn ->
            Process.sleep panic
            exit(:panic) # simulate non-normal exit
          end
      end

    Task.async work
  end

  defp add_task(tasks, task),
    do: Map.put tasks, task.pid, task

  defp purge_task(tasks, pid) do
    case Map.get tasks, pid, :none do
      :none ->
        {tasks, :none}
      task ->
        Process.demonitor task.ref, [:flush]   # purge related :DOWN messages
        {(Map.delete tasks, pid), task}
    end
  end

  defp shutdown_tasks(tasks) do
    tasks
    |> Map.values()
    |> Enum.each(&(Task.shutdown &1, :brutal_kill))
  end

  ## callbacks
  def handle_info(:next_task, {tasks, _ref}) do
    # time to create another task
    new_tasks =
      cond do
        (length (Map.keys tasks)) < 50 ->
          add_task tasks, new_task()
        true ->
          tasks
      end

    timer_ref = Process.send_after self(), :next_task, 500
    {:noreply, {new_tasks, timer_ref}}
  end
  def handle_info({ref, result}, state) when is_reference ref do
    # handle Task completion
    Process.demonitor ref, [:flush]   # purge related :DOWN messages ASAP

    IO.puts "Task #{inspect ref} completed with result #{result}"

    {:noreply, state} # remove task when :EXIT :normal is processed
  end
  def handle_info({:EXIT, from, reason}, {tasks, timer_ref} = state) do
    # Handle exit signal not handled by the behaviour (i.e. signal from parent process)
    case purge_task tasks, from  do
      {_, :none} ->
        IO.puts "Unknown :EXIT #{inspect reason}"
        case reason do
          :normal ->
            {:noreply, state}
          _ ->
            {:stop, reason, state}  # Follow protocol: unknown non-normal exit signal - time to terminate
        end
      {new_tasks, task} ->
        case reason do
          :normal ->
            :ok
          _ -> # The task panicked
            IO.puts "Task #{inspect task.ref} panicked: #{inspect reason}"
        end
        {:noreply, {new_tasks, timer_ref}}
    end
  end

  def init(_args) do
    Process.flag(:trap_exit, true)
    Kernel.send self(), :next_task # start spinning off tasks
    {:ok, {%{}, :none}}
  end

  def terminate(_reason, {tasks, timer_ref}) do
    IO.puts "Terminating"
    cond do
      is_reference timer_ref ->
        Process.cancel_timer timer_ref
      true ->
        0 # no timer, therefore no time left
    end

    shutdown_tasks tasks
    :ok
  end

  ## public interface

  def start_link do
    GenServer.start_link(__MODULE__, [])
  end

  def stop(pid) do
    GenServer.stop pid
  end

  # NOTE:
  # https://hexdocs.pm/elixir/Task.html#await/2-compatibility-with-otp-behaviours
  # It is not recommended to await a long-running task inside an OTP behaviour such as GenServer.
  # Instead, you should match on the message coming from a task inside your
  # GenServer.handle_info/2 callback.
  #
end

.

$iex -S mix
iex(1)> Process.flag :trap_exit, true 
false
iex(2)> {:ok,pid} = Charlie.start_link
{:ok, #PID<0.127.0>}
Task #Reference<0.0.6.653> panicked: :panic
Task #Reference<0.0.6.651> completed with result 1306
Task #Reference<0.0.6.647> panicked: :panic
Task #Reference<0.0.6.658> panicked: :panic
Task #Reference<0.0.5.605> panicked: :panic
Task #Reference<0.0.6.649> panicked: :panic
Task #Reference<0.0.6.661> completed with result 464
Task #Reference<0.0.6.668> panicked: :panic
Task #Reference<0.0.6.677> panicked: :panic
Task #Reference<0.0.6.673> panicked: :panic
Task #Reference<0.0.6.682> completed with result 957
Task #Reference<0.0.6.684> completed with result 665
Task #Reference<0.0.6.675> panicked: :panic
iex(3)> Process.exit pid, :whatever
Terminating
true
iex(4)>
1
Post #18
kaa.python

kaa.python OP 

def handle_info({ref, result}, state) when is_reference ref do
    # handle Task completion
    Process.demonitor ref, [:flush]   # purge related :DOWN messages ASAP

    IO.puts "Task #{inspect ref} completed with result #{result}"

    {:noreply, state} # remove task when :EXIT :normal is processed
  end

That’s not what I’m looking for because I have tasks that don’t complete after the 1st iteration, thus it takes them a few iterations to complete. Also, they’re sending http requests which can either take a long time or cause a timeout depending on how I should set them up. How should you code be adjusted then?

0
Post #19
peerreynders

peerreynders

The tasks take however long they need - no “iterations” involved.

0
Post #20