Pichi's blog

Are Perl and Erlang functional enough for Nemo?

2011-05-29T05:06:00.000-07:00

I have read interesting litmus test for a functional language from stackoverflow user Nemo. Just for curiosity there is mine try for perl:

sub Inc{1+shift}
sub Thrice {
  my $f=shift;
  sub {
    $f->($f->($f->(shift)))
  }
}
$\="\n";
  print
for Thrice(\&Inc)->(0),
    Thrice(Thrice(\&Inc))->(0),
    Thrice(\&Thrice)->(\&Inc)->(0)

and Erlang:

1> Thrice = fun(F) -> fun(X) -> F(F(F(X))) end end.
#Fun
2> Inc = fun(X) -> 1+X end.
#Fun
3> (Thrice(Inc))(0).
3
4> (Thrice(Thrice(Inc)))(0).
9
5> ((Thrice(Thrice))(Inc))(0).
27

Both seems functional enough for mine personal taste.

Closure vs. Object in Perl

2009-10-06T06:36:00.000-07:00

I was curios how closures performs in comparison with objects in Perl. I have tested it on counter as the simplest possible abstraction with state.

use Benchmark qw(cmpthese);

sub make_counter {
    my $counter = shift;
    return sub { $counter++ };
}

package counter;

sub new {
    my ( $class, $counter ) = @_;
    return bless \$counter, $class;
}

sub inc { ${ shift() }++ }

package main;

our $inc = make_counter(1);

our $counter = counter->new(1);

cmpthese(
    -5,
    {   closure_make => q{make_counter(1)},
        object_make  => q{counter->new(1)}
    }
);

cmpthese(
    -5,
    {   closure => q{$main::inc->()},
        method  => q{$main::counter->inc()}
    }
);

And results are:

                 Rate closure_make  object_make
closure_make 420045/s           --         -35%
object_make  643969/s          53%           --
             Rate  method closure
method  2397172/s      --    -35%
closure 3697681/s     54%      --

Well, it is not simple. Both approaches have their benefits but if you want best performance and want little bit abstraction you should pass closure into your hot loop instead objects. But if you want create closure inside loop, than let you try avoid it or change to object. But you should not need it in hot loop anyway. I'm little bit surprised that closure construction is so expensive in perl. I have known that bless is not cheap operation, but closure?

How fast or slow is Moose?

2009-02-15T09:22:00.000-08:00

Moose is a postmodern object system for Perl 5 that takes the tedium out of writing object-oriented Perl. It borrows all the best features from Perl 6, CLOS (LISP), Smalltalk, Java, BETA, OCaml, Ruby and more, while still keeping true to its Perl 5 roots. It is very powerful and I was curious how fast is in current state. I have read that Moose is slow but all articles what I have found is about two years old. For me is important mostly runtime speed but compile time. So my benchmark is focused only to runtime. I also hate getter/setter combined accessors thus my tests are also only about separated getter and setter. Benchmark code follows.

#!/usr/bin/env perl

use strict;
use warnings;
use Data::Dumper;
use Benchmark qw(:all :hireswallclock);

{

 package MooseClassMutable;
 use Moose;

 has var => (
  is       => 'ro',
  reader   => 'get_var',
  writer   => 'set_var',
  required => 1
 );

}
{

 package MooseClassImmutable;
 use Moose;

 has var => (
  is       => 'ro',
  reader   => 'get_var',
  writer   => 'set_var',
  required => 1
 );

 no Moose;
 __PACKAGE__->meta->make_immutable;
}
{

 package PerlClass;

 sub new {
  my ( $class, %args ) = @_;
  die 'var value must be set' unless exists $args{var};
  return bless \%args, $class;
 }

 sub get_var { shift()->{var} }

 sub set_var { $_[0]->{var} = $_[1] }
}

{

 package MooseClassFast;
 use Moose;
 with 'MooseX::Emulate::Class::Accessor::Fast';

 has var => (
  is       => 'ro',
  required => 1
 );

 __PACKAGE__->follow_best_practice;
 __PACKAGE__->mk_accessors('var');

 no Moose;
 __PACKAGE__->meta->make_immutable;
}

cmpthese(
 -5,
 {   map {
   my $class = $_;
   "new $class" => sub { $class->new( var => 1 ) for 1 .. 1000 }
   } qw(MooseClassMutable MooseClassImmutable PerlClass MooseClassFast)
 }
);
my %objs = ( map { $_ => $_->new( var => 1 ) }
  qw(MooseClassMutable MooseClassImmutable PerlClass MooseClassFast) );
cmpthese(
 -5,
 {   map {
   my $class = $_;
   my $obj   = $objs{$class};
   "get $class" => sub { $obj->get_var() for 1 .. 1000 }
   } qw(MooseClassMutable MooseClassImmutable PerlClass MooseClassFast)
 }
);
cmpthese(
 -5,
 {   map {
   my $class = $_;
   my $obj   = $objs{$class};
   "set $class" => sub { $obj->set_var(1) for 1 .. 1000 }
   } qw(MooseClassMutable MooseClassImmutable PerlClass MooseClassFast)
 }
);

I have tested Moose versions 0.54 and 0.68 and just for curiosity also Class::Accessor::Fast emulation which works only with Moose version 0.68. Notice that rate is measured in thousands. Moose 0.54 results comes first.

                          Rate new MooseClassMutable new MooseClassImmutable new PerlClass
new MooseClassMutable   6.85/s                    --                    -96%          -98%
new MooseClassImmutable  192/s                 2697%                      --          -53%
new PerlClass            403/s                 5790%                    111%            --
                          Rate get MooseClassMutable get MooseClassImmutable get PerlClass
get MooseClassMutable   1716/s                    --                     -2%          -25%
get MooseClassImmutable 1754/s                    2%                      --          -23%
get PerlClass           2273/s                   32%                     30%            --
                          Rate set MooseClassImmutable set MooseClassMutable set PerlClass
set MooseClassImmutable 1611/s                      --                   -2%          -16%
set MooseClassMutable   1643/s                      2%                    --          -14%
set PerlClass           1916/s                     19%                   17%            --

Moose version 0.68 follows.

                          Rate new MooseClassMutable new MooseClassFast new MooseClassImmutable new PerlClass
new MooseClassMutable   15.4/s                    --               -84%                    -92%          -96%
new MooseClassFast      98.7/s                  541%                 --                    -48%          -76%
new MooseClassImmutable  190/s                 1138%                93%                      --          -54%
new PerlClass            412/s                 2579%               318%                    116%            --
                          Rate get MooseClassFast get MooseClassImmutable get MooseClassMutable get PerlClass
get MooseClassFast      1716/s                 --                     -2%                   -2%          -24%
get MooseClassImmutable 1743/s                 2%                      --                   -1%          -23%
get MooseClassMutable   1754/s                 2%                      1%                    --          -22%
get PerlClass           2261/s                32%                     30%                   29%            --
                          Rate set MooseClassFast set MooseClassMutable set MooseClassImmutable set PerlClass
set MooseClassFast      78.6/s                 --                  -95%                    -95%          -96%
set MooseClassMutable   1659/s              2011%                    --                     -1%          -15%
set MooseClassImmutable 1680/s              2038%                    1%                      --          -14%
set PerlClass           1950/s              2381%                   18%                     16%            --

Moose seems fast enough for me. If I realize how powerful Moose is results are great. I can persist class definition using make_immutable in most of cases and 190 thousand object constructions per second is enough. There is also big improvement in mutable version between 0.54 and 0.68 and 15 thousand per second is not terrible. Moose accessors are really fast and make_immutable have not any impact here. 1.7 million reads and 1.6 million writes per second is enough and my ugly handcrafted accessors can't make big difference here (2.2Mr/s and 1.9Mw/s). There is strange Class::Accessor:Fast setter result and I'm curious why. Anyway Moose itself performs well and there is not reason using it.

Absolute path resolver

2009-01-28T02:49:00.000-08:00

I was looked for some utility to resolve absolute path of file or directory. I found

namei

but it result returns in form

