nezzart

Periodic tasks in a single GenServer -- not everything yet makes sense

I’ve read on GenServer, but am not able yet to use it because not everything yet makes sense to me.

I want this:

when a user clicks on a button, a new task begins
a task terminates depending on some external condition, say, an external rest-api webservice has returned “terminate = true”
if a task is already running, clicking on a button shouldn’t begin a new
a task poll an external rest-api webservice every 30 seconds

I was adviced to use a single GenServer. My questions:

how can I associate a new task or its name with a user’s id? so that when he clicks on a button I can check if I need to create a new task or do nothing if there’s already one running?
how can I add/remove an item to MyWorker? Meaning, should I keep a list of users id?
if there’re no tasks, GenServer should stop. I think. Is it a wise thing to do? If so, how can I stop/run it?

to re-run a task every 30 seconds I need to call schedule_work() in “poll_external_service”, correct? Thus, to terminate it, depending on a response from an external web rest service, I merely don’t run schedule_work(), correct?

defmodule MyWorker do
use GenServer

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

 def init(state) do
   schedule_work() 
   {:ok, state}
 end

 def add_item(x) do

 end

 def remove_item(x) do

 end


 defp poll_external_service do
   # a request to an external web server

   schedule_work()
 end

 defp schedule_work() do
   Process.send_after(self(), :work, 30 * 1000) 
 end

end

18 comments

#task #genserver

7 4168 18

2017-05-22 13:38:16 UTC

Marked As Solved

net

I may have misunderstood what you’re trying to do, but this might help.

Assuming that you only wanted to store the user ID for each ask, I used a MapSet. If you want to store additional data with the task (such as a specific URL to poll), you should use a Map (with the key being the user ID, as map keys are unique).

In process_tasks/1 we use Enum.filter/2 to both call a function for each task, and to remove tasks that the external API has indicated should be terminated. The case ExternalAPI.poll(id) do is of course, a contrived example. What’s important is that the anonymous function given to Enum.filter/2 returns false if the task should be removed. (You can give Enum.filter/2 a map too, if you go that route. In that event the anonymous function should take a tuple {user_id, data} representing each key-value pair in the map.)

There’s no need to stop the GenServer when there are no tasks. Erlang processes are very lightweight.

defmodule MyWorker do
  use GenServer

  @tick_interval 30_000

  def start_link, do: GenServer.start_link(__MODULE__, nil, name: __MODULE__)

  def add(id), do: GenServer.cast(__MODULE__, {:add, id})

  def remove(id), do: GenServer.cast(__MODULE__, {:remove, id})

  def init(_) do
    tick()

    # Here we use a MapSet as it will reject duplicate items.
    # If you wish to store more data with a task than just a user ID,
    # use a Map instead.
    {:ok, MapSet.new}
  end

  defp tick, do: Process.send_after(self(), :tick, @tick_interval)

  def handle_cast({:add, id}, tasks), do: {:noreply, MapSet.put(tasks, id)}

  def handle_cast({:remove, id}, tasks), do: {:noreply, MapSet.delete(tasks, id)}

  def handle_info(:tick, tasks) do
    tasks = process_tasks(tasks)

    tick()

    {:noreply, tasks}
  end

  defp process_tasks(tasks) do
    tasks_list =
      Enum.filter tasks, fn(id) ->
        # Here we process the individual task, such as polling
        # the external API.
        case ExternalAPI.poll(id) do
          # We return true from this anonymous function if the task should
          # stay, and false if the task should be removed.
          "terminate = true" -> true
          "terminate = false" -> false
        end
      end

    # `Enum.filter/2` returns a list, so we must convert it back into a MapSet.
    MapSet.new(tasks_list)
  end

  def handle_info(_, tasks), do: {:noreply, tasks}
end

Post #2

Also Liked

OvermindDL1

Yes, and it is described in the GenServer documentation.

Imagine genserver implemented this way in pseudo-code:

def start_link(user_module, args) do
  {:ok, spawn_link(fn -> gen_server_init(user_module, args) end)}
end

def gen_server_init(user_module, args) do
  {:ok, state} = user_module.init(args)
  gen_server_loop(user_module, state)
end

def gen_server_loop(user_module, state) do
  receive do
    {:__special_genserver_cast__, msg} ->
      {:noreply, newstate} = user_module.handle_cast(msg, state)
      gen_server_loop(user_module, newstate)
    {:__special_genserver_call__, msg, from} ->
      {:reply, outmsg, newstate} = user_module.handle_call(msg, from, state)
      send(from, outmsg)
      gen_server_loop(user_module, newstate)
    unknown_msg ->
      {:noreply, newstate} = user_module.handle_info(unknown_msg, state)
      gen_server_loop(user_module, newstate)
  end
end

And so forth, but with significantly more error checking and cases and such. GenServer is just an infinite receiving loop passing the state from iteration to iteration.

Post #18

net

You’re right, that is better. Instead of Enum.filter/2 you might consider just using Enum.each/2 and calling MyWorker.remove/1 from the process handling the individual task if the task should be removed.

Post #4

Qqwy

TypeCheck Core Team

The MyWorker uses the :name option as part of the start_link call, which means that the process is registered under a global (node-wide) name. The other calls that use GenServer.call and GenServer.cast use the same name to access it later.

So this means that the process will remain available after starting until your application closes, the process crashes, or you call GenServer.stop/2 from somewhere. In any case, the process will remain regardless of the connecting/disconnecting of users to your web-facing interface.

Post #10