Object passing between threads, and Perl code snippets

A limitation of the Perl threads is that no variables can be shared between them. When a new thread gets created, it gets a copy of all the variables of the parent. Well, of all the plain Perl variables. With the XS extensions your luck may vary: the variables might get copied, might become undef, or just become broken (if the XS module is not threads-aware). Copying the XS variables requires a quite high overhead at all the other times, so Triceps doesn't do it and all the Triceps object become undefined in the new thread.

However there is a way to pass around certain objects through the Nexuses.

First, obviously, the Nexuses are intended to pass through the Rowops. These Rowops coming out of a nexus are not the same Rowop objects that went in. Rowop is a single-threaded object and can not be shared by two threads. Instead it gets converted to an internal form while in the nexus, and gets re-created, pointing to the same Row object and to the correct Label in the local facet.

Then, again obviously, the Facets get imported through the Nexus, together with their row types.

And two more types of objects can be exported through a Nexus: the RowTypes and TableTypes. They get exported through the options as in this example:

$fa = $owner->makeNexus(
    name => "nx1",
    labels => [
        one => $rt1,
        two => $lb,
    ], 
    rowTypes => [
        one => $rt2,
        two => $rt1,
    ], 
    tableTypes => [
        one => $tt1,
        two => $tt2,
    ], 
    import => "writer",
); 

As you can see, the namespaces for the labels, row types and table types are completely independent, and the same names can be reused in each of them for different meaning. All the three sections are optional, so if you want, you can order only the types in the nexus, without any labels.

They can then be extracted from the imported facet as:

$rt1 = $fa->impRowType("one");
$tt1 = $fa->impTableType("one");


Or the whole set of name-value pairs can be obtained with:


@rtset = $fa->impRowTypesHash();
@ttset = $fa->impTableTypesHash();


The exact table types and row types (by themselves or in the table types or labels) in the importing thread will be copied. It's technically possible to share the references to the same row type in the C++ code but it's more efficient to make a separate copy for each thread, and thus the Perl API goes along the more efficient way.

The import is smart in the sense that it preserves the sameness of the row types: if in the exporting thread the same row type was referred from multiple places in the labels, row types and table types sections, in the imported facet that would again be the same row type (even though of course not the one that has been exported but its copy). This again helps with the efficiency when various objects decide if the rows created by this and that type are compatible.

This is all well until you want to export a table type that has an index with a Perl sort condition in it, or an aggregator with the Perl code. The Perl code objects are tricky: they get copied OK when a new thread is created but the attempts to import them through a nexus later causes a terrible memory corruption. So Triceps doesn't allow to export the table types with the function references in it. But it provides an alternative solution: the code snippets can be specified as the source code. It gets compiled when the table type gets initialized. When a table type gets imported through a nexus, it brings the source code with it. The imported table types are always uninitialized, so at initialization time the source code gets compiled in the new thread and works.

It all works transparently: just specify a string instead of a function reference when creating the index, and it will be recognized and processed. For example:

$it= Triceps::IndexType->newPerlSorted("b_c", undef, '
    my $res = ($_[0]->get("b") <=> $_[1]->get("b")
        || $_[0]->get("c") <=> $_[1]->get("c"));
    return $res;
    '
);

Before the code gets compiled, it gets wrapped into a 'sub { ... }', so don't write your own sub in the code string, that would be an error.

There is also the issue of arguments that can be specified for these functions. Triceps is now smart enough to handle the arguments that are one of:

• undef
• integer
• floating-point
• string
• Triceps::RowType object
• Triceps::Row object
• reference to an array or hash thereof

It converts the data to an internal C++ representation in the nexus and then converts it back on import. So, if a TableType has all the code in it in the source form, and the arguments for this code within the limits of this format, it can be exported through the nexus. Otherwise an attempt to export it will fail.

I've modified the SimpleOrderedIndex to use the source code format, and it will pass through the nexuses as well.

The Aggregators have a similar problem, and I'm working on converting them to the source code format too.

A little more about the differences between the code references and the source code format:

When you compile a function, it carries with it the lexical context. So you can make the closures that refer to the "my" variables in their lexical scope. With the source code you can't do this. The table type compiles them at initialization time in the context of the main package, and that's all they can see. Remember also that the global variables are not shared between the threads, so if you refer to a global variable in the code snippet and rely on a value in that variable, it won't be present in the other threads (unless the other threads are direct descendants and the value was set before their creation).

While working with the custom sorted indexes, I've also fixed the way the errors are reported in their Perl handlers. The errors used to be just printed on stderr. Now they propagate properly through the table, and the table operations die with the Per handler's error message. Since an error in the sorting function means that things are going very, very wrong, after that the table becomes inoperative and will die on all the subsequent operations as well.

Multithreaded pipeline, part 4

Let's look at the aggregation by the hour. First, the short version that skips over the actual logic and concentrates on how the nexuses are connected.

sub RawToHourlyMain # (@opts)
{
  my $opts = {};
  Triceps::Opt::parse("traffic main", $opts, {
    @Triceps::Triead::opts,
    from => [ undef, \&Triceps::Opt::ck_mandatory ],
  }, @_);
  my $owner = $opts->{owner};
  my $unit = $owner->unit();

  my $faIn = $owner->importNexus(
    from => $opts->{from},
    as => "input",
    import => "reader",
  );

  # ... create the table and aggregation ...

  my $faOut = $owner->makeNexus(
    name => "data",
    labels => [
      $faIn->getFnReturn()->getLabelHash(),
      hourly => $lbHourlyFiltered,
    ],
    import => "writer",
  );

  # ... connect the input nexus to the table ...
  # ... create the table dump logic ...

  $owner->readyReady();
  $owner->mainLoop(); # all driven by the reader
}

This function inherits the options from Triead::start() as usual and adds the option from of its own. This option's value is then used as the name of nexus to import for reading. The row types of the labels from that imported facet are then used to create the table and aggregation. But they aren't connected to the input labels yet.

First, the output nexus is created. The creation passes through all the incoming data, short-circuiting the input and output, and adds the extra label for the aggregated output. After that the rest of the logic can be connected to the inputs (and to the outputs too).

The reason why this connection order is important is that the labels get caller in the order they are chained from the input label. And when this thread reacts to some event, we want the original event to pass through to the output first and then send the reaction to it.

And after that it's all usual readyReady() and mainLoop().

The full text of the function follows. The logic is based on the previous example from the chapter 13, and the only big change is the use of SimpleAggergator instead of a manually-built one. The HourlyToDailyMain() is very similar, so I won't even show it, you can find the full text in SVN.

# compute an hour-rounded timestamp (in microseconds)
sub hourStamp # (time)
{
  return $_[0]  - ($_[0] % (1000*1000*3600));
}