$ namei /etc/gdc
f: /etc/gdc
d /
d etc
l gdc -> /home/hynek/.gdc
d /
d home
d hynek
l .gdc -> work/tiger/bear/common/etc/gdc
  d work
  d tiger
  l bear -> bear.trunk/
    d bear.trunk
  d common
  d etc
  d gdc

but I suspected something more like

$ abs_path /etc/gdc
/home/hynek/work/tiger/bear.trunk/common/etc/gdc

I haven't found anything better than make alias in my ~/.bashrc

alias abs_path='perl -MCwd -le'\''print Cwd::abs_path($_) foreach @ARGV'\'

Edit: As ZD notice readlink -f does same work. Thanks. readlink -m and readlink -e works as well but differs if some parts missing. I have decided to use

alias abs_path='readlink -m'

Erlang history summary by J. Armstrong

2009-01-14T05:22:00.000-08:00

Joe Armstrong posts nice summary of Erlang history.

The transition was easy - they paid to do this. It became a real shipping project when they decided to use Erlang for the AXD301 - at that stage they put in the necessary $$$'s.

Now why did they choose Erlang for this project? - because all other alternatives had failed - ie it was not the strength of Erlang that was the deciding factor - rather the non-existence of alternatives.

Now how come the Erlang stuff was developed in the first place?

This was a happy accident - In the early 1980's a computer science lab was formed - most of the guys in the newly formed lab had zero experience with technology transfer, so we all thought that all we had to do was "invert stuff" and then "sell the idea to the management" nobody told us that this was like permanently banging your hand against a brick wall.

Inventing stuff is the easy bit ...

The selling stuff was tricky - we were very bad at this but very optimistic (still am :-) - we made all the classic mistakes - insulting people - getting into technical wars -

The turning point came when Erlang was banned - at the time we were very pissed off but like most carefull considered management decsions the net result was the exact opposite of what was planned - the consequences of the ban were difficult to forsee - but chaos was created - so things changed rapidly.

Thinking back the *important* things were:
enthusiasm and optimism (believe in what you do)
serendipity
chaos
smart people
finance

I think we systematically under-rate the significance of chance and chaos. Most significant change takes place in very short time periods of chaos. Erlang had many periods when nothing happened for years then rapid changes could take place in very short time periods, always when a crisis occurred (ie Erlang was banned, a big project failed etc).

Moral - forget about careful planning and move quickly when a crisis occurs - trust your gut feelings.

Cheers

/Joe Armstrong

History of Erlang

2009-01-13T07:19:00.000-08:00

Joe Armstrong wrote nice article about Erlang History. It's worth reading whole article but I found most funny and nice part it's end.

6.4 Finally

It is perhaps interesting to note that the two most significant factors that led to the spread of Erlang were:

The collapse of the AXE-N project.

The Erlang ban.

Both of these factors were outside our control and were unplanned. These factors were far more significant than all the things we did plan for and were within our control. We were fortuitously able to take advantage of the collapse of the AXE-N project by rushing in when the project failed. That we were able to do so was more a matter of luck than planning. Had the collapse occurred at a different site then this would not have happened. We were able to step in only because the collapse of the project happened in the building where we worked so we knew all about it. Eventually Ericsson did the right thing (using the right technology for the job) for the wrong reasons (competing technologies failed). One day I hope they will do the right things for the right reasons.

My interests

2008-12-17T03:00:00.000-08:00

Matin Hassman in his blog post wrote about the Wordle. This is nice toy tool to for generating “word clouds” from text. I have generated two from my bookmarkas and shared items in google reader.

Bookmarks

Shared Items

Beust Challenge in Erlang

2008-09-03T09:01:00.000-07:00

There was a challenge posted by Cedric Beust. I tried reuse my older solution of permutation generator and aplly this aproach on this issue. Here is result:

-module(cbchallenge).

-export([test/1]).

combine(Sufix, 0, Acc, _L) ->
   accept(list_to_integer(lists:reverse(Sufix)), Acc);
combine(Sufix, N, Acc, L) ->
   combine(Sufix, N, Acc, L, []).

combine(_Sufix, _N, Acc, [], _D) -> Acc;
combine(Sufix, N, Acc, [X | T], D) ->
   combine(Sufix, N,
     _NewAcc = combine([X | Sufix], N - 1, Acc,
         lists:reverse(D, T)),
     T, [X | D]).

accept(X, {Count, undefined}) -> {Count + 1, {X, 0}};
accept(X, {Count, {Last, MaxDistance}})
   when X - Last > MaxDistance ->
   {Count + 1, {X, X - Last}};
accept(X, {Count, {_, MaxDistance}}) ->
   {Count + 1, {X, MaxDistance}}.

count(Log10) ->
   {Count, {_, MaxDistance}} = combine([], Log10,
     {0, undefined}, lists:seq($1, $9),
     [$0]),
   {Count, MaxDistance}.

test(MaxLog10) ->
   lists:foldl(fun (Log10, AccIn) ->
   collectResult(count(Log10), AccIn)
  end,
  {0, 0}, lists:seq(1, MaxLog10)).

collectResult({Count, Max}, {OldCount, OldMax})
   when Max > OldMax ->
   {OldCount + Count, Max};
collectResult({Count, _}, {OldCount, Max}) ->
   {OldCount + Count, Max}.

It is more than twice faster than Kevin's crazybob solution on my laptop (kevin ~18s, mine ~8.5s).

> [{X, timer:tc(cbchallenge, test, [X])} || X<-lists:seq(1,10)].
[{1,{16,{9,1}}},
 {2,{52,{90,2}}},
 {3,{420,{738,11}}},
 {4,{3073,{5274,105}}},
 {5,{20019,{32490,1047}}},
 {6,{105757,{168570,10469}}},
 {7,{459955,{712890,104691}}},
 {8,{1587747,{2345850,1046913}}},
 {9,{4522205,{5611770,10469135}}},
 {10,{8563146,{8877690,104691357}}}]

I guess, there is much more faster solution closer to original crazybob's solution which is arithmetical and applicable just only to numbers. My approach is applicable to any non repeated combination of members of any set, but slower.

Arc - mostly macros and syntactic sugar

2008-02-01T02:54:00.000-08:00

Stefan Tilkov said about Arc:

After a quick glance at the tutorial, the most intriguing bit seems to be the support for macros, which work (almost) like function definitions. Interesting, but nothing that gets me overly excited.

I have same experinece. I read tutorial two days ago and I think it is mostly only scheme with macros and syntactic sugar. I am not so much familiar with lisp and scheme, but I think there is nothing what can't be done almost same simply in scheme or other lisp dialects. Updated: Steve dekorte think similar:

My own impression of Arc is that it's not significantly different from Scheme.

How much cores are using WF solutions

2007-11-02T02:42:00.000-07:00

Tim Bray published WF XI: Results and I would like to know how much of all these CPU cores each solution uses. Than I compute this table:

Name            Language    Elapsed     User        System  Parallel CPU work
-----------------------------------------------------------------------------
clv5            Gawk        46.73       40.63       6.1         1
tbray5          Erlang      01:04.32    35:33.35    00:45.84    33.88
wfinder1_1      Erlang      6.46        34.07       8.02        6.52
report-counts   Ruby        01:43.71    01:27.11    00:16.60    1
?               Groovy      02:21.83    02:22.97    00:19.95    1.15
wf_p            Ruby        50.16       37.58       12.5        1
wf-2            Python      41.04       34.8        6.24        1
wf-6(2)         Python      16.91       3.62        1.86        0.32
wf-6(4)         Python      9.08        3.66        1.89        0.61
wf-6(8)         Python      5.81        *           *           *
wf-6(16)        Python      4.38        *           *           *
wf              OCaml       49.69       41.94       7.75        1
widefinder      PHP         01:29.81    01:23.10    00:06.71    1
wf_pichi3       Erlang      8.28        51.98       9.38        7.41
tbray5          Erlang      00:20.74    03:51.33    00:08:00    34.3

Nice, that erlang implementations can use cores well, but in this task is not so much good generally. Erlang manages parallel processes well, but those processes can be better written in other languages and used as ports. Especially when this task is string operations on big amount of data.

