An gen_server for mass regexp computing... (LibTre)

This is the first test session of my 'tregex_srv' that provides some nice regexp features:

266> l(tregex_srv).                                                 
{module,tregex_srv}
267> tregex_srv:start_link().                                       
{ok,<0.4045.0>}
268> tregex_srv:store( [<<"[0-9+] pid">>, <<"[a-z]+.tmp">>]).       
ok
269> tregex_srv:grep(<<"test 9405904.tmp acuu.tmpmulaor 10+ pid">>).
[[[{34,39,<<"+ pid">>}],[{17,25,<<"acuu.tmp">>}]]]
270> tregex_srv:store( [{ <<"test">>, fun(X) -> io:format("found: ~p~n", [X]) end},  <<"[0-9][0-9]">>]).
ok
271> tregex_srv:grep(<<"test 9405904.tmp acuu.tmpmulaor 10+ pid">>).                                    
found: [{0,4,<<"test">>}]
[[[{34,39,<<"+ pid">>}],[{17,25,<<"acuu.tmp">>}]],
 [[{0,4,<<"test">>}],[{5,7,<<"94">>}]]]
272> tregex_srv:store( [{ <<"SRC=[^ ]+">>, fun(X) -> 
 [{_,_,M}] = X, io:format("Source: ~p~n", [M]) 
end}]).                   
ok
273> tregex_srv:grep(<<"test 9405904.tmp acuu.tmpmulaor 10+ pid">>).                                      
found: [{0,4,<<"test">>}]
[[[{34,39,<<"+ pid">>}],[{17,25,<<"acuu.tmp">>}]],
 [[{0,4,<<"test">>}],[{5,7,<<"94">>}]],
 []]
274> tregex_srv:grep(<<"tst SRC=192.135.15.1 pid">>).               
Source: <<"SRC=192.135.15.1">>
[[[{19,24,<<"1 pid">>}]],
 [[{8,10,<<"19">>}]],
 [[{4,20,<<"SRC=192.135.15.1">>}]]]
275> tregex_srv:store( [{ <<"SRC=([^ ]+)">>, fun(X) -> 
 [{_,_,_}, {_,_,M}] = X, io:format("Source IP: ~p~n", [M]) 
 end}]).
ok
276> tregex_srv:grep(<<"tst SRC=192.135.15.1 pid">>).                                                                   
Source IP: <<"192.135.15.1">>
Source: <<"SRC=192.135.15.1">>
[[[{19,24,<<"1 pid">>}]],
 [[{8,10,<<"19">>}]],
 [[{4,20,<<"SRC=192.135.15.1">>}]],
 [[{4,20,<<"SRC=192.135.15.1">>},{8,20,<<"192.135.15.1">>}]]]

As you can see, you can associate Funs with regexp Matches. This means that you can bind action to regexp...
First we store (in fact add regexp to the existing regexp list) new tuples {RE, Fun}:

275> tregex_srv:store( [{ <<"SRC=([^ ]+)">>, fun(X) -> 
 [{_,_,_}, {_,_,M}] = X, io:format("Source IP: ~p~n", [M]) 
 end}]).
ok

Now the exec does call already registered funs, but call the new one since our regexp matches and you can see that the IP number is only printed, the "submatches" feature works as expected:

276> tregex_srv:grep(<<"tst SRC=192.135.15.1 pid">>).                                                                   
Source IP: <<"192.135.15.1">>
Source: <<"SRC=192.135.15.1">>
...

The gen_server state is the following:

-record(state, {
                requests,
                reindex,
                re = [],
                pids = []
                }).

Its init function is:

init(_Args) ->
        process_flag(trap_exit, true),
        {ok, #state{  
 re = ets:new(?MODULE, [set,private]), 
 requests = 0, 
 reindex = 1 }}.

Internally the module calls 'treregex:compile' to compile regexp and store the resulting #port into a list that is stored in the 'ets' table. Every call to 'tregex_srv:store' create a new entry in the ets table.

