3
\$\begingroup\$

I am coming from Java and I am trying to create a process pool in Erlang.

Like the Java API:

ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler)

I can define core threads, time to kill thread, and policy to discard/reject tasks. I created some code in Erlang and it was working, but I am not confident about it.

The questions:

  1. Am I using it correctly?
  2. Any potential bugs in the code?
  3. How can I optimize the code?

-module(worker_pool).
-export([ init/0, add_job/1]).

-define(TIMEOUT, 6000).
-define(POOL, pool).
-define(MAX, 3).

init() ->
   ets:new(?POOL, [public, named_table, ordered_set]),
   Pid = spawn(?MODULE, controller, []),
   register(?MODULE, Pid).

 %% add job to pool by sending message.
add_job(Value) ->
   ?MODULE ! {new_worker, Value}.

worker() ->
   receive
      {do_work, Value} ->
      example(Value);
    _ ->
      exit(no_valid)
    after ?TIMEOUT ->
        io:format("I (worker ~p) will die now ...~n", [self()]),
      exit(no_activity)
   end.

  %% create pool of process and trigger it by message.
  controller() ->
    receive
      {new_worker, Value} ->
        Pid = insert_job(),
        Pid ! {do_work, Value},
       controller();
    {'DOWN', _Ref, process, Pid, Reason} ->
       io:format(" worker ~p died (~p)~n", [Pid, Reason]),
        delete_job(Pid),
        controller();
    Msg ->
       io:format(" Controller ~p ~n", [Msg]),
        controller()
  end.

 %% internal functions.
   example(Value) ->
      timer:sleep(2000),
     io:format("sleep [~p] done ~n", [Value]).

%% insert to ets table, ordered by time.
  insert_job() ->
     Jobs = ets:info(?POOL, size),
      insert_job(Jobs, ?MAX).
   insert_job(Jobs, Max) when Jobs < Max ->
          {Pid, _} = spawn_monitor(fun() -> worker() end),
           ets:insert(?POOL, {{erlang:monotonic_time(seconds), pid_to_list(Pid)}, []}),
            Pid;
     insert_job(_Jobs, _Max) ->
           First = ets:first(?POOL),
          {_, Key} = First,
          exit(list_to_pid(Key), cancelled),
          ets:delete(?POOL, First),
            {Pid, _} = spawn_monitor(fun() -> worker() end),
            ets:insert(?POOL, {{erlang:monotonic_time(seconds), pid_to_list(Pid)}, []}),
            Pid.

    %% delete job from the pool if it was finished.
    delete_job(Pid) ->
     Pid_list = pid_to_list(Pid),
     ets:match_delete(?POOL, {{'_', Pid_list}, '_'}).`
\$\endgroup\$
  • \$\begingroup\$ usually in Erlang process pool create with OTP, for example look [hottub] (github.com/bfrog/hottub) . Btw, several questions on your implementation - why ets table and why do the functions of the controller and workers exported? \$\endgroup\$ – user110702 Jul 29 '16 at 16:03
  • \$\begingroup\$ @Kvaz1r thanks for your comments. the reason for ets was to limit the generated process. Queue was also considered here, but I still need a ets table to store the queue. agreed with you, controller and workers should not be exported. \$\endgroup\$ – user3644708 Jul 29 '16 at 23:08
  • \$\begingroup\$ good, however in this case you can't use spawn/3 it's only for exported functions. Use register(?MODULE, spawn(fun()-> controller() end )) instead. \$\endgroup\$ – user110702 Jul 30 '16 at 18:33
1
\$\begingroup\$

As far as I can tell, the implementation is normal,but something in the code can be simplified:

1. ets can store any type of Erlang including Pid, it makes no sense to convert to list and back

2. Remove code duplication in insert_job endure in a separate function actually added to the table:

insert_job() ->
  {Pid, _} = spawn_monitor(fun() -> worker() end),
  ets:insert(?POOL, {{erlang:monotonic_time(seconds), Pid}, []}),
  Pid.

3. Even more unload of function insert job , for that creating a new function to remove the first task:

remove_first_job() ->
  First = ets:first(?POOL),
  {_, Key} = First,
  exit(Key, cancelled),
  ets:delete(?POOL, First).

4. Rewrite called from controller() insert_job so that to remove the guard expression.

Putting it all together, we get:

-module(worker_pool).
-export([init/0, add_job/1]).

-define(TIMEOUT, 6000).
-define(POOL, pool).
-define(MAX, 3).

init() ->
  ets:new(?POOL, [public, named_table, ordered_set]),
  register(?MODULE, spawn(fun() -> controller() end)).

%% add job to pool by sending message.
add_job(Value) ->
  ?MODULE ! {new_worker, Value}.

worker() ->
  receive
    {do_work, Value} ->
      example(Value);
    _ ->
      exit(no_valid)
  after ?TIMEOUT ->
    io:format("I (worker ~p) will die now ...~n", [self()]),
    exit(no_activity)
  end.

%% create pool of process and trigger it by message.
controller() ->
  receive
    {new_worker, Value} ->
      Pid = insert_job(ets:info(?POOL, size), ?MAX),
      Pid ! {do_work, Value},
      controller();
    {'DOWN', _Ref, process, Pid, Reason} ->
      io:format(" worker ~p died (~p)~n", [Pid, Reason]),
      delete_job(Pid),
      controller();
    Msg ->
      io:format(" Controller ~p ~n", [Msg]),
      controller()
  end.

%% internal functions.
example(Value) ->
  timer:sleep(2000),
  io:format("sleep [~p] done ~n", [Value()]).

%% insert to ets table, ordered by time.
insert_job(?MAX, ?MAX) ->
  remove_first_job(),
  insert_job();
insert_job(_, _) ->
  insert_job().

remove_first_job() ->
  First = ets:first(?POOL),
  {_, Key} = First,
  exit(Key, cancelled),
  ets:delete(?POOL, First).

insert_job() ->
  {Pid, _} = spawn_monitor(fun() -> worker() end),
  ets:insert(?POOL, {{erlang:monotonic_time(seconds), Pid}, []}),
  Pid.

%% delete job from the pool if it was finished.
delete_job(Pid) ->
  ets:match_delete(?POOL, {{'_', Pid}, '_'}).

Among other things would be good to add a check, for example, correctly handle the call add_job/1 before call init/0 and realize opportunity the stop with the unregistered of process.

As mentioned earlier in the comments, usually in Erlang process pool create with OTP, for example look hottub.

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.