Faster than ruby but scalable

2007-10-28T10:52:00.000-07:00

Update 2007-11-01: Correction of typo in wf_pichi3.erl.

@@ -14,7 +14,7 @@

 -compile([native]).

-main([File]) -> start(File), halt().
+main([File]) -> start_bmets(File), halt().

 start_bmets(FileName) ->
     {ok, F} = nlt_reader:open(FileName),

I worked on Wide Finder Project again. But what happen? I improved Anders Nygren's code with suggestion from my previous blog and also big suggestion from Caoyuan's blog. First I made some bricks: chunk_reder.erl with read ahead reading and consequential support, nlt_reader.erl with concurrent new line terminated block splitter and catenator and file_map_reduce.erl engine. And I plugged it together in wf_pichi3.erl wide finder. And what is great? It's about 40% faster on single core than ruby code and still scalable:

$ time ruby1.8 tbray.rb o1M.ap
8900: 2006/09/29/Dynamic-IDE
2000: 2006/07/28/Open-Data
1300: 2003/07/25/NotGaming
800: 2006/01/31/Data-Protection
800: 2003/09/18/NXML
800: 2003/10/16/Debbie
700: 2003/06/23/SamsPie
600: 2006/01/08/No-New-XML-Languages
600: 2005/11/03/Cars-and-Office-Suites
600: 2005/07/27/Atomic-RSS

real    0m7.469s
user    0m6.528s
sys     0m0.940s
$ time erl -noshell -run wf_pichi3 main o1M.ap
8900: 2006/09/29/Dynamic-IDE
2000: 2006/07/28/Open-Data
1300: 2003/07/25/NotGaming
800: 2003/09/18/NXML
800: 2003/10/16/Debbie
800: 2006/01/31/Data-Protection
700: 2003/06/23/SamsPie
600: 2006/01/08/No-New-XML-Languages
600: 2006/09/07/JRuby-guys
600: 2005/07/27/Atomic-RSS

real    0m5.370s
user    0m4.412s
sys     0m0.952s

It's big improvement from my last code, about 365% ;-) Good thing, that it's nice jigsaw and very powerful. I think, it is just what Tim Bray want when started Wide Finder Project.

Scalable splitting is possible

2007-10-27T02:40:00.000-07:00

