That of course means you can't take data strictly from STDIN, but really that's a silly way to process 4.1GB of data!

If you really, really need to read it from a stream and output it in-order then you'll have to have a parent reading the stream, forking kids and then collecting the results in buffers so you can output in-order. Start with Parallel::ForkManager, which will handle giving out the work and then mix in some IO::Pipe and IO::Select for collecting the results. Be sure you divide the work into sizable chunks, forking a new child for each line isn't going to help very much!

Now go give it a try and don't come back until you have some code to post!

-sam

sort  --field-separator=,
      --key=1,1
      --numeric-sort
      --merge
      worker_output_*
| cut --delimiter=,
      --fields=2-
[download]

--merge is key.

Would Tie::File be helpful?
if you knew which line it came from you could slot right back in where it started.

tie @array, 'Tie::File', filename or die ...;

### at the end
my ($Toad, $LineIn) = $MagicWand -> HocusPocus
$array[$LineIn] = $Toad;
http://search.cpan.org/~mjd/Tie-File-0.96/lib/Tie/File.pm

Definitely not. The goal is to speed things up, not set the hard drive on fire.

This is untested on a multicore machine, but should be free from any deadlocking or syncing issues.

And it will definitely need tuning for best performance, but you could try this on your data and see how you fare?

Run as: scriptname inputfile >outputfile

#! perl -sw
use strict;
use threads;
use threads::shared;
use Thread::Queue;

sub worker {
    my( $Qin, $outRef ) = @_;
    while( my $in = $Qin->dequeue ) {
        my( $no, $text ) = split ':', $in, 2;
        ########## Manipulate data below ########
        my $out = $text;
        lock $outRef;
        $outRef->[ $no ] = $out;
    }
}

my @out :shared;
my $Qin = new Thread::Queue;

my @threads = map threads->create( \&worker, $Qin, \@out ), 1 .. 4;

my $last :shared = 0;

async {
    while( <> ) {
        sleep 1 while $Qin->pending > 100;
        $Qin->enqueue( "$.:$_" );
        lock $last;
        $last = $.;
    }
};

my $next = 1;

while( threads->list( threads->running ) ) {
    sleep 1 until defined $out[ $next ];
    print $out[ $next ];
    undef $out[ $next++ ];
}

while( $next < $last ) {
    sleep 1 until defined $out[ $next ];
    print $out[ $next ];
    undef $out[ $next++ ];
}

$_->join for @threads;
[download]

use strict;
use Thread::Pool;
#use Thread::Semaphore;

open( OUTFILE, ">/crnch_data/foo.csv" ) or die "Cannot open output";
print OUTFILE "\n";