%% Storing RE and Funs
%% Creating simple fun when there's none provided...
%%
store([], Res, State) ->
        ets:insert(State#state.re, { State#state.reindex, Res});
store([ { Regexp, Fun } | List ], Res, State) ->
        {ok, Re } = treregex:compile(iolist_to_binary(Regexp), [extended]),
        store(List, [ { Re, Fun } | Res ], State);
store([ Regexp | List ], Res, State) ->
        {ok, Re } = treregex:compile(iolist_to_binary(Regexp), [extended]),
        store(List, [ { Re, fun(_) -> false end} | Res ], State).

The 'tregex_srv:grep' just uses 'ets:foldl' to compute results:

handle_call({grep, Line}, _Node, State) ->
        Requests = State#state.requests,
        Grep = fun({_Reindex, ReList}, Acc) ->
                        [ exec(ReList, Line, []) | Acc]
                end,
        {reply, ets:foldl(Grep, [], State#state.re), State#state{ requests = Requests + 1} }.

%% exec, using a List of {Re, Funs}
exec([], _Line, Acc) ->
        Acc;
exec([ { Re, Fun } | ReList ], Line, Acc) ->
        case treregex:exec(Re, Line) of
                {ok, Matches} ->
                        Fun(Matches),
                        exec(ReList, Line, [ Matches | Acc ]);

                {error, nomatch} ->
                        exec(ReList, Line, Acc)
        end;
exec([ _Any | ReList ], Line, Acc) ->
        exec(ReList, Line, Acc).

The code is still young, but seems to work.

The main purpose here, is to be able to massively process lines of logs. I want to be able to
spawn multiple process on multiples nodes that will be able to extract valuable content from
various lines. This is the first step forward :-)

I may cleanup the 'grep' fun since it will returns empty list whenever a regexp didn't match anything from the supplied line...

I'm really excited to think that I will be able to use the 'gen_server:multi_call' with this module :)

Parallelizing simple external commands ... Part II

Our loop/3 fun looks like this:

loop(_Max, 0, []) ->
     unregister(computing_master),
     exit(normal);

Whenever our list of jobs is empty, we deregister the 'computing_master' process and quit normally.

loop(Max, Current, []) ->
     receive
             stop ->
                     unregister(computing_master),
                     exit(normal);

             {exited, _Result} ->
                     io:format("Still ~p childs~n", [Current]),
                     loop(Max, Current - 1, []);

             E ->
                     io:format("Unhandled message: ~p~n", [E])

     after 60000 ->
             io:format("~p: Waiting for the last process ~p/~p~n", [erlang:now(), Max, Current]),
             loop(Max, Current, [])
     end;

In this case, we have are computing the last external process since our job list is empty.
And finally this version of loop/3 is the main one:

loop(Max, Current, List)  ->
     receive
             stop ->
                     unregister(computing_master),
                     exit(normal);

             {update, NewMax} ->
                     upto(NewMax, Max, List);

             {exited, _Result} ->
                     io:format("Still ~p childs~n", [Max]),
                     upto(Max, Max - 1, List);

             E ->
                     io:format("Unhandled message: ~p~n", [E])

     after 60000 ->
             io:format("~p: Running ~p processes~n", [erlang:now(), Max]),
             upto(Max, Current, List)
     end.

Here we have a non empty list of job and a number of job to start.

• Every 60 seconds we write how many processes are running.
• The message {update, NewMax} let's you alter the number max of concurrent tasks
• The message {exited, _Result} is received whenever a child process dies, so we restart another job...
  

Bonus Code, a simple function to test the code:

sleep(Ident) ->
        io:format("Waiting ~p~n", [Ident]),
        Delay = [ "5", "3", "15", "8" ],
        Time = lists:nth(random:uniform(4), Delay),
        Cmd = [ "sleep ", Time ], 
        io:format("Starting: ~p~n", [Cmd]),
        Status = os:cmd(Cmd),
        computing_master ! {exited, Status}.

This code just calls the 'sleep' command with various arguments picked randomly... Once a process stops the 'os:cmd/1' fun exits and 'computing_master' will receive the {exited, Status} message (explained above)

Parallelizing simple external commands ... Part I

Once upon a time I need to parse enormous files to find simple patterns... My prefered tools were so far the shell based one, i.e. 'grep'.

But now I have a Magical Ability, Erlang Magic... So I decide to split this enormous file, using the 'split' comand, 'split -l 10000' for example.

Now that I have a lot of smaller file, I can parallelize their parsing, and this is were erlang comes...

First, let's design a bit:

• I need a central process that will control all my processes
• Processes and master must be able to communicate

That's all. Hopefully the latter is directly provided by erlang, this the ! operator.
The master process will be a little more tricky, but this is 'easyerl' remember, so here we go:

doit(Step) ->
    Master = spawn(?MODULE, test, [Step]),
    register(computing_master, Master).

test(Step) ->
    file:set_cwd("/home/rolphin/Work"),
    List = filelib:wildcard("seg-a*"),
    upto(Step, 0, List).

We create a process running the test function, whose job is starting the upto/3 fun...
What's interesting here is the 'filelib' function that provides me the list of file contained in the directory '/home/rolphin/Work'.

Now we go describe the 'upto/3' fun :

upto(Max, Current, []) ->
    loop(Max, Current, []);

upto(Max, Max, List) ->
    loop(Max, Max, List);

upto(Max, Current, [New|List]) ->
    io:format("upto: ~p/~p~n", [Max, Current]),
    spawn(?MODULE, grep, ["user.list", New, ["result-", New]]),
    upto(Max, Current + 1, List).

More details:

• upto with an empty list will just call the loop/3 fun
• upto with the Max number of processe allowed equals the current number of process, will call loop/3
• upto with less active process than the max, with a non empty list, will spawn a child process

The child process is a 'grep' command, and here it is:

grep(File, Source, Result) ->
       % Command line is: "grep -f motif_file sourcefile > result"
       Cmd = [ "grep -f ", File , $ , Source, $>, Result ],
       io:format("Starting: ~p~n", [Cmd]),
       Status = os:cmd(Cmd),
       computing_master ! {exited, Status}.

Okay that's it for today ! It's a little late now ! And I need some sleep to succesfully pass the required skill tests for my new job !

More of this tomorrow...