In my previous post I thought that reading and splitting are unscalable processes. It's not true. Reading is scalable, but on current HW it is not useful, because sequential reading is more than twenty times faster than random from disks. But what about splitting and concatenating read chunk by new line. Splitting I can do in parallel and what concatenating? Yes, I can if can keep sequential information and send parts to correct process. Than I wrote scatter-getter algorithm with splitter and concatenator. I was also changed fold-reduce to map-reduce. Code is separated to three modules. Main module is tbray_pichi1

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Yet another Erlang solution to Tim Bray's Wide Finder project
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Author: Hynek (Pichi) Vychodil (http://pichis_blog.blogspot.com/), 23 October 2007.

-module(tbray_pichi1).

-export([main/1, start/1, start/2, start/3]).

start(FileName, ChunkSize) ->
    start(FileName, ChunkSize,
          erlang:system_info(schedulers) * 8).

start(FileName) -> start(FileName, 1024 * 32).

main([FileName, ChunkSize, N]) ->
    start(FileName, list_to_integer(ChunkSize),
          list_to_integer(N)),
    halt();
main([FileName, ChunkSize]) ->
    start(FileName, list_to_integer(ChunkSize)), halt();
main([FileName]) -> start(FileName), halt().

start(FileName, ChunkSize, N) ->
    Start = now(),
    Result = nlt_map_reduce:map_reduce(
        FileName,
        _Mapper = fun (Chunk) ->
                scan(binary_to_list(Chunk), dict:new())
            end,
        _Reducer = fun (A, B) ->
                dict:merge(fun (_, V1, V2) -> V1 + V2 end, A, B)
            end,
        ChunkSize, N),
    Delta = timer:now_diff(now(), Start) / 1000,
    print_result(Result),
    if Delta > 1000 ->
           io:format("Time: ~.3f s~n", [Delta / 1000]);
       true -> io:format("Time: ~.3f ms~n", [Delta])
    end,
    ok.

print_result(Dict) ->
    [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10 | _] =
        lists:reverse(lists:keysort(2, dict:to_list(Dict))),
    lists:foreach(fun ({Word, Count}) ->
                          io:format("~p get requests for ~s~n", [Count, Word])
                  end,
                  [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10]).

scan("GET /ongoing/When/" ++
       [_, _, _, $x, $/, Y1, Y2, Y3, Y4, $/, M1, M2, $/, D1,
        D2, $/
        | Rest],
     Dict) ->
    case scan_key(Rest) of
      {[_ | _] = Key, NewRest} ->
          scan(NewRest,
               dict:update_counter(
                    [Y1, Y2, Y3, Y4, $/, M1, M2, $/, D1, D2, $/ | Key],
                    1, Dict));
      {[], NewRest} -> scan(NewRest, Dict)
    end;
scan([_ | Rest], Dict) -> scan(Rest, Dict);
scan([], Dict) -> Dict.

scan_key(L) -> scan_key(L, []).

scan_key([$\s | Rest], Key) ->
    {lists:reverse(Key), Rest};
scan_key([$\n | Rest], _) -> {[], Rest};
scan_key([$. | Rest], _) -> {[], Rest};
scan_key([C | Rest], Key) -> scan_key(Rest, [C | Key]);
scan_key([], _) -> {[], []}.

Second new line terminated chunks map reducer (nlt_map_reduce)

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% MapReduce for new line terminated blocks of file
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Author: Hynek (Pichi) Vychodil (http://pichis_blog.blogspot.com/), 22 October 2007.

-module(nlt_map_reduce).

%-compile([native, {hipe, [o3]}]).
-import(chunk_reader).

-export([map_reduce/5]).

% mapper reducer process info
-record(map_reduce,
        {mapper, reducer, result = '', data, receiver}).

% process spawner loop info
-record(context,
        {mr, cr, proc = [], n = 0, max = 1, maxR = 1}).

map_reduce(FileName, Mapper, Reducer, ChunkSize, N) ->
    {ok, ChunkReader} = chunk_reader:open(FileName,
                                          ChunkSize),
    Result = start(#map_reduce{mapper = Mapper,
                               reducer =
                                   fun ('', A) -> A;
                                       (A, '') -> A;
                                       (A, B) -> Reducer(A, B)
                                   end,
                               receiver = self()},
                   ChunkReader, N),
    ok = chunk_reader:close(ChunkReader),
    Result.

start(MR, ChunkReader, N) ->
    spawn_proc(#context{mr = MR, cr = ChunkReader, max = N},
               chunk_reader:read(ChunkReader), you_first).

loop(#context{cr = ChunkReader, proc = [{_, Last} | _],
              n = N, max = Max} =
         C)
    when N < Max ->
    receive
      map_done -> loop(C#context{n = N - 1})
      after 0 ->
                spawn_proc(C, chunk_reader:read(ChunkReader), Last)
    end;
loop(#context{n = N} = C) ->
    receive map_done -> loop(C#context{n = N - 1}) end.

spawn_proc(#context{mr = MR, n = N, proc = Proc,
                    maxR = R} =
               C,
           {ok, Chunk}, Last) ->
    loop(C#context{
           proc = send_result_requests(
                [{0, spawn_opt(fun () ->
                            split(MR#map_reduce{data = Chunk}, Last)
                        end,
                        [link])}
                | Proc]),
           n = N + 1,
           maxR =
               if R > N + 1 -> R;
                  true -> N + 1
               end});
spawn_proc(_, eof, you_first) ->
    '';                     % empty file
spawn_proc(#context{proc = Proc}, eof,
           Last) ->         % finalise
    Last ! you_last,
    send_final_result_requests(Proc),
    wait_for_result().

wait_for_result() ->
    receive
      {result, Result} -> Result;
      _ -> wait_for_result()  % clear pending messages if any
    end.

send_result_requests([{L1, P1}, {L1, P2} | T]) ->
    P2 ! {send_result_to, P1, L1},
    send_result_requests([{L1 + 1, P1} | T]);
send_result_requests(T) -> T.

send_final_result_requests(T) ->
    [{L1, P} | R] = lists:reverse(T),
    L = send_final_result_requests_i(L1, P, R),
    P ! {send_result_to, self(), L}.

send_final_result_requests_i(L, P, [{L1, P1} | T]) ->
    P1 ! {send_result_to, P, L1},
    send_final_result_requests_i(L + 1, P, T);
send_final_result_requests_i(L, _, []) -> L.

% mapper reducer process states
split(#map_reduce{data = Data} = MR, you_first) ->
    map_it(MR#map_reduce{data = join_next(Data)});
split(#map_reduce{data = Data} = MR, PrevPid) ->
    case split_on_nl(Data) of
      {_, none} ->    % do nothing yourself, send it
          PrevPid ! {your_next_part, join_next(Data)},
          map_done(MR#map_reduce{data = done});
      {Line, Rest} ->
          PrevPid ! {your_next_part, Line},
          map_it(MR#map_reduce{data = join_next(Rest)})
    end.

join_next(Data) ->
    receive
      {your_next_part, Next} -> <<Data/binary, Next/binary>>;
      you_last -> Data
    end.

map_it(#map_reduce{mapper = Mapper, data = Data} =
           MR) ->
    map_done(MR#map_reduce{data = done,
                           result = Mapper(Data)}).

map_done(#map_reduce{receiver = Master} = MR) ->
    Master ! map_done,      % notice master you done map
    reduce_and_wait(MR, 0).

reduce_and_wait(#map_reduce{result = Acc,
                            reducer = Reducer} =
                    MR,
                N) ->
    receive
      {send_result_to, Receiver, WaitForN} ->
          reduce(MR#map_reduce{receiver = Receiver}, N, WaitForN);
      {result, Result} ->
          reduce_and_wait(MR#map_reduce{result =
                                            Reducer(Acc, Result)},
                          N + 1)
    end.

reduce(#map_reduce{result = Acc, reducer = Reducer} =
           MR,
       N, WaitForN)
    when N < WaitForN ->
    receive
      {result, Result} ->
          reduce(MR#map_reduce{result = Reducer(Acc, Result)},
                 N + 1, WaitForN)
    end;
reduce(#map_reduce{receiver = Receiver,
                   result = Result},
       _, _) ->
    Receiver ! {result, Result}.    % We are finished

%splitter
split_on_nl(B) -> split_on_nl(B, 0, size(B)).

split_on_nl(B, N, S) when N < S ->
    case B of
      <<Line:N/binary, $\n, Tail/binary>> -> {Line, Tail};
      _ -> split_on_nl(B, N + 1, S)
    end;
split_on_nl(B, _, _) -> {B, none}.

And last is read ahead chunk reader

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Chunk reader process with read ahead
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Author: Hynek (Pichi) Vychodil (http://pichis_blog.blogspot.com/), 22 October 2007.

-module(chunk_reader).

-export([close/1, open/2, read/1]).

-export([test_read/2]).

% how many second have to wait for response
-define(TIMEOUT, 60).

% open(FileName, ChunkSize) -> {ok, ChunkReader}
open(FileName, ChunkSize) ->
    M = self(),
    {ok,
     {chunk_reader,
      spawn_opt(fun () ->
                        {ok, File} = file:open(FileName, [read, raw, binary]),
                        process_flag(trap_exit, true),
                        loop(M, File, file:read(File, ChunkSize), ChunkSize)
                end,
                [link, {priority, high}])}}.

% close(ChunkReader) -> ok | {error, invalid}
close({chunk_reader, Pid}) when is_pid(Pid) ->
    case is_process_alive(Pid) of
      true -> Pid ! close, ok;
      false -> {error, invalid}
    end.

% read(ChunkReader) -> eof | {ok, Data} | {error, invalid | closed}
read({chunk_reader, Pid}) when is_pid(Pid) ->
    case is_process_alive(Pid) of
      true -> Pid ! {read, self()}, wait_response(Pid, 0);
      false -> {error, invalid}
    end.

wait_response(Pid, N) when N < (?TIMEOUT) ->
    receive
      {ok, _} = Msg -> Msg;
      eof -> eof
      after 1000 ->   % take it long?
                case is_process_alive(Pid) of
                  true -> wait_response(Pid, N + 1);
                  false -> {error, closed}
                end
    end;
wait_response(_, _) -> {error, timeout}.

loop(Master, File, Chunk, ChunkSize) ->
    receive
      {read, From} ->
          From ! Chunk,
          case Chunk of
            {ok, _} ->
                loop(Master, File, file:read(File, ChunkSize),
                     ChunkSize);
            eof -> file:close(File), eof_loop(Master, From)
          end;
      close -> file:close(File);
      {'EXIT', Master, _} -> file:close(File);
      _ ->
          loop(Master, File, Chunk, ChunkSize)  % ignore unknow
    end.

eof_loop(Master) ->  % wait for eof request
    receive
      {read, From} -> From ! eof, eof_loop(Master);
      close -> ok;
      {'EXIT', Master, _} -> ok;
      _ -> eof_loop(Master)
    end.

% speed testing function
% test_read(FileName, ChunkSize) -> ok | {error, invalid}
test_read(FileName, ChunkSize) ->
    {ok, File} = open(FileName, ChunkSize),
    eof = test_read_loop(File, read(File)),
    close(File).

test_read_loop(File, {ok, _}) ->
    test_read_loop(File, read(File));
test_read_loop(_, eof) -> eof.

But nlt_map_reduce code is too complicated, bad readable and what is the worst, 20% slower on single core. All this indicate, that there is some problem and I think it is dictionary sending between processes. Dictionary is copied every chunk and it cost to much. Then I want rewrite it to more fancy code and fall back to fold-reduce concept, because this concept send less dictionaries.

Wide Finder Project - fold&reduce

2007-10-21T10:38:00.000-07:00

I have made new version of Wide Finder Project. I was inspired by last Caoyuan's last work. I thought about i/o operation too, but I think, parallelisation of file reading is not good idea. Instead of this I tried split file reading and new line finding in two independent processes. One process which read and one which searching for new line. It's looks expensive, send big messages, but I send binaries and binaries less than 64 bytes are not copied, but only pointers passed. I also look for new line from head, because I think binary splitting is faster when first part is smaller than second. Second part can be keep on its place and only pointer is moved and smaller first part is copied to new position. But when glue second part from previous read chunk with first part of current, I must copy bigger part and this is expensive. It looks like same as Caoyuan do, but I don't do it in splitter, but in worker. Both parts I send as binary apart. It's cheap. Why all this? Make minimal work in one process. One process only reads as fast as possible. One process splitting by new line and don't gluing and all other work I can do in parallel. But when one splitter calls reader for new chunk, it must not wait for reader until it read next chunk. Better if reader have next chunk prepared. And splitter dtto. Splitter must have chunk split prepared before any worker calls for new parts. Then I made read ahead file reader and chunk splitter.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Chunk reader process with read ahead
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

file_open(FileName, ChunkSize, chunk) ->        % raw chunks
M = self(),
{ok, {chunk_reader, spawn_link(fun() ->
       {ok, File} = file:open(FileName, [read, raw, binary]),
       process_flag(trap_exit, true),
       process_flag(priority, high),
       file_loop(M, File, file:read(File, ChunkSize), ChunkSize)
   end)}};
file_open(FileName, ChunkSize, nlt_chunk) ->    % new line terminated chunks
M = self(),
{ok, {nlt_chunk_reader, spawn_link(fun() ->
       {ok, CR} = file_open(FileName, ChunkSize, chunk),
       process_flag(trap_exit, true),
       process_flag(priority, high),
       {ok, First_Read} = file_read(CR),
       cr_loop(
           M,
           CR,
           cr_read_n_split(CR, First_Read, file_read(CR)))
   end)}}.

file_read({Type, Pid}) when Type == chunk_reader; Type == nlt_chunk_reader ->
case is_process_alive(Pid) of
   true ->
       Pid ! {read, self()},
       receive
           {ok, B} -> {ok, B};
           eof -> eof
       after 60000 -> timeout   % Possible race condition with is_process_alive
       end;
   false -> error
end.

file_close({Type, Pid}) when Type == chunk_reader; Type == nlt_chunk_reader ->
case is_process_alive(Pid) of
   true -> Pid ! close, ok;
   false -> error
end.

file_loop(Master, File, Chunk, ChunkSize) ->
receive
   {read, From} ->
       From ! Chunk,
       case Chunk of
           {ok, _} ->
               file_loop(Master, File, file:read(File, ChunkSize), ChunkSize);
           eof ->
               file:close(File),
               file_eof_loop(Master)
       end;
   close -> file:close(File);
   {'EXIT', Master, _} -> file:close(File);
   _ -> file_loop(Master, File, Chunk, ChunkSize)  % ignore unknow
end.

file_eof_loop(Master) ->  % wait for eof request
receive
   {read, From} ->
       From ! eof,
       file_eof_loop(Master);
   close -> ok;
   {'EXIT', Master, _} -> ok;
   _ -> file_eof_loop(Master)
end.

cr_loop(Master, CR, {Prev, Line, Next}) ->
receive
   {read, From} ->
       From ! {ok, {Prev, Line}},
       case Next of
           _ when is_binary(Next) ->
               cr_loop(Master, CR, cr_read_n_split(CR, Next, file_read(CR)));
           eof ->
               file_close(CR),
               file_eof_loop(Master)
       end;
   close -> file_close(CR);
   {'EXIT', Master, _} -> file_close(CR);
   _ -> cr_loop(Master, CR, {Prev, Line, Next})    % ignore unknow
end.

cr_read_n_split(CR, Prev, {ok, B}) ->
case split_on_nl(B) of
   {Line, Rest} when is_binary(Rest) ->    % nonempty remaining part
       { Prev, Line, Rest };
   {Line, none} -> % new line not found, read again, should be very rare
       cr_read_n_split(CR, <<Prev/binary, Line/binary>>, file_read(CR))
end;
cr_read_n_split(_CR, Prev, eof) ->
{<<>>, Prev, eof}.  % easier joining at this order

split_on_nl(B) -> split_on_nl(B, 0, size(B)).

split_on_nl(B, N, S) when N < S ->
case B of
   <<Line:N/binary, $\n, Tail/binary>> -> {Line, Tail};
   _ -> split_on_nl(B, N+1, S)
end;
split_on_nl(B, _, _) -> {B, none}.

% speed testing functions
file_test_read(FileName, ChunkSize, Type) ->
{ok, File} = file_open(FileName, ChunkSize, Type),
eof = file_test_read_loop(File, file_read(File)),
file_close(File).

file_test_read_loop(File, {ok, _}) ->
file_test_read_loop(File, file_read(File));
file_test_read_loop(_, eof) ->
eof.

When I have this file like devices, I thought about Tim Bray's request more readable and cleaner code. So what I want to do? Some like map_reduce but not exactly map_reduce. It looks like fold_reduce. I want fold over each chunk aka scan for some pattern and than I want collect all results and I want do it in parallel. Then I made fold_reduce operator over new line terminated chunk read from file, just fold_reduce_file.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Proof of concept of fold&reduce
% on file by new line terminated chunks
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

-record(context, {acc,
               chunkNum,
               processedNum = 0,
               reducer}).

fold_reduce_file(FileName, Acc0, Folderer, Reducer, ChunkSize, N) ->
 {ok, CR} = file_open(FileName, ChunkSize, nlt_chunk),
 M = self(),
 do_n(fun() ->
     spawn_link(fun()-> folderer(CR, Acc0, Folderer, M) end)
     end, 0, N),
 Result = collect_loop(#context{
     acc=Acc0,
     chunkNum=N,
     reducer=Reducer}),
 file_close(CR),
 Result.

do_n(What, Start, Stop) when Start < Stop -> What(), do_n(What, Start+1, Stop);
do_n(_, _, _) -> ok.

folderer(CR, Acc0, Folderer, Collector) ->
 case file_read(CR) of
     {ok, {A, B}} ->
         folderer(CR, Folderer(
                 Acc0,
                 binary_to_list(A) ++ binary_to_list(B)
             ), Folderer, Collector);
     eof ->
         Collector ! {result, Acc0}
 end.

collect_loop(#context{acc=Acc0,
                   chunkNum=ChunkNum,
                   processedNum=ProcessedNum,
                   reducer=Reducer}=Context) ->
 case ProcessedNum of
     ChunkNum ->
         Acc0;
     _ ->
         receive
             {result, Result} ->
                 collect_loop(Context#context{
                     acc = Reducer(Acc0, Result),
                     processedNum = ProcessedNum+1})
         end
 end.

It looks complicated but, its only infrastructure. Now I have tool to make Tim Bray's exercise easy, but not only this one, but any similar task. Tim Bray's exercise implementation with this tool is here.

start(FileName) -> start(FileName, 1024*32, 1).
start(FileName, N) -> start(FileName, 1024*32, N).
start(FileName, ChunkSize, N) ->
 Start = now(),
 Result = fold_reduce_file(
     FileName,
     _Acc0 = dict:new(),
     _Folderer = fun(Acc, Chunk) -> scan(Chunk, Acc) end,
     _Reducer = fun(Acc, Result) -> dict:merge(
             fun(_,V1,V2) -> V1+V2 end,
             Acc,
             Result
         ) end,
     ChunkSize,
     N
 ),
 Delta = timer:now_diff(now(), Start) / 1000,
 print_result(Result),
 if
     Delta > 1000 -> io:format("Time: ~.3f s~n", [Delta/1000]);
     true -> io:format("Time: ~.3f ms~n", [Delta])
 end,
 ok.

print_result(Dict) ->
 [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10 | _] = lists:reverse(lists:keysort(2, dict:to_list(Dict))),
 lists:foreach(fun ({Word, Count}) ->
                       io:format("~p get requests for ~s~n", [Count, Word])
               end, [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10]).

scan("GET /ongoing/When/" ++ [_,_,_,$x,$/,Y1,Y2,Y3,Y4,$/,M1,M2,$/,D1,D2,$/|Rest], Dict) ->
 case scan_key(Rest) of
     {[_|_] = Key, NewRest} ->
         scan(NewRest, dict:update_counter([Y1,Y2,Y3,Y4,$/,M1,M2,$/,D1,D2,$/|Key], 1, Dict));
     {[], NewRest} -> scan(NewRest, Dict)
 end;
scan([_|Rest], Dict) -> scan(Rest, Dict);
scan([], Dict) -> Dict.

scan_key(L) -> scan_key(L, []).

scan_key([$ |Rest], Key) -> {lists:reverse(Key), Rest};
scan_key([$\n|Rest], _) -> {[], Rest};
scan_key([$.|Rest], _) -> {[], Rest};
scan_key([C|Rest], Key) -> scan_key(Rest, [C|Key]);
scan_key([], _) -> {[],[]}.

Good new is, this is faster than my tbray2 and also Caoyuan's tbray4 on single core. But I can't test it on multi core now. All source code for testing is bellow. When N is number of processor threads, interesting will be test N-1, N , 2*N-1 or 2*N folderer processes.

-module(tbray6).

%-compile([debug_info, native, {hipe, [o3]}]).

-export([start/1, start/2, start/3]).
-export([file_open/3, file_read/1, file_close/1, file_test_read/3]).
-export([fold_reduce_file/6]).

start(FileName) -> start(FileName, 1024*32, 1).
start(FileName, N) -> start(FileName, 1024*32, N).
start(FileName, ChunkSize, N) ->
 Start = now(),
 Result = fold_reduce_file(
     FileName,
     _Acc0 = dict:new(),
     _Folderer = fun(Acc, Chunk) -> scan(Chunk, Acc) end,
     _Reducer = fun(Acc, Result) -> dict:merge(
             fun(_,V1,V2) -> V1+V2 end,
             Acc,
             Result
         ) end,
     ChunkSize,
     N
 ),
 Delta = timer:now_diff(now(), Start) / 1000,
 print_result(Result),
 if
     Delta > 1000 -> io:format("Time: ~.3f s~n", [Delta/1000]);
     true -> io:format("Time: ~.3f ms~n", [Delta])
 end,
 ok.

print_result(Dict) ->
 [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10 | _] = lists:reverse(lists:keysort(2, dict:to_list(Dict))),
 lists:foreach(fun ({Word, Count}) ->
                       io:format("~p get requests for ~s~n", [Count, Word])
               end, [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10]).

scan("GET /ongoing/When/" ++ [_,_,_,$x,$/,Y1,Y2,Y3,Y4,$/,M1,M2,$/,D1,D2,$/|Rest], Dict) ->
 case scan_key(Rest) of
     {[_|_] = Key, NewRest} ->
         scan(NewRest, dict:update_counter([Y1,Y2,Y3,Y4,$/,M1,M2,$/,D1,D2,$/|Key], 1, Dict));
     {[], NewRest} -> scan(NewRest, Dict)
 end;
scan([_|Rest], Dict) -> scan(Rest, Dict);
scan([], Dict) -> Dict.

scan_key(L) -> scan_key(L, []).

scan_key([$ |Rest], Key) -> {lists:reverse(Key), Rest};
scan_key([$\n|Rest], _) -> {[], Rest};
scan_key([$.|Rest], _) -> {[], Rest};
scan_key([C|Rest], Key) -> scan_key(Rest, [C|Key]);
scan_key([], _) -> {[],[]}.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Proof of concept of fold&reduce
% on file by new line terminated chunks
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

-record(context, {acc,
               chunkNum,
               processedNum = 0,
               reducer}).

fold_reduce_file(FileName, Acc0, Folderer, Reducer, ChunkSize, N) ->
 {ok, CR} = file_open(FileName, ChunkSize, nlt_chunk),
 M = self(),
 do_n(fun() ->
     spawn_link(fun()-> folderer(CR, Acc0, Folderer, M) end)
     end, 0, N),
 Result = collect_loop(#context{
     acc=Acc0,
     chunkNum=N,
     reducer=Reducer}),
 file_close(CR),
 Result.

do_n(What, Start, Stop) when Start < Stop -> What(), do_n(What, Start+1, Stop);
do_n(_, _, _) -> ok.

folderer(CR, Acc0, Folderer, Collector) ->
 case file_read(CR) of
     {ok, {A, B}} ->
         folderer(CR, Folderer(
                 Acc0,
                 binary_to_list(A) ++ binary_to_list(B)
             ), Folderer, Collector);
     eof ->
         Collector ! {result, Acc0}
 end.

collect_loop(#context{acc=Acc0,
                   chunkNum=ChunkNum,
                   processedNum=ProcessedNum,
                   reducer=Reducer}=Context) ->
 case ProcessedNum of
     ChunkNum ->
         Acc0;
     _ ->
         receive
             {result, Result} ->
                 collect_loop(Context#context{
                     acc = Reducer(Acc0, Result),
                     processedNum = ProcessedNum+1})
         end
 end.