#select(STDIN); $| = 1;
open( STDIN, "/usr/sbin/tcpdump -vttttnner /crnch_data/tcpdump_infile 
+|" );
select(OUTFILE); $| = 1;

my $res = "";
my $pool = Thread::Pool->new(
    {
       workers => 10,
       do      => \&do,
       stream  => \&monitor,
    }
);

$pool->job($_) while (<STDIN>);
$pool->shutdown;

sub do {
    chomp;
    my $input = $_;
    $input = shift;

    if ( $input =~ /(^\d\d\d\d-\d\d-\d\d .*)/ ) {
        if ( $input =~
/(.*) \(tos (\S+), ttl +(\d+), id (\d+), offset (\d+), flags ([\S\+]+)
+(?:, ([\S\+]+))?, proto +(\S+ \(\d+\)), length (\d+)(?:, .*?)?\) (.*)
+/
           )
        {
            $input = "$1 $2 $3 $4 $5 $6$7 $8 $9 $10";
        }
        else {
            if ( $input =~
/(.*) (((?:(\d{1,2}|[a-fA-F]{1,2}){2})(?::|-*)){6}) (\>) (((?:(\d{1,2}
+|[a-fA-F]{1,2}){2})(?::|-*)){6}), (.*?), length (\d+)\:(.*)/
               )
            {
                $input = "$1 $2 x $6 x x x x $10";
            }
            else {
                $res = "error";
            }
        }
    }
    else {
        $res = "error";
    }

    my @fields = split( " ", $input );
    my $timestamp   = $fields[0] . " " . $fields[1];
    my $microsecond = $fields[1];
    $timestamp   =~ s/(.*?)\.\d+$/\1/;
    $microsecond =~ s/(.*?)\.(\d+$)/\2/;
    my $sourcemac = $fields[2];
    my $destmac   = $fields[4];
    $destmac =~ s/,//g;
    $fields[4] =~ s/,$//;
    my $len = $fields[9];
    $len =~ s/:$//;
    my $tos = $fields[11];
    my $ttl     = $fields[12];
    my $id      = $fields[13];
    my $offset  = $fields[14];
    my $ipflags = $fields[15] . " " . $fields[16];

    $ipflags =~ s/\[(.*)\]/\1/g;

    my $sip = $fields[18];
    $sip =~ s/([^\.]+\.[^\.]+\.[^\.]+\.[^\.]+).*/\1/;
    $sip =~ s/:$//;

    my $sport = $fields[18];
    if ( $sport =~ /[^\.]+\.[^\.]+\.[^\.]+\.[^\.]+\.(.*)/ ) {
        $sport = $1;
        $sport =~ s/:$//;
    }
    else { $sport = "null"; }

    my $dip = $fields[20];

    $dip =~ s/([^\.]+\.[^\.]+\.[^\.]+\.[^\.]+).*/\1/;
    $dip =~ s/:$//;

    my $dport = $fields[20];
    if ( $dport =~ /[^\.]+\.[^\.]+\.[^\.]+\.[^\.]+\.(.*)/ ) {
        $dport = $1;
        $dport =~ s/:$//;
    }
    else { $dport = "null"; }

    my $proto = "null";
    my $flags = $fields[21] if ( $fields[21] =~ /[SRPU.]+/ );
    my $proto = "tcp" if ( $fields[21] =~ /[SRPU.]+/ );
    $_ = "//$timestamp//$microsecond//$sourcemac//$destmac//$sip//$dip
+//$sport//$dport//$proto//$flags//$len//$ttl//$id//$tos//$ipflags//$o
+ffset?";
}

sub monitor : locked {
    my $line = $_;

# my $semaphore = new Thread::Semaphore;
    $line = shift;
    unless ( $line eq "error" ) {
# $semaphore->down;
        print OUTFILE $line;
# $semaphore->up;
    }
}
close(OUTFILE);
[download]

The problem is that every worker writes to the file.
As written before you should bundle the writting in one seperate thread so that you need no symchronisation for the file.

Hi,

I have kind of similar need. Can you please paste some sample and easy (as I am not a pro in perl) code; how can I bundle the writing in one separate thread ?

Here's a cut down of some code i'm using which is based on Parallel::Fork::BossWorker to speed up some processing. You'll need to monkey with it to get it to do what you want...

my $Worker_count = 2;

fork_workers();

sub autoflush { select((select(shift), $| = 1)[0]) }

sub fork_workers {
    # worker sends pid to boss when ready for more work
    pipe my $from_workers, my $to_boss;
    autoflush($to_boss);
    my %workers;
    foreach (1 .. $Worker_count) {
    # pipe for boss to send 'work' to worker
    pipe my $from_boss, my $to_worker;
    # pipe for worker to send message back to boss
    pipe my $from_worker2, my $to_boss2;
    autoflush($to_worker);
    autoflush($to_boss2);
    my $pid = fork;
    die "failed to fork: $!" unless defined $pid;
    if ($pid) {
        $workers{$pid} = [$to_worker, $from_worker2];
        close $from_boss;
        close $to_boss2;
    } else {
        close $from_workers;
        close $from_worker2;
        close $to_worker;
        close STDIN;
        send_msg($to_boss, $$);
        worker($from_boss, $to_boss, $to_boss2);
        exit;
    }
    }
    close $to_boss;

    my $INT_handler = $SIG{INT};
    local $SIG{INT} = sub {
    close $from_workers;
    for (keys %workers) {
        my ($to_worker, $from_worker2) = @{$workers{$_}};
        close $to_worker;
        close $from_worker2;
    }
    $INT_handler->('INT');
    };

    my @work = get_work(); # ???

    while (my $pid = receive_msg($from_workers)) {
    my ($to_worker, $from_worker2) = @{$workers{$pid}};
    my $msg = receive_msg($from_worker2);
    if (@work) {
        my $whatever = something(@work);
        send_msg($to_worker, $whatever);
        @work = ();
    } else {
        close $to_worker;
        close $from_worker2;
        delete $workers{$pid};
    }
    if (length $msg) {
        # something
    }
    }
    close $from_workers;
}


sub worker {
    my ($from_boss, $to_boss, $to_boss2) = @_;
    while (my $whatever = receive_msg($from_boss)) {
    my $msg = something_else($whatever);
    send_msg($to_boss, $$);
    send_msg($to_boss2, $msg);
    }
}


sub receive_msg {
    my $fh = shift;
    local $/ = "\003\000";
    my $msg = <$fh>;
    chomp $msg if defined $msg;
    return $msg;
}


sub send_msg {
    my ($fh, $msg) = @_;
    print $fh $msg, "\003\000";
}
[download]

There is one issue with the above mechanism that i'd be interested in having resolved... is it _safe_ for multiple child proceses to write to the same pipe? In the above code the  send_msg($to_boss, $$) is used by each child to notify the parent that it has something for it to process. The code used by Parallel::Fork::BossWorker allows the children to write arbitrarily large amounts to this pipe, but i found that this could occasionally result in corrupted data when writing large amounts of data.

However by changing to the scheme used in the above code of only sending the pid (a small amount of data) down the shared pipe i've yet to see any corruption.

So... is it safe to write small amounts (less than some internal perlio buffer size) to a shared pipe; or, is it unsafe, and i've just yet to hit the problem?

Thanks for any tips! :-) alex.