sub RawToHourlyMain # (@opts)
{
  my $opts = {};
  Triceps::Opt::parse("traffic main", $opts, {
    @Triceps::Triead::opts,
    from => [ undef, \&Triceps::Opt::ck_mandatory ],
  }, @_);
  my $owner = $opts->{owner};
  my $unit = $owner->unit();

  # The current hour stamp that keeps being updated;
  # any aggregated data will be propagated when it is in the
  # current hour (to avoid the propagation of the aggregator clearing).
  my $currentHour;

  my $faIn = $owner->importNexus(
    from => $opts->{from},
    as => "input",
    import => "reader",
  );

  # the full stats for the recent time
  my $ttPackets = Triceps::TableType->new($faIn->getLabel("packet")->getRowType())
    ->addSubIndex("byHour",
      Triceps::IndexType->newPerlSorted("byHour", undef, sub {
        return &hourStamp($_[0]->get("time")) <=> &hourStamp($_[1]->get("time"));
      })
      ->addSubIndex("byIP",
        Triceps::IndexType->newHashed(key => [ "local_ip", "remote_ip" ])
        ->addSubIndex("group",
          Triceps::IndexType->newFifo()
        )
      )
    )
  or confess "$!";

  # type for a periodic summary, used for hourly, daily etc. updates
  my $rtSummary;
  Triceps::SimpleAggregator::make(
    tabType => $ttPackets,
    name => "hourly",
    idxPath => [ "byHour", "byIP", "group" ],
    result => [
      # time period's (here hour's) start timestamp, microseconds
      time => "int64", "last", sub {&hourStamp($_[0]->get("time"));},
      local_ip => "string", "last", sub {$_[0]->get("local_ip");},
      remote_ip => "string", "last", sub {$_[0]->get("remote_ip");},
      # bytes sent in a time period, here an hour
      bytes => "int64", "sum", sub {$_[0]->get("bytes");},
    ],
    saveRowTypeTo => \$rtSummary,
  );

  $ttPackets->initialize() or confess "$!";
  my $tPackets = $unit->makeTable($ttPackets,
    &Triceps::EM_CALL, "tPackets") or confess "$!";

  # Filter the aggregator output to match the current hour.
  my $lbHourlyFiltered = $unit->makeDummyLabel($rtSummary, "hourlyFiltered");
  $tPackets->getAggregatorLabel("hourly")->makeChained("hourlyFilter", undef, sub {
    if ($_[1]->getRow()->get("time") == $currentHour) {
      $unit->call($lbHourlyFiltered->adopt($_[1]));
    }
  });

  # It's important to connect the pass-through data first,
  # before chaining anything to the labels of the faIn, to
  # make sure that any requests and raw inputs get through before
  # our reactions to them.
  my $faOut = $owner->makeNexus(
    name => "data",
    labels => [
      $faIn->getFnReturn()->getLabelHash(),
      hourly => $lbHourlyFiltered,
    ],
    import => "writer",
  );

  my $lbPrint = $faOut->getLabel("print");

  # update the notion of the current hour before the table
  $faIn->getLabel("packet")->makeChained("processPackets", undef, sub {
    my $row = $_[1]->getRow();
    $currentHour = &hourStamp($row->get("time"));
    # skip the timestamp updates without data
    if (defined $row->get("bytes")) {
      $unit->call($tPackets->getInputLabel()->adopt($_[1]));
    }
  });

  # the dump request processing
  $tPackets->getDumpLabel()->makeChained("printDump", undef, sub {
    $unit->makeArrayCall($lbPrint, "OP_INSERT", $_[1]->getRow()->printP() . "\n");
  });
  $faIn->getLabel("dumprq")->makeChained("dump", undef, sub {
    if ($_[1]->getRow()->get("what") eq "packets") {
      $tPackets->dumpAll();
    }
  });

  $owner->readyReady();
  $owner->mainLoop(); # all driven by the reader
}

Multithreaded pipeline, part 3

The rest of this example might be easier to understand by looking at an example of a run first. The lines starting with a "!" are the copies of the input lines that ReaderMain() sends and PrintMain() faithfully prints.

input.packet are the rows that reach the PrintMain on the "print" label (remember, "input" is the name with which it imports its input nexus). input.hourly is the data aggregated by the hour intervals (and also by the IP addresses, dropping the port information), and input.daily further aggregates it per day (and again per the IP addresses). The timestamps in the hourly and daily rows are rounded down to the start of the hour or day.

And the lines without any prefixes are the dumps of the table contents that again reach the PrintMain() through the "print" label:

! new,OP_INSERT,1330886011000000,1.2.3.4,5.6.7.8,2000,80,100
input.packet OP_INSERT time="1330886011000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="100"
input.hourly OP_INSERT time="1330884000000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="100"
input.daily OP_INSERT time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="100"
! new,OP_INSERT,1330886012000000,1.2.3.4,5.6.7.8,2000,80,50
input.packet OP_INSERT time="1330886012000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="50"
input.hourly OP_DELETE time="1330884000000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="100"
input.daily OP_DELETE time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="100"
input.hourly OP_INSERT time="1330884000000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
input.daily OP_INSERT time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
! new,OP_INSERT,1330889612000000,1.2.3.4,5.6.7.8,2000,80,150
input.packet OP_INSERT time="1330889612000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="150"
input.hourly OP_INSERT time="1330887600000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
input.daily OP_DELETE time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
input.daily OP_INSERT time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="300"
! new,OP_INSERT,1330889811000000,1.2.3.4,5.6.7.8,2000,80,300
input.packet OP_INSERT time="1330889811000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="300"
input.hourly OP_DELETE time="1330887600000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
input.daily OP_DELETE time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="300"
input.daily OP_INSERT time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
input.hourly OP_INSERT time="1330887600000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="450"
input.daily OP_DELETE time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
input.daily OP_INSERT time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="600"
! new,OP_INSERT,1330972411000000,1.2.3.5,5.6.7.9,3000,80,200
input.packet OP_INSERT time="1330972411000000" local_ip="1.2.3.5" remote_ip="5.6.7.9" local_port="3000" remote_port="80" bytes="200"
input.hourly OP_INSERT time="1330970400000000" local_ip="1.2.3.5" remote_ip="5.6.7.9" bytes="200"
input.daily OP_INSERT time="1330905600000000" local_ip="1.2.3.5" remote_ip="5.6.7.9" bytes="200"
! new,OP_INSERT,1331058811000000
input.packet OP_INSERT time="1331058811000000"
! new,OP_INSERT,1331145211000000
input.packet OP_INSERT time="1331145211000000"
! dump,packets
time="1330886011000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="100"
time="1330886012000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="50"
time="1330889612000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="150"
time="1330889811000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" local_port="2000" remote_port="80" bytes="300"
time="1330972411000000" local_ip="1.2.3.5" remote_ip="5.6.7.9" local_port="3000" remote_port="80" bytes="200"
! dump,hourly
time="1330884000000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="150"
time="1330887600000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="450"
time="1330970400000000" local_ip="1.2.3.5" remote_ip="5.6.7.9" bytes="200"
! dump,daily
time="1330819200000000" local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="600"
time="1330905600000000" local_ip="1.2.3.5" remote_ip="5.6.7.9" bytes="200"

Note that the order of the lines is completely nice and predictable, nothing goes out of order. Each nexus preserves the order of the rows put into it, and the fact that there is only one writer per nexus and that every thread is fed from only one nexus, avoids the races.

Multithreaded pipeline, part 2

The reader thread drives the pipeline:

sub ReaderMain # (@opts)
{
  my $opts = {};
  Triceps::Opt::parse("traffic main", $opts, {@Triceps::Triead::opts}, @_);
  my $owner = $opts->{owner};
  my $unit = $owner->unit();

  my $rtPacket = Triceps::RowType->new(
    time => "int64", # packet's timestamp, microseconds
    local_ip => "string", # string to make easier to read
    remote_ip => "string", # string to make easier to read
    local_port => "int32",
    remote_port => "int32",
    bytes => "int32", # size of the packet
  ) or confess "$!";

  my $rtPrint = Triceps::RowType->new(
    text => "string", # the text to print (including \n)
  ) or confess "$!";

  my $rtDumprq = Triceps::RowType->new(
    what => "string", # identifies, what to dump
  ) or confess "$!";

  my $faOut = $owner->makeNexus(
    name => "data",
    labels => [
      packet => $rtPacket,
      print => $rtPrint,
      dumprq => $rtDumprq,
    ],
    import => "writer",
  );

  my $lbPacket = $faOut->getLabel("packet");
  my $lbPrint = $faOut->getLabel("print");
  my $lbDumprq = $faOut->getLabel("dumprq");

  $owner->readyReady();

  while(<STDIN>) {
    chomp;
    # print the input line, as a debugging exercise
    $unit->makeArrayCall($lbPrint, "OP_INSERT", "! $_\n");

    my @data = split(/,/); # starts with a command, then string opcode
    my $type = shift @data;
    if ($type eq "new") {
      $unit->makeArrayCall($lbPacket, @data);
    } elsif ($type eq "dump") {
      $unit->makeArrayCall($lbDumprq, "OP_INSERT", $data[0]);
    } else {
      $unit->makeArrayCall($lbPrint, "OP_INSERT", "Unknown command '$type'\n");
    }
    $owner->flushWriters();
  }

  {
    # drain the pipeline before shutting down
    my $ad = Triceps::AutoDrain::makeShared($owner);
    $owner->app()->shutdown();
  }
}