%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%
% Chunk reader process with read ahead
%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%

file_open(FileName, ChunkSize, chunk) ->        % raw chunks
 M = self(),
 {ok, {chunk_reader, spawn_link(fun() ->
         {ok, File} = file:open(FileName, [read, raw, binary]),
         process_flag(trap_exit, true),
         process_flag(priority, high),
         file_loop(M, File, file:read(File, ChunkSize), ChunkSize)
     end)}};
file_open(FileName, ChunkSize, nlt_chunk) ->    % new line terminated chunks
 M = self(),
 {ok, {nlt_chunk_reader, spawn_link(fun() ->
         {ok, CR} = file_open(FileName, ChunkSize, chunk),
         process_flag(trap_exit, true),
         process_flag(priority, high),
         {ok, First_Read} = file_read(CR),
         cr_loop(
             M,
             CR,
             cr_read_n_split(CR, First_Read, file_read(CR)))
     end)}}.

file_read({Type, Pid}) when Type == chunk_reader; Type == nlt_chunk_reader ->
 case is_process_alive(Pid) of
     true ->
         Pid ! {read, self()},
         receive
             {ok, B} -> {ok, B};
             eof -> eof
         after 60000 -> timeout   % Possible race condition with is_process_alive
         end;
     false -> error
 end.

