First come first served job scheduler with max concurrency

scheduler.erl

-module(scheduler).

-export([
    run/0
]).

-define(MAX_CONCURRENCY, 3).

run() ->
    Msgs = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10],
    run(Msgs, ?MAX_CONCURRENCY, 0, 0, 0).

run([], _, 0, Sent, Received) -> [{sent, Sent}, {received, Received}];
run([], Max, Cur, Sent, Received) ->
    ok = wait(),
    run([], Max, Cur-1, Sent, Received+1);
run([Msg | Msgs], Max, Cur, Sent, Received) when Cur =:= Max ->
    ok = wait(),
    work(Msg),
    run(Msgs, Max, Cur, Sent+1, Received+1);
run([Msg | Msgs], Max, Cur, Sent, Received) when Cur < Max ->
    work(Msg),
    run(Msgs, Max, Cur+1, Sent+1, Received).

work(Msg) ->
    Self = self(),
    Pid = spawn(fun() -> do_work(Self, Msg) end),
    io:format("sent work ~p to ~p ~n", [Msg, Pid]).

wait() ->
    receive
        {From, Msg} ->
            io:format("got result ~p from ~p ~n", [Msg, From]),
            ok
    end.

do_work(From, Msg) ->
    Self = self(),
    Rand = rand:uniform(10), % let's say each message requires max 10 seconds to complete
    timer:sleep(Rand * 1000),
    From ! {Self, Msg}.

17> scheduler:run().
sent work 1 to <0.180.0>
sent work 2 to <0.181.0>
sent work 3 to <0.182.0>
got result 1 from <0.180.0>
sent work 4 to <0.183.0>
got result 4 from <0.183.0>
sent work 5 to <0.184.0>
got result 3 from <0.182.0>
sent work 6 to <0.185.0>
got result 2 from <0.181.0>
sent work 7 to <0.186.0>
got result 5 from <0.184.0>
sent work 8 to <0.187.0>
got result 6 from <0.185.0>
sent work 9 to <0.188.0>
got result 9 from <0.188.0>
sent work 10 to <0.189.0>
got result 8 from <0.187.0>
got result 7 from <0.186.0>
got result 10 from <0.189.0>
[{sent,10},{received,10}]
18>