It starts by creating the nexus with the initial set of the labels: for the data about the network packets, for the lines to be printed at the end of the pipeline and for the dump requests to the tables in the other threads. It gets exported for the other threads to import, and also imported right back into this thread, for writing. And then the setup is done, readyReady() is called, and the processing starts.

It reads the CSV lines, splits them, makes a decision if it's a data line or dump request, and one way or the other sends it into the nexus. The data sent to a facet doesn't get immediately forwarded to the nexus. It's collected internally in a tray, and then flushWriters() sends it on. The mainLoop() shown in the last post calls flushWriters() automatically after every tray it processes from the input. But when reading from a file you've got to do it yourself. Of course, it's more efficient to send through multiple rows at once, so a smarter implementation would check if multiple lines are available from the file and send them in larger bundles.

The last part is the shutdown. After the end of file is reached, it's time to shut down the application. You can't just shut down it right away because there still might be data in the pipeline, and if you shut it down, that data will be lost. The right way is to drain the pipeline first, and then do the shutdown when the app is drained. AutoDrain::makeShared() creates a scoped drain: the drain request for all the threads is started when this object is created, and the object construction completes when the drain succeeds. When the object is destroyed, that lifts the drain. So in this case the drain succeeds and then the app gets shut down.

The shutdown causes the mainLoop() calls in all the other threads to return, and the threads to exit. Then startHere() in the first thread has the special logic in it that joins all the started threads after its own main function returns and before it completes. After that the script continues on its way and is free to exit.

Multithreaded pipeline, part 1

The Perl API for the threads is pretty much done now, time for the examples.

The first one re-does a variation of already shown example, the traffic data aggregation from the Chapter 13. The short recap is that it gets the data for each network packet going through and keeps it for a some time, aggregates the data by the hour and keeps it for a longer time, and aggregates it by the day and keeps for a longer time yet. This multi-stage computation naturally matches the pipeline approach.

Since this new example highlights different features than the one in chapter 13, I've changed it logic a little too: it updates both the hourly and daily summaries on every packet received. And I didn't bother to implement the part with the automatic cleaning of the old data, it doesn't add anything to the pipeline works.

The pipeline topologies are quite convenient for working with the threads. The parallel computations create a possibility of things happening in an unpredictable order and producing unpredictable results. The pipeline topology allows the parallelism and at the same time also keeps the data in the same predictable order, with no possibility of rows overtaking each other.

The computation can be split into the following threads:

• read the input, convert and send the data into the model
• store the recent data and aggregate it by the hour
• store the hourly data and aggregate it by the day
• store the daily data
• get the data at the end of the pipeline and print it

Technically, each next stage only needs the data from the previous stage, but to get the updates to the printing stage, they all go all the way through.

Dumping the contents of the tables also requires some special support. Each table is local to its thread and can't be access from the other threads. To dump its contents, the dump request needs to be sent to its thread, which would extract the data and send it through. There are multiple ways to deal with the dump results. One is to have a special label for each table's dump and propagate it to the last stage to print. If all that is needed is text, one label that allows to send strings is good enough, all the dumps can send the data converted to text into it, and it would go all the way through the pipeline. For this example I've picked the last approach.

And now is time to show some code. The main part goes like this:

Triceps::Triead::startHere(
  app => "traffic",
  thread => "print",
  main => \&PrintMain,
);


The startHere() creates an App and starts a Triead in the current OS thread. "traffic" is the app name, "print" the thread name. This thread will be the end of the pipeline, and it will create the rest of the threads. This is a convenient pattern when the results of the model need to be fed back to the current thread, and it works out very conveniently for the unit tests. PrintMain() is the body function of this printing thread:

sub PrintMain # (@opts)
{
  my $opts = {};
  Triceps::Opt::parse("traffic main", $opts, {@Triceps::Triead::opts}, @_);
  my $owner = $opts->{owner};
  my $unit = $owner->unit();

  Triceps::Triead::start(
    app => $opts->{app},
    thread => "read",
    main => \&ReaderMain,
  );
  Triceps::Triead::start(
    app => $opts->{app},
    thread => "raw_hour",
    main => \&RawToHourlyMain,
    from => "read/data",
  );
  Triceps::Triead::start(
    app => $opts->{app},
    thread => "hour_day",
    main => \&HourlyToDailyMain,
    from => "raw_hour/data",
  );
  Triceps::Triead::start(
    app => $opts->{app},
    thread => "day",
    main => \&StoreDailyMain,
    from => "hour_day/data",
  );

  my $faIn = $owner->importNexus(
    from => "day/data",
    as => "input",
    import => "reader",
  );

  $faIn->getLabel("print")->makeChained("print", undef, sub {
    print($_[1]->getRow()->get("text"));
  });
  for my $tag ("packet", "hourly", "daily") {
    makePrintLabel($tag, $faIn->getLabel($tag));
  }

  $owner->readyReady();
  $owner->mainLoop(); # all driven by the reader
}

startHere() accepts a number of fixed options plus arbitrary options that itself doesn't care about but passes to the thread's main function, which are then the responsibility of the main function to parse. To reiterate, the main function gets all the options from the call of startHere(), both these that startHere() parses and these that it simply passes through. startHere() also adds one more option on its own: owner containing the TrieadOwner object that the thread uses to communicate with the rest of the App.

In this case PrintMain() doesn't have any extra options on its own, it's just happy to get startHere()'s standard set that it takes all together from @Triceps::Triead::opts.

It gets the TrieadOwner object $owner from the option appended by startHere(). Each TrieadOwner is created with its own Unit, so the unit is obtained from it to create the thread's model in it. Incidentally, the TrieadOwner  also acts as a clearing trigger object for the Unit, so when the TrieadOwner is destroyed, it properly clears the Unit.

Then it goes and creates all the threads of the pipeline. The start() works very much like startHere(), only it actually creates a new thread and starts the main function in it. The main function can be the same whether it runs through start() or startHere(). The special catch is that the options to start() must contain only the plain Perl values, not Triceps objects. It has to do with how Perl works with threads: it makes a copy of every value for the new thread, and it cant's copy the XS objects, so they simply become undefined in the new thread.

All but the first thread in the pipeline have the extra option from: it tells the input nexus for this thread, and each thread creates an output nexus "data". As mentioned before, the nexus namespaces are per thread that created it, so when the option from says "read/data", it's the nexus "data" created by the thread "read".

So, the pipeline gets all connected sequentially until eventually PrintMain() imports the nexus at its tail. importNexus() returns a facet, which is the thread's API to the nexus. A facet looks very much like an FnReturn for most purposes, with a few additions. It even has a real FnReturn in it, and you work with the labels of that FnReturn. The option as of importNexus gives the name to the facet and to its same-named FnReturn (without it the facet would be named the same as the short name of the nexus, in this case "data"). The option import tells whether this thread will be reading or writing to the nexus, and in this case it's reading.

By the time the pipeline gets to the last stage, it connects a few label:

• print - carries the direct text lines to print in its field "text", and its contents gets printed
• dumprq - carries the dump requests to the tables, so the printing thread doesn't care about it
• packet - carries the raw data about the packets
• hourly - carries the hourly summaries
• daily - carries the daily summaries

The last three get also printed but this time as whole rows.

And after everything is connected, the thread both tells that it's ready and waits for all the other threads to be ready by calling readyReady(). Then its the run time, and mainLoop() takes care of it: until it's told to shutdown, it keeps reading data from the nexus and processes it. The shutdown will be controlled by the file reading thread at the start of the pipeline.The processing is done by getting the rowops from the nexus and calling them on the appropriate label in the facet, which then calls the the labels chained from it, and that gets all the rest of the thread's model running.

file descriptor revocation

In my book "The practice of parallel programming" I went through a few ways to interrupt a thread that is sleeping trying to read a file descriptor. And as it shows, the system call dup2() copying a descriptor of /dev/null over the target file descriptor is the best solution. Only it doesn't always work. I've thought that on Linux it works reliably but now I've found that it works on the sockets but not on the pipes, and even with sockets the accept() seems to ignore it.