file_close({Type, Pid}) when Type == chunk_reader; Type == nlt_chunk_reader ->
 case is_process_alive(Pid) of
     true -> Pid ! close, ok;
     false -> error
 end.

file_loop(Master, File, Chunk, ChunkSize) ->
 receive
     {read, From} ->
         From ! Chunk,
         case Chunk of
             {ok, _} ->
                 file_loop(Master, File, file:read(File, ChunkSize), ChunkSize);
             eof ->
                 file:close(File),
                 file_eof_loop(Master)
         end;
     close -> file:close(File);
     {'EXIT', Master, _} -> file:close(File);
     _ -> file_loop(Master, File, Chunk, ChunkSize)  % ignore unknow
 end.

file_eof_loop(Master) ->  % wait for eof request
 receive
     {read, From} ->
         From ! eof,
         file_eof_loop(Master);
     close -> ok;
     {'EXIT', Master, _} -> ok;
     _ -> file_eof_loop(Master)
 end.

cr_loop(Master, CR, {Prev, Line, Next}) ->
 receive
     {read, From} ->
         From ! {ok, {Prev, Line}},
         case Next of
             _ when is_binary(Next) ->
                 cr_loop(Master, CR, cr_read_n_split(CR, Next, file_read(CR)));
             eof ->
                 file_close(CR),
                 file_eof_loop(Master)
         end;
     close -> file_close(CR);
     {'EXIT', Master, _} -> file_close(CR);
     _ -> cr_loop(Master, CR, {Prev, Line, Next})    % ignore unknow
 end.

cr_read_n_split(CR, Prev, {ok, B}) ->
 case split_on_nl(B) of
     {Line, Rest} when is_binary(Rest) ->    % nonempty remaining part
         { Prev, Line, Rest };
     {Line, none} -> % new line not found, read again, should be very rare
         cr_read_n_split(CR, <<Prev/binary, Line/binary>>, file_read(CR))
 end;
cr_read_n_split(_CR, Prev, eof) ->
 {<<>>, Prev, eof}.  % easier joining at this order

split_on_nl(B) -> split_on_nl(B, 0, size(B)).

split_on_nl(B, N, S) when N < S ->
 case B of
     <<Line:N/binary, $\n, Tail/binary>> -> {Line, Tail};
     _ -> split_on_nl(B, N+1, S)
 end;
split_on_nl(B, _, _) -> {B, none}.

% speed testing functions
file_test_read(FileName, ChunkSize, Type) ->
 {ok, File} = file_open(FileName, ChunkSize, Type),
 eof = file_test_read_loop(File, file_read(File)),
 file_close(File).

file_test_read_loop(File, {ok, _}) ->
 file_test_read_loop(File, file_read(File));
file_test_read_loop(_, eof) ->
 eof.

Is bfile faster than old erlang file?

2007-10-07T23:01:00.000-07:00

Steve Vinoski is using klacke’s bfile module in his Wide Finder Project work, but I don't know why bfile should be faster than erlang OTP file. Well, then I measured. I tried Steve's read test and my test on my old home desktop (model name : AMD Athlon(tm) processor, stepping : 2, cpu MHz : 1199.805, cache size : 256 KB).

-module(readold).
-export([start/1, start/2]).
-compile([native]).

scan_file(F, Readsize, Total) ->
   Rd = file:read(F, Readsize),
   case Rd of
       {ok, Bin} -> scan_file(F, Readsize, size(Bin)+Total);
       eof -> Total
   end.
scan_file(F, Readsize) -> scan_file(F, Readsize, 0).

start(File, Readsize) ->
   {ok, F} = file:open(File, [raw, binary, read]),
   T = scan_file(F, Readsize),
   io:format("read ~p bytes~n", [T]),
   file:close(F).