All the performance problems with Thread::Pool are inherent in its over-engineered designed and convoluted implementation. There's very little you can do from the outside to improve it. Despite its 5-star (member of the club) CPAN rating, it should have been strangled at birth.

---

Examine what is said, not who speaks -- Silence betokens consent -- Love the truth but pardon error.

"Science is about questioning the status quo. Questioning authority".

In the absence of evidence, opinion is indistinguishable from prejudice.

"Too many [] have been sedated by an oppressive environment of political correctness and risk aversion."

Or, to look at it another way, you are passing the input a line at a time to the worker threads -- so two things are happening:

1. all of the thread communication overhead is being carried by every line -- I think this is the real issue.

2. you are streaming the output, so Thread::Pool arranges for the output of each job to be done in job order.

   Having looked at the code, the way it does this is quite cunning: when a do routine completes a job before earlier jobs have completed, the results are put to one side, and the thread loops round to do the next job. When a job completes and all earlier have completed, the thread outputs its results and any results put aside earlier which can now be output.

   Note that this means that if one job takes a very long time indeed, all the other threads will continue taking jobs and generating and storing the output. This is a Good Thing, because slower jobs are not blocking faster ones. However, I don't see a mechanism to stop jobs being dispatched if the streamed output backs up. But that could be because I haven't studied the code hard enough.

   So apart from the cost of all this cunning, I don't think this is the issue.

Without changing the structure of what you have too much, I'd change:

  $pool->job($_) while (<STDIN>);
[download]

to something that reads a bunch of lines (say a 100K bytes worth), something like:

  while (!eof(STDIN)) {
    my $buffer ;
    read STDIN, $buffer, 100*1024   or die "$!" ;
    if (!eof(STDIN)) {
      my $nl = rindex($buffer, "\n") + 1  or die "No \\n in lump ?" ;
      if (my $trim = $nl - length($buffer) {
        seek STDIN, $trim, 1   or die "Seek failed ??" ;
        substr($buffer, $trim) = '' ;
      } ;
    } ;

    $pool->job($buffer) ;
  } ;
[download]

(I haven't tested this, so caveat emptor !). Then the processing subroutine needs to break its input into lines and process into another buffer, and then return that lump of results. Something like:

  sub do {
    my ($lump) = @_ ;

    my $result = '' ;
    while ($lump =~ m/(.+)/g) { # Ignores blank lines
      my $line = $1 ;

      ... whatever ...

      $result .= ...whatever..  ."\n" ;
    } ;

    return $result ;
  } ;
[download]

(also untested). There's some saving to be made by removing my ($lump) = @_ and doing while ($_[0] =~ m/(.+)/g) directly instead.

This will amortise the job overhead over a reasonable chunk of lines. Obviously you can experiment to see if larger chunks means faster processing.

BTW, your do subroutine starts:

  sub do {
    chomp;
    my $input = $_;
    $input = shift;
[download]

which doesn't look right to me. The arguments are passed to do() in the usual way, so hacking around $_ appears redundant.

You are suffering from NOT buffering. This becomes obvious by following the data flow of the pipe in upstream direction:

1) Your perl script accepts the next line of input.

2) Tcpdump refills the pipe buffer and blocks.

3) Tcpdump requests a new "line" from tcpdump_infile and transforms it .

As a consequence piping as a cheap form of using more processors does not work efficiently. It results in small read requests to the disk of one line each. The delay between the small requests starves tcpdump and your perl script. You should optimise the earliest bottleneck first.

All disks like large read requests, so the solution is to add large buffers (e. g. 100 MB each) before and after tcpdump. This allows disk reading, tcpdump and the perl script to run in parallel. I like to use mbuffer for the task. Your input pipe could look like

mbuffer -i tcpdump_infile -m 100M -P90 -p10 | \
tcpdump |\
mbuffer -m 100M -P90 -p10 | \
perl
[download]

This gives you 100 MB buffers that write ou a large chunk when 90% full and start reading again when 10% full.

I'd check if tcpdump is the first bottleneck. It has to do a lot of work and I doubt it is multithreaded. "time tcpdump -options >/dev/null" gives the minimum run time for the perl script. Look at the load or "top" while doing this.

Recording smaller tcpdump_infiles or splitting them up at a good boundary allows running multiple tcpdump processes.

OTOH it might be justified to parse the tcpdump_infile directly from perl.

Define "too long".

What have you used to profile your current code and what are the bottle necks?

How long does a simple script that copies STDIN to STDOUT take on your system compared to the time you have available for the task?

Perhaps I'm missing something. If it's a static file (that is, not an input stream) you can start multiple worker processes each with a different byte offset into the file. Since you've established that it's newline-based, you can skip bytes until after the first newline (except for the worker at offset 0) to guarantee that you're processing a line at a time. Then your workers each build a "chapter" of output which you can then cat together later. These workers can be processes or threads. "Simple" seeks get you to the offset.

The reason I say this is that it's not clear if accepting pipe-ish data (STDIN) is part of the problem statement or part of the solution (a design approach). If the former...nevermind.