Well, I think I've found a better solution: use a dup2() but then also send a signal (SIGUSR1 or SIGUSR2) to the target thread (of course, this requires some handler registered for this signal, at least a dummy one, to avoid killing the processs). Even if dup2() gets ignored by the current system call, the signal will get through and either return EINTR to make the user code retry or cause a system call restart in the OS. In either case the new file descriptor copied by dup2() will be discovered on the next attempt and cause the desired interruption. And unlike the signal used by itself, dup2() closes the race window around the signal.

the Triead lifecycle

Each Triead goes through a few stages in its life:

• declared
• defined
• constructed
• ready
• waited ready
• requested dead
• dead

Note by the way that it's the stages of the Triead object, the OS-level thread as such doesn't know much about them, even though these stages do have some connections to its state.

These stages always go in order and can not be skipped. However for convenience you can move directly to a further stage, this will just automatically pass through all the intermediate stages. Although, well, there is one exception: the "waited ready" and "requested dead" stages can get skipped on the way to "dead". Other than that, there is always the sequence, so if you find out that a Triead is dead, you can be sure that it's also declared, defined, constructed and ready. The attempts to go to a previous stage are silently ignored.

Now, what do these stages mean?

Declared: The App knows the name of the thread and that this thread will eventually exist. When an App is asked to find the resources from this thread (such as Nexuses, and by the way, the nexuses are associated with the threads that created them) it will know to wait until this thread becomes constructed, and then look for the resources. It closes an important race condition: the code that defines the Triead normally runs in a new OS thread but there is no way to tell when exactly will it run and do its work. If you spawned a new thread and then attempted to get a nexus from it before it actually runs, the App would tell you that there is no such thread and fail. To get around it, you declare the thread first and then start it. Most of the time there is no need to declare explicitly, the library code that wraps the thread creation does it for you.

Defined: The Triead object has been created and connected to the App. Since this is normally done from the new OS thread, it also implies that the thread is running and is busy about constructing the nexuses and whatever its own internal resources.

Constructed: The Triead had constructed and exported all the nexuses that it planned to. This means that now these nexuses can be imported by the other threads (i.e. connected to the other threads). After this point the thread can not construct any more nexuses. However it can keep importing the nexuses from the other threads. It's actually a good idea to do all your exports, mark the thread constructed, and only then start importing. This order guarantees the absence of deadlocks (which would be detected and will cause the App to be aborted). There are some special cases when you need to import a nexus from a thread that is not fully constructed yet, and it's possible, but requires more attention and a special override. I'll talk about it in more detail later.