start(File) ->
   start(File, 512*1024).

And there are results here:

2> timer:tc(readold,start,["o1M.ap"]).
read 200995500 bytes
{1041306,ok}
3> timer:tc(readold,start,["o1M.ap"]).
read 200995500 bytes
{836876,ok}
4> c(readold).
{ok,readold}
5> timer:tc(readold,start,["o1M.ap"]).
read 200995500 bytes
{837501,ok}
6> timer:tc(read,start,["o1M.ap"]).  
read 200995500 bytes
{1353678,true}
7> timer:tc(read,start,["o1M.ap"]).
read 200995500 bytes
{1237174,true}
8> timer:tc(read,start,["o1M.ap"]).
read 200995500 bytes
{1318029,true}
9> timer:tc(readold,start,["o1M.ap"]).
read 200995500 bytes
{856662,ok}

In generally, I don't know why erlang's file should be slower. I don't know why bfile is 45% slower than file on my old home desktop, but why should be faster anywhere? I tested it on Linux, may be bfile using BSD file implementation is faster on Darwin aka BSD clone? The file implementation is fast enough on my Erlang/OTP R11B-5.

erlang-base-hipe.deb don't contain native compiled modules

2007-10-07T11:10:00.001-07:00

When I measured Caoyuan and my Wide Finder Project solution I found my binary solution faster than list, but Caoyuan measured reverse result. Then I looked why. Caoyuan is using MacOS X and I'm using Debian. I don't know if MacOS port of erlang has base modules native compiled (with HiPE) but I looked on my debian's:

5> proplists:get_value(compile, lists:module_info()).
[{options,[{inline,[{merge3_12,7},
                 {merge3_21,7},
                 {rmerge3_12,7},
                 {rmerge3_21,7}]},
        {inline,[{umerge3_12,8},
                 {umerge3_21,8},
                 {rumerge3_12a,7},
                 {rumerge3_12b,8}]},
        {inline,[{keymerge3_12,12},
                 {keymerge3_21,12},
                 {rkeymerge3_12,12},
                 {rkeymerge3_21,12}]},
        {inline,[{ukeymerge3_12,13},
                 {ukeymerge3_21,13},
                 {rukeymerge3_12a,11},
                 {rukeymerge3_21a,13},
                 {rukeymerge3_12b,12},
                 {rukeymerge3_21b,12}]},
        {cwd,"/tmp/buildd/erlang-11.b.5dfsg/lib/stdlib/src"},
        {outdir,"/tmp/buildd/erlang-11.b.5dfsg/lib/stdlib/src/../ebin"},
        {i,"/tmp/buildd/erlang-11.b.5dfsg/lib/stdlib/src/../include"},
        {i,"/tmp/buildd/erlang-11.b.5dfsg/lib/stdlib/src/../../kernel/include"},
        warn_obsolete_guard,
        debug_info,
        {inline,[{merge3_12,7},
                 {merge3_21,7},
                 {rmerge3_12,7},
                 {rmerge3_21,7}]},
        {inline,[{umerge3_12,8},
                 {umerge3_21,8},
                 {rumerge3_12a,7},
                 {rumerge3_12b,8}]},
        {inline,[{keymerge3_12,12},
                 {keymerge3_21,12},
                 {rkeymerge3_12,12},
                 {rkeymerge3_21,12}]},
        {inline,[{ukeymerge3_12,13},
                 {ukeymerge3_21,13},
                 {rukeymerge3_12a,11},
                 {rukeymerge3_21a,13},
                 {rukeymerge3_12b,12},
                 {rukeymerge3_21b,12}]}]},
{version,"4.4.5"},
{time,{2007,9,28,11,10,32}},
{source,"/tmp/buildd/erlang-11.b.5dfsg/lib/stdlib/src/lists.erl"}]

There isn't native option. If MacOS port is native compiled there can be this unexpected difference. I have tried to create my own erlang-base-hipe package from source one, but I don't know how to put the option on into the making process. I have tried: debian/rules configure-hipe but it didn't work so I searched where is native option used.

$ grep -r +native .
./lib/asn1/src/Makefile:ERL_COMPILE_FLAGS += +native
./lib/megaco/src/app/megaco.mk:ERL_COMPILE_FLAGS += +native
./lib/megaco/test/Makefile:ERL_COMPILE_FLAGS += +native -Dmegaco_hipe_special=true
./lib/megaco/examples/meas/Makefile:ERL_COMPILE_FLAGS += +native
./README:       erlc +native Module.erl

Then I tried

export ERL_COMPILE_FLAGS=+native

It didn't affect binary-erlang-base and only generated warning message, but during binary-erlang-base-hipe error occured.

erlc -W  +native +debug_info +debug_info +debug_info +warn_obsolete_guard -I/home/hynek/work/erlang-11.b.5dfsg/lib/stdlib/include -o../ebin yecc.erl
./yecc.erl:none: internal error in native_compile;
crash reason: {undef,[{hipe,compile,
                            [yecc,
                             [],
                             <<70,79,82,49,0,1,91,124,66,69,65,77,65,116,111,
...
97,252,150,236,255,7,193,199,127,8,0,0>>,
                             []]},
                      {compile,native_compile_1,1},
                      {compile,'-internal_comp/4-anonymous-1-',2},
                      {compile,fold_comp,3},
                      {compile,internal_comp,4},
                      {compile,internal,3}]}
make[4]: Leaving directory `/home/hynek/work/erlang-11.b.5dfsg/lib/parsetools/src'
make[3]: Leaving directory `/home/hynek/work/erlang- 11.b.5dfsg/lib/parsetools'
make[2]: Leaving directory `/home/hynek/work/erlang-11.b.5dfsg/lib'
make[1]: Leaving directory `/home/hynek/work/erlang-11.b.5dfsg'

Then I tested if yecc can be compiled native and it can be.

3> c("/usr/lib/erlang/lib/parsetools-1.4.1.1/src/yecc.erl", [native, {i, "/usr/lib/erlang/lib/stdlib-1.14.5/include/"}, {outdir, "."}]).
{ok,yecc}
4> proplists:get_value(compile, yecc:module_info()).                                                                                    
[{options,[{inline,[{compute_closure,3}]},
           {nowarn_unused_function,{function_name,2}},
           {inline,[{set_empty,0}]},
           {inline,[{set_member,2}]},
           {inline,[{set_delete,2}]},
           {inline,[{set_union,2}]},
           {inline,[{set_is_subset,2}]},
           {inline,[{is_terminal,2}]},
           native,
           {i,"/usr/lib/erlang/lib/stdlib-1.14.5/include/"},
           {outdir,"."},
           {inline,[{compute_closure,3}]},
           {nowarn_unused_function,{function_name,2}},
           {inline,[{set_empty,0}]},
           {inline,[{set_member,2}]},
           {inline,[{set_delete,2}]},
           {inline,[{set_union,2}]},
           {inline,[{set_is_subset,2}]},
           {inline,[{is_terminal,2}]}]},
 {version,"4.4.5"},
 {time,{2007,10,7,17,53,55}},
 {source,"/usr/lib/erlang/lib/parsetools-1.4.1.1/src/yecc.erl"}]

yecc (and almost all other modules) can be native compiled, but I don't know how to do it. I'm totally messed up by debian packaging system and don't know what dpkg-buildpackage does. It's difficult to do it with installed version because many packages need some special compiling options at least {i,"/usr/lib/erlang/lib/stdlib-1.14.5/include/"} for included .hrl files and so. I will be glad if anyone give me some advice how to make package with native compiled modules or how to recompile only modules from source package to make some workaround.

Binaries really faster than lists

2007-10-06T13:40:00.000-07:00

Caoyuan did his second round of Wide Finder Project using lists, but expected lists traversing is faster than binary traversing. But he made little mistake. He assumed, that timer:tc(tbray, travel_list, [binary_to_list(Bin)]) including binary_to_list(Bin) time, but it is not true. binary_to_list(Bin) is done before timer:tc call and result passed as parameter. But there are more performance problems in his solution. He often use lists:reverse and ++ which twice reverse first parameter too. Bigger memory usage of list and often reversing must cause bad performance. When I tried write same algorithm using binary I take 3 times speed up.