Ready: The thread had imported all the nexuses it wanted and fully initialized all its internals (for example, if it needs to load data from a file, it would do that before telling that it's ready). After this point no more nexuses can be imported. A fine point is that the other threads may still be created, and they may do their exporting and importing, but once a thread is marked as ready, it's cast in bronze. And in the simple cases you don't need to worry about separating the constructed and ready stages, just initialize everything and mark the thread as ready.

Waited ready: Before proceeding further, the thread has to wait for all the threads in App to be ready, or it would lose data when it tries to communicate with them. It's essentially a barrier. Normally both the stages "ready" and "waited ready" are advanced to with a single call readyReady(), the thread says "I'm ready, and let me continue when everyone is ready". After that the actual work can begin. It's still possible to create more threads after that (normally parts of the transient fragments), and until they all become ready, the App may temporarily become unready again, but that's a whole separate advanced topic.

Requested dead: This is a way to request a thread to exit. Normally some control thread will decide that the App needs to exit and will request all its threads to die. The threads will get these requests, perform their last rites and exit. The threads don't have to get this request to exit, that can also always decide to exit on their own. When a thread is requested to die, all the data communication with it stops. No more data will get to it through the nexuses and any data it sends will be discarded. It might churn a little bit through the data in its input buffers but any results produced will be discarded. The good practice is to make sure that all the data is drained before requesting a thread to die. Note that the nexuses created by this thread aren't affected at all, they keep working as usual. It's the data connections between this thread and any nexuses that get broken.

Dead: The thread had completed its execution and exited. Normally you don't need to mark this explicitly. When the thread's main function exits, the library will do it for you. Marking the thread dead also drives the harvesting of the OS threads: the harvesting logic will perform a join() (not to be confused with SQL join) of the thread and thus free the OS resources. The dead Trieads are still visible in the App (except for some special cases with the fragments), and their nexuses continue working as usual (even including the special cases with the fragments), the other threads can keep communicating through them for as long as they want.

The XS comments

The comments are treated in a funny way in the Perl XS files. Well, inside functions they are the normal C/C++ comments. But if you want to put a block comment before a function, XS won't let you just use "//" for its lines. It will say that it's a syntax error.

If you use "#" there, XS doesn't complain. It's a little funny, having two styles of comments in the same file but it kind of works. But only kind of. As it turns out, XS sees these lines simply as the preprocessor lines, and feeds them to the preprocessor. Which in turn happily ignores them. Until one of your lines happens to start with "if" or another preprocessor keywords. Then the preprocessor complains.

The complete solution here is to use "#//". This first makes XS send the line to the preprocessor, which then finds a C-style comment and ignores it.

Triceps Multithreading Concepts

The multithreading support has solidified to the point where I can start documenting it. The full description will have to wait until I finalize the Perl API but the concepts are pretty much settled.

The idea of the multithreading support in Triceps is to make writing the multithreaded model easier. To make writing the good code easy and writing the bad code hard. But of course you don't have to use it, you can always make your own if you wish (just as you could before now).

Without the further ado, the diagram of a multithreaded Triceps application:

Fig. 1. Triceps application.

The Triceps application is embodied in the class App. It's possible to have multiple Apps in one program.

Each thread has multiple parts to it. First, of course, there is the OS-level (or, technically, library-level, or Perl-level) thread where the code executes. And then there is a class that represents this thread and its place in the App. To reduce the naming conflict, this class is creatively named Triead (pronounced still "thread"). In the discussion I use the word "thread" for both concepts, the OS-level thread and the Triead, and it's usually clear from the context which one I mean. But sometimes it's particularly important to make the distinction, and then I name one or the other explicitly.

The class Triead itself is largely opaque, allowing only a few methods for introspection. But there is a control interface to it, called TrieadOwner. The Triead is visible from the outside, the TrieadOwner object is visible only in the OS thread that owns the Triead. The TrieadOwner manages the thread state and acts as the intermediary in the thread's communications with the App.

The data is passed between the threads through the Nexuses. A Nexus is unidirectional, with data going only one way, however it may have multiple writers and multiple readers. All the readers see the exact same data, with rowops going in the exact same order (well, there will be other policies in the future as well, but for now there is only one policy).

A Nexus passes through the data for multiple labels, very much like an FnReturn does (and indeed there is a special connection between them). A Nexus also allows to export the row types and table types from one thread to another.

A Nexus gets connected to the Trieads to though the Facets. A Facet is a connection point between the Nexus and the Triead. Each Facet is for either reading or writing. And there may be only one Facet between a given Nexus and a given Triead, you can't make multiple connections between them. As a consequence, a thread can't both write and read to the same Nexus, it can do only one thing. This might actually be an overly restrictive limitation and might change in the future but that's how things work now.

Each Nexus also has a direction: either direct ("downwards") or reverse ("upwards").  And yes, the reverse Nexuses allow to build the models with loops. However the loops consisting of only the direct Nexuses are not allowed, nor of only reverse Nexuses. They would mess up the flow control. The proper loops must contain a mix of direct and reverse Nexuses.

The direct Nexuses have a limited queue size and stop the writers when the queue fills up, until the data gets consumed, thus providing the flow control. The reverse Nexuses have an unlimited queue size, which allows to avoid the circular deadlocks.

Normally an App is built once and keeps running in this configuration until it stops. But there is a strong need to have the threads dynamically added and deleted too. For example, if the App running as a server and clients connect to it, each client needs to have its thread(s) added on connection and then deleted when the client disconnects. This is handled through the concept of fragments. There is no Fragment class but when you create a Triead, you can specify a fragment name for it. Then it becomes possible to shut down and dispose the threads in a fragment after the fragment's work is done.

initialization templates

Some of the helper initialization templates have been already shown in the examples, and here finally is the systematic description.

Trices has the common approach of the building of the complex objects by the chained mehod calls (shown on a fictional class Object):

Autoref<Object> o = Object::make()->addOption1(arg)->addOption2(arg);

Here Object::make() is a convenience wrapper for "new Object", because new has an inconvenient priority. The new or make() returns a pointer to the newly constructed object, and then each method in the chain returns the same pointer (this, from its standpoint) to facilitate the chaining.

While the chain executes, the pointer stays a simple pointer, not a reference. So the methods in the chain can't throw any exceptions, or the memory will leak. Instead they collect the error messages in an Errors object that has to be checked afterwards, like:

if (o->getErrors()->hasError()) ...

Note that the object gets a reference to it created first, so that on an error it would be properly destroyed.

The convenience template checkOrThrow() allows to do the check along with the chain, and if an error is found, convert it to an exception:

Autoref<Object> o = checkOrThrow(Object::make()->addOption1(arg)->addOption2(arg));

It does all the right things with the references.

Some objects  need to be initialized after all the options have been set, since it's much easier to check things once and get the interaction of the options right rather than check on every option. And since the initialization might create the references to the object, to get it right, it has to be done after the "main" reference is created.

The template initialize() takes care of it:

Autoref<Object> o = initialize(Object::make()->addOption1(arg)->addOption2(arg));

For some objects the initialization can't fail (nor any other errors can be created by the options). For the others, the error needs to be checked afterwards. The template initializeOrThrow() takes care of both the initialization, the check, and exception throwing on errors:

Autoref<Object> o = initializeOrThrow(Object::make()->addOption1(arg)->addOption2(arg));

That's basically it. All these templates are defined in common/Initialize.h.

perl wrapping for the C++ objects

There is one more item in the C++ part of Triceps that I haven't touched upon yet. It's not a part of the C++ API as such but the connection between the C++ and Perl APIs. You need to bother about it if you want to write more of the components in C++ and export them into Perl.

When exporting the C++ (or any compiled language) API into Perl (or into any scripting language) there are two things to consider:

1. The script must never crash the interpterer. The interpreted program might die but the interpreter itself must never crash. If you've ever deal with wksh, you know how horrible is the debugging of such crashes.

2. Perl has its memory management by reference counting, which needs to be married with the memory management at the C++ level.

The solution to the second problem is fairly straightforward: have an intermediate wrapper structure. Perl has its reference counting for the pointer to this structure.  When you construct a Perl object, return the pointer to a newly allocated instance of this structure. When the Perl reference count gows down to zero, if calls the method DESTROY for this object, and then you destroy this structure.

And inside this structure will be the C++ reference to the actual object. When the wrapper structure gets created, it gets a new reference and when the wrapper structure gets destroyed, it releases this reference.

Here is a small example of how the RowType object gets created and destroyed:


WrapRowType *
Triceps::RowType::new(...)
    CODE:
        RowType::FieldVec fld;

        .....

        Onceref<RowType> rt = new CompactRowType(fld);
        Erref err = rt->getErrors();
        if (!err.isNull() && !err->isEmpty()) {
            setErrMsg("Triceps::RowType::new: " + err->print());
            XSRETURN_UNDEF;
        }

        RETVAL = new WrapRowType(rt);
    OUTPUT:
        RETVAL

void
DESTROY(WrapRowType *self)
    CODE:
        delete self;

Now to the first problem. How does Perl know to call a particular XS method for a object? From the package this object is blessed to. If the package happens to be an XS package, the XS method will be called. However it's entirely possible to re-bless the object to a completely different package. If the user does this, a completely wrong method may be called and will crash when it tries to reach a wrong object at the pointer.


Things get even worse for the other arguments of the methods. For example, the argument "other" here:

int
equals(WrapRowType *self, WrapRowType *other)

Well, Perl lets you provide a code snippet that would check that the object is as expected in the typemap file. But how would that snipped know? Going just by the blessed package is unreliable.


Triceps solves this problem by placing an 8-byte magic code at the front of every wrap object. Each class has its own magic value for this field. 8 bytes allows to have great many unique codes, and is quick to check because it's just one CPU word.


This magic code is defined as:



struct WrapMagic {
    char v_[8]; // 8 bytes to make a single 64-bit comparison
    bool operator!=(const WrapMagic &wm) const
    {
        return (*(int64_t *)v_) != (*(int64_t *)wm.v_);
    }
};


Then the wrapper contains the fields:

    WrapMagic magic_;
    Autoref<Class> ref_; // referenced value

    static WrapMagic classMagic_;

The check is done with:

    // returns true if the magic value is bad
    bool badMagic() const
    {
        return magic_ != magic;
    }

And the typemap entry is:

O_WRAP_OBJECT
    if( sv_isobject($arg) && (SvTYPE(SvRV($arg)) == SVt_PVMG) ) {
        $var = ($type)SvIV((SV*)SvRV( $arg ));
        if ($var == 0 || $var->badMagic()) {
            setErrMsg( \"${Package}::$func_name(): $var has an incorrect magic for $ntype\" );
            XSRETURN_UNDEF;
        }
    } else{
        setErrMsg( \"${Package}::$func_name(): $var is not a blessed SV reference to $ntype\" );
        XSRETURN_UNDEF;
    }

It checks that this is an object, of an XS package, and then that the pointer to the C/C++ object is not NULL, and that the value at this pointer starts with the right magic.

setErrMsg() is the Triceps finction for setting its error code, and will soon be replaced with a Perl error indication.

The repeatable build of the wrapper object is handled in Triceps with a template that you don't need to use directly and a macro that you do use. For example,  the definition for the RowType wrap in wrap/Wrap.h is:

DEFINE_WRAP(RowType);

The second component, the magic definition goes into Wrap.cpp:

WrapMagic magicWrapRowType = { "RowType" };

Just make sure that the string contains no more than 7 characters.

Some objects in Triceps are reached through the special references that know both the object and its type (such as rows and row handles). For them there is a separate macro:

DEFINE_WRAP2(const RowType, Rowref, Row);
DEFINE_WRAP2(Table, Rhref, RowHandle);

The arguments are the type class, the reference class and finally the object class itself.

And there is one more twist: sometimes the object are self-contained but when you use them, you must use them only with a correct parent object. Right now there is only one such class: the Tray must be used with its correct Unit, and the Perl code checks it. In this case the wrapper has the reference to both Tray and the Unit, and is defined as:

DEFINE_WRAP_IDENT(Unit, Tray);

a better unit tracer

I've done a few small improvements to the Unit::Tracer in the C++ code.

The first one, I've moved the message buffer support from Unit::StringTracer into the base class Unit::Tracer. The buffer is used in pretty much any tracer, so it looks like a good idea. And if some subclass doesn't want to use it, it doesn't have to.

So, now in Unit::Tracer is a protected element for the use of subclasses:

Erref buffer_;

And a couple of methods for manipulating this buffer:

virtual Erref getBuffer();
virtual void clearBuffer();

They are virtual to let the subclasses do whatever they please, but the default implementation is exactly as it was in the StringTracer: return the buffer_ element, and put a new clean Errors object into the buffer_ reference.

The next improvement allows to add a custom row printer. Now not only the Perl code but also C++ code can print the rows in tracing (though you have to provide that C++ code that prints the interesting portion of the row or the whole row). This separation of the general trace logic and of the row printing will eventually make it to Perl too , but not yet.

There are two ways to do it. First, there is a virtual method

virtual void printRow(string &res, const RowType *rt, const Row *row);

You can re-define it in your subclass and do the printing. The job of this method is to append the information from the row row of the type rt to the result string res. Append, not replace.

The second way is by providing a pointer to a simple C-style function of the type:

typedef void RowPrinter(string &res, const RowType *rt, const Row *row);

The arguments are exactly the same as for the method. This pointer can be given to the Tracer constructor:

Tracer(RowPrinter *rp = NULL);
StringTracer(bool verbose = false, RowPrinter *rp = NULL);
StringNameTracer(bool verbose = false, RowPrinter *rp = NULL);

The default implementation of the method printRow() simply calls the function at this pointer if it's not NULL. So if you redefine this method in your subclass, the row printer function pointer will stop working.

synchronization with events

The events feel like an easier form of synchronization to use than dealing directly with the condition variables, and for the simple cases they are. However once you try to do something real-sized, it always turns out that you want to do something more while holding that mutex, not just set a flag and sleep (or the other way around, not just wake up) as the internals of an event do. And no, holding a separate mutex and waiting for an event with it doesn't work, because the separate mutex doesn't get unlocked when the thread goes to sleep.

And so far in Triceps always, always, every single time an event had become some handcrafted logic around a mutex and condition variable. There is one event in it now but it's about to be split up and converted.

Perhaps a good solution to make the events more general and more usable would be to split the mutex from the rest of the event logic, make it like the mutex-condvar pair that is sometimes called a monitor (or in the Ptwrap library term, a pmcond). Or, to turn the same idea sideways, make public not only the combined API but also the constituent parts, the mutex and the logic that is protected behind the mutex.

It has also turned out that for the auto-reset events the ability to read the event state without resetting it comes quite handy.

more printf for the errors

I've been using strprintf() repeatedly for the error messages and exceptions, and I've come up with a better way for it.

First, Ive added a var-args version of strprintf(), in common/Strprintf.h:

string vstrprintf(const char *fmt, va_list ap);

You can use it to create strings from other functions taking the printf-like arguments.

Next go the extensions to the Errors class. The consistent theme there is "check if the Errors reference (Erref) is NULL, if it is, allocate a new Errors, and then add a formatted error message to it". So I've added the new methods not to Errors but to Erref. They check if the Erref object is NULL, allocate a new Errors object into it if needed, and then format the arguments. The simplest one is:

void f(const char *fmt, ...);

It adds a simple formatted message, always marked as an error. You use it like this:

Erref e; // initially NULL
...
e.f("a message with integer %d", n);

The message may be multi-line, it will be split appropriately, like in Errors::appendMultiline().

The next one is even more smart:

bool fAppend(Autoref<Errors> clde, const char *fmt, ...);

It first checks that the child errors object is not NULL and contains an error, and if it does then it does through the dance of allocating a new Errors object if needed, appends the formatted message, and the child errors. The message goes before the child errors, unlike the method signature. So you can use it blindly like this to do the right thing:

Autoref<Errors> checkSubObject(int idx);
...
for (int i = 0; i < sz; i++)
  e.fAppend(checkSubObject(i),  "error in the sub-object %d:", i);

Same as before, you can use the multi-line error messages.

Next goes the same thing for Exception:

 static Exception f(const char *fmt, ...);
 static Exception fTrace(const char *fmt, ...);

these are the static factory methods that create an Exception object with the message, and either without and with the stack trace. They are used like

throw Exception::f("a message with integer %d", n);

And the similar methods for construction with the nested errors:

static Exception f(Onceref<Errors> err, const char *fmt, ...);
static Exception fTrace(Onceref<Errors> err, const char *fmt, ...);

Unlike the Erref method, these work unconditionally (since their result is normally used in throw, and it's too late to do anything by that time), so you better make sure in advance that there is a child error. A typical usage would be like this:

try {
  ...
} catch (Exception e) {
  throw Exception(e.getErrors(), "error at stage %d:",  n);
}

Again, in the resulting exception the message goes before the nested errors.

atomic int vs mutex

I've been wondering, how much faster is the native implementation of an atomic integer (such as Triceps imports from NSPR) versus a simple mutex-based approach (lock the mutex, change the value, unlock the mutex).

I've finally got interested enough to measure it. On my machine, compiled with GCC optimization -O3, the difference is at about 2.5-3 times. Both with and without contention, the difference is about the same. So it's faster but not hugely so.

An interesting thing is that with contention of two threads for the same mutex, Linux shows the total user CPU time about the same as the real time. With contention of two threads for the same atomic integer, the user CPU time is twice higher than the real time. It looks like Linux manages to account for the wait time properly even in these tiny increments and without adding a huge amount of overhead.

And the real time with contention of two threads for the same mutex (that is, threads doing nothing but locking-unlocking the mutex in a loop) is about 3-3.5 times higher than with the same two threads running sequentially. The same goes for the atomic integers as well.

The program I used was:

#include <stdlib.h>
#include <vector>
#include "pw/ptwrap.h"
#include "mem/Atomic.h"

using namespace Triceps;

const int NTHREADS = 2;

pw::pmutex m;
AtomicInt ai(0);

class TestThread: public pw::pwthread
{
public:
  TestThread():
    p_(NULL)
  { }

  virtual void *execute()
  {
    m.lock();
    m.unlock();
    for (int i = 0; i < 10*1000*1000; i++) {
      if (p_)
        p_();  
      // ai.inc();
      m.lock();
      m.unlock();
    }  
  }

  void (*p_)();
};

std::vector<TestThread *> t;

int main(int argc, char **argv)
{
  m.lock();
  for (int i = 0; i < NTHREADS; i++) {
    TestThread *job = new TestThread;
    t.push_back(job);
    job->start();
  }
  sched_yield();
  sched_yield();
  sched_yield();
  sched_yield();
  sched_yield();
  sched_yield();
  sched_yield();
  sched_yield();
  sched_yield();
  m.unlock(); // the run starts!
  for (int i = 0; i < NTHREADS; i++) {
    t[i]->join();
    delete t[i];
  }
  return 0;
}


Comment, uncomment and adjust the constants as needed. The iteration count of 10 millions may be much too low for the newer faster machines.

The build command for me was:

g++ -DTRICEPS_NSPR=4 -I $TRICEPS/trunk/cpp -O3 -o x x.cpp -lnspr4 -lpthread $TRICEPS/trunk/cpp/build/libtriceps.a

Adjust the library names and TRICEPS_NSPR value as needed (for Ubuntu it would likely be "-DTRICEPS_NSPR=0 -lnspr").To try the simple-minded implementation of atomics through the mutex, remove "-DTRICEPS_NSPR".

To measure the time, run it in the time command:

time ./x

When comparing time, don't forget that when you run 2 threads, they make twice more iterations!

Aggregator in C++, Part 4

And finally here is the reference of actions available in the aggregator handler:

parentIndexType->groupSize(gh)

Get the size of the group. The result is of the type size_t. This is pretty much the only method of IndexType that should be called directly, and only in the aggregation. The rest of the IndexType methods should be accessed through the similar methods in the Table, and I won't even document them. However if you really, really want to, you can find the description of the other methods type/IndexType.h and call them in the aggregation as well.

gadget()->getLabel()->getType()

Get the result row type.

The rest of the actions are done by calling the Index methods on the argument index.Same as with the IndexType, the aggregation is the only place where these methods are called directly, everywhere else the equivalent actions are done through the Table methods. Because of this I've grouped their description with the aggregation and not deparately.

const IndexType *getType() const;

Get the type of this index.

RowHandle *begin() const;


Get the handle of the first row of the group, in the default order according to its first leaf index type. Note that here it's not the whole table's first leaf index type but the first leaf in the index type subtree under this index's type. All the iteration methods return NULL if there are no more rows.

RowHandle *next(const RowHandle *cur) const;

Get the handle of the next row (or NULL if that was the last one) in the default order. The NULL argument makes the NULL result.

RowHandle *last() const;

Get the handle of the last row in the group in the default order.

The rest of the methods of Index aren't really to be used directly.

Unlike the Perl API of AggregatorContext, there are no direct analogs of beginIdx() and such in C++ Index. To get them in C++, you need to translate the iteration to another index type through the Table (and of course, just like in Perl, you would need somehow to get the reference to another index type into your aggregator, and that index type better be in the subtree of the parentIndexType). To translate through the Table, you take any row from the group, usually the first one, and use it with the table methods that accept a sample row.

For example:

RowHandle *sample = index->begin();
RowHandle *rhend =  table->nextGroupIdx(otherIndexType, sample);
for (RowHandle *rhit = table->firstOfGroupIdx(otherIndexType, sample); rhit != rhend; rhit = table->nextIdx(otherIndexType, rhit)) {
  ...
}

This concludes the discussion of the aggregators. This also concludes the description of the whole C++ API, except for the most recent additions.

Aggregator in C++, Part 3

Doing a proper custom aggregator is more involved, and requires making subclasses of both Aggregator and AggregatorType. The test case aggSum shows an example of aggregator that can sum any 64-bit field.

The subclass of Aggregator contains only one method that is very similar to the BasicAggregator handler shown before:

class MySumAggregator: public Aggregator
{
public:
  virtual void handle(Table *table, AggregatorGadget *gadget, Index *index,
      const IndexType *parentIndexType, GroupHandle *gh, Tray *dest,
      Aggregator::AggOp aggop, Rowop::Opcode opcode, RowHandle *rh);
};

void MySumAggregator::handle(Table *table, AggregatorGadget *gadget, Index *index,
    const IndexType *parentIndexType, GroupHandle *gh, Tray *dest,
    Aggregator::AggOp aggop, Rowop::Opcode opcode, RowHandle *rh)
{
  // don't send the NULL record after the group becomes empty
  if (opcode == Rowop::OP_NOP || parentIndexType->groupSize(gh) == 0)
    return;

  int fidx = gadget->typeAs<MySumAggregatorType>()->fieldIdx();

  int64_t sum = 0;
  for (RowHandle *rhi = index->begin(); rhi != NULL; rhi = index->next(rhi)) {
    sum += table->getRowType()->getInt64(rhi->getRow(), fidx, 0);
  }

  // pick the rest of fields from the last row of the group
  RowHandle *lastrh = index->last();

  // build the result row; relies on the aggregator result being of the
  // same type as the rows in the table
  FdataVec fields;
  table->getRowType()->splitInto(lastrh->getRow(), fields);
  fields[fidx].setPtr(true, &sum, sizeof(sum));

  gadget->sendDelayed(dest, fields, opcode);
}

The difference is that the field index is not hardcoded but taken from the aggregator type. The aggregator type is found with

gadget->typeAs<MySumAggregatorType>()

The typeAs template is a very recent addition, I've added it when writing this example. The previous (and still available) equivalent way to do it was to call gadget->getType() and then cast it to the pointer to the proper subclass. The method fieldIdx() is a custom addition to the MySumAggregatorType, not inherited from any base class.

The version of AggregatorGadget::sendDelayed() used here is different:

void sendDelayed(Tray *dest, FdataVec &data, Rowop::Opcode opcode) const;

I've actually added it after writing the first version of this post, then edited the post. It conveniently handles the construction of the row from fields and sending it.

Then the aggregator type needs to be defined with a fixed set of inherited virtual methods plus any needed custom parts.

class MySumAggregatorType: public AggregatorType
{
public:
  // @param fname - field name to sum on
  MySumAggregatorType(const string &name, const string &fname):
    AggregatorType(name, NULL),
    fname_(fname),
    fidx_(-1)
  { }
  // the copy constructor works fine
  // (might set the non-NULL row type, but it will be overwritten
  // during initialization)

  // constructor for deep copy
  // (might set the non-NULL row type, but it will be overwritten
  // during initialization)
  MySumAggregatorType(const MySumAggregatorType &agg, HoldRowTypes *holder):
    AggregatorType(agg, holder),
    fname_(agg.fname_),
    fidx_(agg.fidx_)
  { }

  virtual AggregatorType *copy() const
  {
    return new MySumAggregatorType(*this);
  }

  virtual AggregatorType *deepCopy(HoldRowTypes *holder) const
  {
    return new MySumAggregatorType(*this, holder);
  }

  virtual bool equals(const Type *t) const
  {
    if (this == t)
      return true; // self-comparison, shortcut

    if (!AggregatorType::equals(t))
      return false;

    const MySumAggregatorType *sumt = static_cast<const MySumAggregatorType *>(t);

    if (fname_ != sumt->fname_)
      return false;

    return true;
  }

  virtual bool match(const Type *t) const
  {
    if (this == t)
      return true; // self-comparison, shortcut

    if (!AggregatorType::match(t))
      return false;

    const MySumAggregatorType *sumt = static_cast<const MySumAggregatorType *>(t);

    if (fname_ != sumt->fname_)
      return false;

    return true;
  }

  virtual AggregatorGadget *makeGadget(Table *table, IndexType *intype) const
  {
    return new AggregatorGadget(this, table, intype);
  }

  virtual Aggregator *makeAggregator(Table *table, AggregatorGadget *gadget) const
  {
    return new MySumAggregator;
  }

  virtual void initialize(TableType *tabtype, IndexType *intype)
  {
    const RowType *rt = tabtype->rowType();
    setRowType(rt); // the result has the same type as the argument
    fidx_ = rt->findIdx(fname_);
    if (fidx_ < 0)
      errors_.fAppend(new Errors(rt->print()), "Unknown field '%s' in the row type:", fname_.c_str());
    else {
      if (rt->fields()[fidx_].arsz_ != RowType::Field::AR_SCALAR
      || rt->fields()[fidx_].type_->getTypeId() != Type::TT_INT64)
        errors_.fAppend(new Errors(rt->print()),
          "Field '%s' is not an int64 scalar in the row type:", fname_.c_str());
    }
    AggregatorType::initialize(tabtype, intype);
  }

  // called from the handler
  int fieldIdx() const
  {
    return fidx_;
  }

protected:
  string fname_; // name of the field to sum, must be an int64
  int fidx_; // index of field named fname_
};

The constructor accepts the aggregator name and the name of the field on which it will sum. The field name will be translated to field index during initialization, and made available to the MySumAggregator::handler() via the method fieldIdx(). The aggregator type starts with the result row type of NULL, with the actual row type set during initialization. This ability to set the row type later had the latent existence all along, but it's the first time I've made it explicit. It hasn't propagated to the Perl API yet. The idea here is that the result row type of this aggregator is always equal to the input row type, so rather than specifying the result type explicitly and then having to check it for compatibility, why not just take the table's row type when it becomes available? And it works beautifully.

The copy constructor and the constructor with HoldRowTypes are the implementations of the virtual methods copy() and deepCopy(). The deepCopy() is yet another recent addition used in the multithreaded support. I'll describe it in detail later, for now just Do It Like This.

The methods match() and equals() follow the same general shape as everywhere else. makeGadget() creates a generic gadget, and makeAggregator() creates an instance of aggregator for each group.

The interesting stuff starts happening in the initialization. The row type gets found from the table and set as the result type. Then the aggregation field is found in the row type and checked for being of the proper type. Its index is remembered for the later use.

errors_.fAppend() is a new part of the Erref API that makes the error construction more convenient. It is smart enough to check the reference for NULL and allocate a new Errors if so, then append a printf-formatted message and nested errors to it. I've been adding the direct printf-formatting for both Errors and Exceptions, this part of the API is to be documented yet.

Aggregator in C++, part 2

Building an aggregator is a fairly complex subject, so next I want to show a working example. It turns out that I've never actually written an aggregator in C++ before, other than some tiny fragments; I was always working through Perl. So I've written some more interesting examples just now. The full text can be found in the unit test file cpp/table/t_Agg.cpp.

First, if your aggregator is truly stateless and fully hardcoded, the easier way to do it as by defining a plain function with the same handler arguments and building a BasicAggregatorType with it. And here is one that sums the values of an int64 field (the test case aggBasicSum):

void sumC(Table *table, AggregatorGadget *gadget, Index *index,
        const IndexType *parentIndexType, GroupHandle *gh, Tray *dest,
    Aggregator::AggOp aggop, Rowop::Opcode opcode, RowHandle *rh)
{
  // don't send the NULL record after the group becomes empty
  if (opcode == Rowop::OP_NOP || parentIndexType->groupSize(gh) == 0)
    return;

  int64_t sum = 0;
  for (RowHandle *rhi = index->begin(); rhi != NULL; rhi = index->next(rhi)) {
    sum += table->getRowType()->getInt64(rhi->getRow(), 2, 0); // field c at idx 2
  }

  // pick the rest of fields from the last row of the group
  RowHandle *lastrh = index->last();

  // build the result row; relies on the aggregator result being of the
  // same type as the rows in the table
  FdataVec fields;
  table->getRowType()->splitInto(lastrh->getRow(), fields);
  fields[2].setPtr(true, &sum, sizeof(sum)); // set the field c from the sum

  Rowref res(gadget->getLabel()->getType(), fields);
  gadget->sendDelayed(dest, res, opcode);
}

...
  Autoref<TableType> tt = TableType::make(rt1)
    ->addSubIndex("Hashed", HashedIndexType::make( // will be the default index
        (new NameSet())->add("e")
      )->addSubIndex("Fifo", FifoIndexType::make()
        ->setAggregator(new BasicAggregatorType("aggr", rt1, sumC))
      )
    );
...

As I've told before, you create the BasicAggregatorType by giving it the aggregator name, aggregator result row type, and the handler function.

In this case the handler function is completely hardcoded. It works on the int64 field at index 2. The row type I used in this example is:

row {
  uint8[10] a,
  int32[] b,
  int64 c,
  float64 d,
  string e,
}

So the field is actually named "c", and that's why the aggregator function is named "sumC".  But since in this case everything is known in advance, to make it more efficient, the look-up of the field by name has been skipped, and the field index has been pre-computed and placed into the function.

The general outline of the aggregator is exactly the same as in Perl: check for an empty group, then iterate through all the rows in the group and compute a the sum, fill the rest of the fields from the last row, and send the result. The difference is that there is no AggregatorContext, and the calls are done directly on the bits and pieces received as arguments.

The group size is computed as

parentIndexType->groupSize(gh)

This is pretty much the reason for parentIndexType and gh to be passed as arguments to the handler. Why it's done like this is a long story. It allows an individual index (as in instance, not type!) to not care about the number of rows in it. Remember, a group may have multiple "parallel" indexes defined on it, with all of them containing the same rows, and thus the same number of rows. The group handle (gh) contains the common information about the group, including the group size. And the parent index type is the one that manages the group handles inside it, and knows how to extract the information from them.

The iteration is done as usual with the begin() and next(),  only called directly on the index. It will return NULL after the last row. The last row can also be found directly with index->last(); on an empty group it would return NULL, so be careful with that.


The input row type for reading the rows from the group is found as:


table->getRowType()

The result row is built in this example by copying the last row and replacing one field. The data from the last row is split into FdataVec (the data itself is not copied at this point but the data descriptors in the construction vector are made to point to the data in the original row). Then the descriptor for the field "c" is changed to point to the computed sum. Then the new row is built from the descriptor.

In this particular case the type of the input rows and of the result rows is the same, so either could have been used to construct the result rows.  There are two ways to find the result type:


gadget()->getType()->getRowType()
gadget->getLabel()->getType()


They are exactly the same, just there are two paths leading to the same object.


Finally, the constructed row is sent. sendDelayed() takes care of constructing the rowop from the components.



Potentially, you could even call directly


sendDelayed(dest, gadget->getLabel()->getType()->makeRow(fields), opcode);

 However the caveat is that if the table's enqueuing mode is set to EM_IGNORE (which is an old idea, nowadays everyone should use EM_CALL and eventually there will be no other way than EM_CALL, but for now it could happen), the rows would leak. The reason for that leak is that sendDelayed() will skip the rowop creation with EM_IGNORE, and thus will skip the reference creation, and the row would never have a reference to it created and would never be freed. Creating an explicit reference makes sure that the row will always be cleaned up when it becomes unused.

Aggregator in C++, part 1

I've described the AggregatorType and AggregatorGadget before. Now, the last piece of the puzzle is the Aggregator class.

An object of subclass of AggregatorType defines the logic of the aggregation, and there is one of them per a TableType. An object of subclass of AggregatorGadget provides the aggregation output of a Table, and there is one of them per table. An object of subclass of Aggregator keeps the state of an aggregation group, and there is one of them per Index that holds a group in the table.

Repeating what I wrote before:

"The purpose of the Aggregator object created by AggregatorType's makeAggregator() is to keep the state of the group. If you're doing an additive aggregation, it allows you to keep the previous results. If you're doing the optimization of the deletes, it allows you to keep the previous sent row.

What if your aggregator keeps no state? You still have to make an Aggregator for every group, and no, you can't just return NULL, and no, they are not reference-countable, so you have to make a new copy of it for every group (i.e. for every call of makeAggergator()). This looks decidedly sub-optimal, and eventually I'll get around to straighten it out. The good news though is that most of the real aggerators keep the state anyway, so it doesn't matter much."

Once again, the Aggregator object doesn't normally do  the logic by itself, it calls its AggregatorType for the logic. It only provides storage for one group's state and lets the AggergatorType logic use that storage.

The Aggregator is defined in table/Aggregator.h. It defines two things: the constants and the virtual method.

The constants are in the enum Aggregator::AggOp, same as defined before in Perl:

AO_BEFORE_MOD,
AO_AFTER_DELETE,
AO_AFTER_INSERT,
AO_COLLAPSE,

Their meaning is also the same. They can be converted to and from the string representation with methods

static const char *aggOpString(int code, const char *def = "???");
static int stringAggOp(const char *code);

They work the same way as the other constant conversion methods.

Notably, there is no explicit Aggregator constructor. The base class Aggregator doesn't have any knowledge of the group state, so the default constructor is good enough. You add this knowledge of the state in your subclass, and there you define your own constructor. Whether the constructor takes any arguments or not is completely up to you, since your subclass of AggregatorType will be calling that constructor.

And the actual work then happens in the method

virtual void handle(Table *table, AggregatorGadget *gadget, Index *index,
    const IndexType *parentIndexType, GroupHandle *gh, Tray *dest,
    AggOp aggop, Rowop::Opcode opcode, RowHandle *rh);

Your subclass absolutely has to define this method because it's abstract in the base class. The arguments provide the way to get back to the table, its components and its type, so you don't have to pass them through your Aggregator constructor and keep them in your instance, this saves memory.

Notably absent is the pointer to the AggregatorType. This is because the AggregatorGadget already has this pointer, so all you need to to is call

gadget->getType();

You might also want to cast it to your aggregator's type in case if you plan to call your custom methods, like:

MyAggregatorType *at = (MyAggregatorType *)gadget->getType();

Some of the argument types, Index and GroupHandle, have not been described yet, and will be described shortly.

But in general the method handle() is expected to work very much like the aggregator handler does in Perl. It's called at the same times (and indeed the Perl handler is called from the C++-level of the handler), with the same aggop and opcode. The rh is still the current modified row.

A bit of a difference is that the Perl way hides the sensitive arguments in an AggregatorContext while in C++ they are passed directly.

The job of the handle() is to do whatever is necessary, possibly iterate through the group, possibly use the previous state, produce the new state (and possibly remember it), then send the rowop to the gadget. This last step usually is:

gadget->sendDelayed(dest, newrow, opcode);



Here gadget is the AggregatorGadget from the arguments, opcode also helpfully comes from the arguments (and if it's OP_NOP, there is no need to send anything, it will be thrown away anyway), and newrow is the newly computed result Row (not RowHandle!). Obviously, the type of the result row must match the output row type of the AggregatorGadget, or all kinds of bad things will happen. dest is also an argument of handle(), and is a tray that will hold the aggregator results until a proper time when they can be sent. The "delayed" part of "sendDelayed" means exactly that: the created rowops are not sent directly to the gadget's output but are collected in a Tray until the Table decides that it's the right time to send them.

You absolutely must not use the Gadget's default method send(), use only sendDelayed(). To help with this discipline, AggregatorGadget hides the send() method and makes only the sendDelayed() visible.

little table type additions

As I'm working on the multithreading, this has created some little additions that have been missed before.

First, I've added a method to copy a whole table type. In Perl it is:

$newtt = $oldtt->copy();

In C++ it is:

TableType *copy();

This copies the table type, along with copying all the index types in it, since each table type must have its own instances of the index types. The copied table is always uninitialized.

In case if the table type collected errors, the errors aren't copied, and you should not copy such a table type. This caveat applies to the C++ code, since the Perl code would not let you create a table type with errors, it would fail on these errors immediately.

Second, in the Perl implementation of AggregatorType I've added a method to get back the row type:

$rt = $aggtype->getRowType();