-module(tbray2).

-compile([native, {hipe,[o3]}]).

-export([start/1, start/2]).

-record(context, {main,
                 dict,
                 chunkNum,
                 processedNum = 0}).

start(FileName) -> start(FileName, 1024*32).
start(FileName, ChunkSize) ->
   Start = now(),
   Main = self(),
   Collector = spawn_link(fun () -> collect_loop(#context{main = Main,
                                                     dict = dict:new(),
                                                     processedNum = 0}) end),
   ChunkNum = foreach_chunk(
           fun(Chunk) ->
               spawn_link(fun() -> Collector ! scan_lines(Chunk) end)
           end,
           FileName,
           ChunkSize
       ),
   Collector ! {chunkNum, ChunkNum},
  
   %% don't terminate, wait here, until all tasks done.
   receive
       stop -> io:format("Time: ~p ms~n", [timer:now_diff(now(), Start) / 1000])
   end.

foreach_chunk(Fun, FileName, SizeOfChunk) ->
   {ok, File} = file:open(FileName, [raw, binary]),
   N = foreach_chunk(Fun, File, <<>>, SizeOfChunk, 0),
   file:close(File),
   N.

foreach_chunk(Fun, File, PrevRest, SizeOfChunk, N) ->
   {Chunk, Rest} = read_chunk(File, PrevRest, SizeOfChunk),
   Fun(Chunk),
   case Rest of
       <<>> -> N+1;
       _ -> foreach_chunk(Fun, File, Rest, SizeOfChunk, N+1)
   end.

read_chunk(File, PrevRest, N) ->
   case file:read(File, N) of
       {ok, B} ->
           {Line, Rest} = split_on_nl(B),
           Chunk = <<PrevRest/binary, Line/binary>>,
           case Rest of
               <<>> ->
                   read_chunk(File, Chunk, N);
               _ ->
                   {Chunk, Rest}
           end;
       eof ->
           {PrevRest, <<>>}
   end.

split_on_nl(B) -> split_on_nl(B, 0, size(B)).

split_on_nl(B, N, S) when N < S -> case B of
       <<Line:N/binary, $\n, Tail/binary>> -> {Line, Tail};
       _ -> split_on_nl(B, N+1, S)
   end;
split_on_nl(B, _, _) -> {B, <<>>}.

collect_loop(#context{main=Main,
                     dict=Dict,
                     chunkNum=ChunkNum,
                     processedNum=ProcessedNum}=Context) ->
   case ProcessedNum of
       ChunkNum ->
           print_result(Dict),
           Main ! stop;
       _ ->
           receive
               {chunkNum, N} -> collect_loop(Context#context{chunkNum = N});
               DictX ->
                   collect_loop(Context#context{
                       dict = dict:merge(fun (_, V1, V2) -> V1 + V2 end, Dict, DictX),
                       processedNum = ProcessedNum+1})
           end
   end.

print_result(Dict) ->
   [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10 | _] =
       lists:reverse(lists:keysort(2, dict:to_list(Dict))),
   lists:foreach(fun ({Word, Count}) ->
                         io:format("~p get requests for ~s~n", [Count, Word])
                 end, [R1, R2, R3, R4, R5, R6, R7, R8, R9, R10]).

scan_lines(Bin) -> scan_lines(Bin, dict:new()).
scan_lines(Bin, Dict) ->
   case naive_search(Bin) of
       {ok, Key, Rest} ->
           scan_lines(
               element(2, split_on_nl(Rest)),
               dict:update_counter(Key, 1, Dict));
       false -> Dict
   end.

%% naive_search(Binary()) -> false | {ok, Key, Rest}
naive_search(B) -> naive_search(B, 0, size(B)-18).

naive_search(B, N, S) when N < S ->
   case B of
       <<_:N/binary, "GET /ongoing/When/", Rest/binary>> ->
           case keyMatch(Rest) of
               Result = {ok, _Key, _Rest2} -> Result;
               false -> naive_search(Rest)
           end;
       _ -> naive_search(B, N+1, S)
   end;
naive_search(_, _, _) -> false.

%% keyMatch(Binary()) -> false | {ok, Key, Rest}
keyMatch(<<C, _/binary>>) when C == $ ; C == $. -> false;   % empty
keyMatch(B) -> keyMatch(B, 1, size(B)).

keyMatch(B, N, S) when N<S ->
   case B of
       % end with space
       <<Key:N/binary, $ , Rest/binary>> -> {ok, Key, Rest};
       <<_:N/binary, $., _/binary>> -> false;
       _ -> keyMatch(B, N+1, S)
   end;
keyMatch(_, _, _) -> false.

Result is less memory consuming and faster program. Problem is partitioned same way as Caoyuan did and I suppose that scale same way, but I can't test it because I don't have any multi core computer.

Keep your mailbox empty

2007-10-03T07:34:00.001-07:00

Matthias wrote Too much mail is bad for you. Yes, it's true. He solved problem by synchronous communication between producer and consumer, but this solution blocks producer and is not "appropriate solution in general" (second comment). I think, one good solution is make your own queue manager and keep process mailbox empty. It's principally solution what recommends Vlad in that discussion. And how to do it?

-module(mqueue).

-export([timed_run/3]).

producer(0, ConsumerPid, _WithAck) ->
  ConsumerPid ! {done, self()},
  receive
      done -> ok
  end;
producer(N, ConsumerPid, WithAck = false) ->
  ConsumerPid ! {msg, self()},
  producer(N-1, ConsumerPid, WithAck);
producer(N, ConsumerPid, WithAck = true) ->
  ConsumerPid ! {acked_msg, self()},
  receive
      ack -> ok
  end,
  producer(N-1, ConsumerPid, WithAck).

consumer(M, EchoPid) ->
  receive
      {msg, _From} ->
          call_echo(M, EchoPid),
          consumer(M, EchoPid);
      {acked_msg, From} ->
          From ! ack,
          call_echo(M, EchoPid),
          consumer(M, EchoPid);
      {done, From} ->
          EchoPid ! done,
          From ! done
  end.

queue_keeper_empty(ConsumerPid) ->
  receive
      {msg, _From} ->
          ConsumerPid ! {acked_msg, self()},
          queue_keeper_waiting(queue:new(), ConsumerPid);
      {done, From} ->
          ConsumerPid ! {done, self()},
          receive
              done ->
                  From ! done
          end
  end.

queue_keeper_waiting(Q, ConsumerPid) ->
  receive
      {msg, _From} ->
          queue_keeper_waiting(
              queue:in({acked_msg, self()}, Q),
              ConsumerPid
          );
      ack -> case queue:out(Q) of
              {{value, Msg}, Q1} ->
                  ConsumerPid ! Msg,
                  queue_keeper_waiting(Q1, ConsumerPid);
              {empty, Q} -> queue_keeper_empty(ConsumerPid)
          end
  end.

call_echo(0, _EchoPid) ->
  ok;
call_echo(M, EchoPid) ->
  EchoPid ! {hello, self()},
  receive
      hello -> call_echo(M-1, EchoPid)
  end.

echo() ->
  receive
      {Msg, From} ->
          From ! Msg,
          echo();
      done -> ok
  end.

run(N, M, WithAck) ->
  EchoPid     = spawn_link(fun echo/0),
  ConsumerPid = spawn_link(
                  fun () ->
                          consumer(M, EchoPid) end),
  case WithAck of
      true ->
          producer(N, ConsumerPid, WithAck);
      false ->
          producer(N, spawn_link(
                  fun() ->
                      queue_keeper_empty(ConsumerPid)
                  end
              ), WithAck)
  end.

time(F) ->
  Start = erlang:now(),
  F(),
  timer:now_diff(erlang:now(), Start).

timed_run(N, M, WithAck) ->
  time(fun() -> run(N, M, WithAck) end).

This solution is only about 25% slower than synchronous, but producer is not blocked.