The same way as the FnReturns can be used to get back the direct results of the operations on the tables, can be also used on the templates in general. Indeed, it's a good idea to have a method that would create an FnReturn in all the templates. So I went ahead and added it to the LookupJoin, JoinTwo and Collapse.
For the joins, the resulting FnReturn has one label "out". It's created similarly to the table's:
my $fret = $join->fnReturn();
And then it can be used as usual. The implementation of this method is fairly simple:
sub fnReturn # (self)
{
my $self = shift;
if (!defined $self->{fret}) {
$self->{fret} = Triceps::FnReturn->new(
name => $self->{name} . ".fret",
labels => [
out => $self->{outputLabel},
],
);
}
return $self->{fret};
}
All this kind of makes the method lookup() of LookupJoin redundant, since now pretty much all the same can be done with the streaming function API, and even better, because it provides the opcodes on rowops, can handle the full processing, and calls the rowops one by one without necessarily creating an array. But it could happen yet that the lookup() has some more convenient uses too, so I didn't remove it yet.
For Collapse the interface is a little more complicated: the FnReturn contains a label for each data set, named the same as the data set. The order of labels follows the order of the data set definitions (though right now it's kind of moot, because only one data set is supported). The implementation is:
sub fnReturn # (self)
{
my $self = shift;
if (!defined $self->{fret}) {
my @labels;
for my $n (@{$self->{dsetnames}}) {
push @labels, $n, $self->{datasets}{$n}{lbOut};
}
$self->{fret} = Triceps::FnReturn->new(
name => $self->{name} . ".fret",
labels => \@labels,
);
}
return $self->{fret};
}
It uses the new element $self->{dsetnames} that wasn't present in the code shown before. I've added it now to keep the array of data set names in the order they were defined.
Use these examples to write the fnReturn() in your templates.
Showing posts with label collapse. Show all posts
Showing posts with label collapse. Show all posts
Saturday, November 10, 2012
Saturday, September 29, 2012
More of Collapse with functions
The Collapse as shown before sends all the collected deletes before all the collected inserts. For example, if it has collected the updates for four rows, the output will be (assuming that the Collapse element is named "collapse" and the data set in it is named "idata"):
collapse.idata.out OP_DELETE local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="100"
collapse.idata.out OP_DELETE local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="100"
collapse.idata.out OP_DELETE local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="100"
collapse.idata.out OP_DELETE local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="100"
collapse.idata.out OP_INSERT local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="300"
collapse.idata.out OP_INSERT local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="300"
collapse.idata.out OP_INSERT local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="300"
collapse.idata.out OP_INSERT local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="300"
What if you want the deletes followed directly by the matching inserts? Like this:
collapse.idata.out OP_DELETE local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="100"
collapse.idata.out OP_INSERT local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="300"
collapse.idata.out OP_DELETE local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="100"
collapse.idata.out OP_INSERT local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="300"
collapse.idata.out OP_DELETE local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="100"
collapse.idata.out OP_INSERT local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="300"
collapse.idata.out OP_DELETE local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="100"
collapse.idata.out OP_INSERT local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="300"
With the procedural version it required doing an look-up in the insert table after processing each row in the delete table and handling it if found. So I've left it out to avoid complicating the example. But in the streaming function form it becomes easy, just change the binding a little bit:
my $lbInsInput = $dataset->{tbInsert}->getInputLabel();
my $fbind = Triceps::FnBinding->new(
name => $self->{name} . "." . $dataset->{name} . ".bndTbl",
on => $fret,
unit => $unit,
labels => [
del => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->adopt($_[1]));
# If the INSERT is available after this DELETE, this
# will produce it.
$unit->call($lbInsInput->adopt($_[1]));
}
},
ins => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->makeRowop($OP_INSERT, $_[1]->getRow()));
}
},
],
);
The "del" binding first sends the result out as usual and then forwards the DELETE rowop to the insert table's input. Which then causes the insert rowop to be sent of a match is found. Mind you, the look-up and conditional processing still happens. But now it all happens inside the table machinery, all you need to do is add one more line to invoke it.
Let's talk in a little more detail, what happens when the clearing of the Delete table deletes the row with (local_ip="3.3.3.3" remote_ip="7.7.7.7").
Since when the INSERTs are send after DELETEs, their data is removed from the Insert table too, the following clear() of the Insert table won't find them any more and won't send any duplicates; it will send only the inserts for which there were no matching deletes.
You may notice that the code in the "del" handler only forwards the rows around, and that can be replaced by a chaining:
my $lbDel = $unit->makeDummyLabel(
$dataset->{tbDelete}->getOutputLabel()->getRowType(),
$self->{name} . "." . $dataset->{name} . ".lbDel");
$lbDel->chain($lbOut);
$lbDel->chain($lbInsInput);
my $fbind = Triceps::FnBinding->new(
name => $self->{name} . "." . $dataset->{name} . ".bndTbl",
on => $fret,
unit => $unit,
labels => [
del => $lbDel,
ins => sub {
$unit->call($lbOut->makeRowop($OP_INSERT, $_[1]->getRow()));
},
],
);
This shows another way of label definition in FnBinding: an actual label is created first and then given to FnBinding, instead of letting it automatically create a label from the code. The "if ($_[1]->isDelete())" condition has been removed from the "ins", since it's really redundant, and the delete part with its chaining doesn't do the same check anyway.
This code works just as well and even more efficiently than the previous version, since no Perl code needs to be invoked for "del", it all propagates internally through the chaining. However the price is that the DELETE rowops coming out of the output label will have the head-of-the-chain label in them:
collapse.idata.lbDel OP_DELETE local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="100"
collapse.idata.out OP_INSERT local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="300"
collapse.idata.lbDel OP_DELETE local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="100"
collapse.idata.out OP_INSERT local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="300"
collapse.idata.lbDel OP_DELETE local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="100"
collapse.idata.out OP_INSERT local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="300"
collapse.idata.lbDel OP_DELETE local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="100"
collapse.idata.out OP_INSERT local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="300"
The "ins" side can't be handled just by chaining because it has to replace the opcode in the rowops. A potential different way to handle this would be to define various label types in C++ for many primitive operations, like replacing the opcode, and then build by combining them.
The final item is that the code shown in this post involved a recursive call of the streaming function. Its output from the "del" label got fed back to the function, producing more output on the "ins" label. This worked because it invoked a different code path in the streaming function than the one that produced the "del" data. If it were to form a topological loop back to the same path with the same labels, that would have been an error. The recursion will be discussed in more detail later.
collapse.idata.out OP_DELETE local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="100"
collapse.idata.out OP_DELETE local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="100"
collapse.idata.out OP_DELETE local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="100"
collapse.idata.out OP_DELETE local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="100"
collapse.idata.out OP_INSERT local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="300"
collapse.idata.out OP_INSERT local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="300"
collapse.idata.out OP_INSERT local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="300"
collapse.idata.out OP_INSERT local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="300"
What if you want the deletes followed directly by the matching inserts? Like this:
collapse.idata.out OP_DELETE local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="100"
collapse.idata.out OP_INSERT local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="300"
collapse.idata.out OP_DELETE local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="100"
collapse.idata.out OP_INSERT local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="300"
collapse.idata.out OP_DELETE local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="100"
collapse.idata.out OP_INSERT local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="300"
collapse.idata.out OP_DELETE local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="100"
collapse.idata.out OP_INSERT local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="300"
With the procedural version it required doing an look-up in the insert table after processing each row in the delete table and handling it if found. So I've left it out to avoid complicating the example. But in the streaming function form it becomes easy, just change the binding a little bit:
my $lbInsInput = $dataset->{tbInsert}->getInputLabel();
my $fbind = Triceps::FnBinding->new(
name => $self->{name} . "." . $dataset->{name} . ".bndTbl",
on => $fret,
unit => $unit,
labels => [
del => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->adopt($_[1]));
# If the INSERT is available after this DELETE, this
# will produce it.
$unit->call($lbInsInput->adopt($_[1]));
}
},
ins => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->makeRowop($OP_INSERT, $_[1]->getRow()));
}
},
],
);
The "del" binding first sends the result out as usual and then forwards the DELETE rowop to the insert table's input. Which then causes the insert rowop to be sent of a match is found. Mind you, the look-up and conditional processing still happens. But now it all happens inside the table machinery, all you need to do is add one more line to invoke it.
Let's talk in a little more detail, what happens when the clearing of the Delete table deletes the row with (local_ip="3.3.3.3" remote_ip="7.7.7.7").
- The Delete table sends a rowop with this row and OP_DELETE to its output label collapse.idata.tbDelete.out.
- Which then gets forwarded to a chained label in the FnReturn, collapse.idata.retTbl.del.
- FnReturn has a FnBinding pushed into it, so the rowop passes to the matching label in the binding, collapse.idata.bndTbl.del.
- The Perl handler of that label gets called, first forwards the rowop to the Collapse output label collapse.idata.out, and then to the Insert table's input label collapse.idata.tbInsert.in.
- The Insert table looks up the row by the key, finds it, removes it from the table, and sends an OP_DELETE rowop to its output label collapse.idata.tbInsert.out.
- Which then gets forwarded to a chained label in the FnReturn, collapse.idata.retTbl.ins.
- FnReturn has a FnBinding pushed into it, so the rowop passes to the matching label in the binding, collapse.idata.bndTbl.ins.
- The Perl handler of that label gets called and sends the rowop with the opcode changed to OP_INSERT to the Collapse output label collapse.idata.out.
Since when the INSERTs are send after DELETEs, their data is removed from the Insert table too, the following clear() of the Insert table won't find them any more and won't send any duplicates; it will send only the inserts for which there were no matching deletes.
You may notice that the code in the "del" handler only forwards the rows around, and that can be replaced by a chaining:
my $lbDel = $unit->makeDummyLabel(
$dataset->{tbDelete}->getOutputLabel()->getRowType(),
$self->{name} . "." . $dataset->{name} . ".lbDel");
$lbDel->chain($lbOut);
$lbDel->chain($lbInsInput);
my $fbind = Triceps::FnBinding->new(
name => $self->{name} . "." . $dataset->{name} . ".bndTbl",
on => $fret,
unit => $unit,
labels => [
del => $lbDel,
ins => sub {
$unit->call($lbOut->makeRowop($OP_INSERT, $_[1]->getRow()));
},
],
);
This shows another way of label definition in FnBinding: an actual label is created first and then given to FnBinding, instead of letting it automatically create a label from the code. The "if ($_[1]->isDelete())" condition has been removed from the "ins", since it's really redundant, and the delete part with its chaining doesn't do the same check anyway.
This code works just as well and even more efficiently than the previous version, since no Perl code needs to be invoked for "del", it all propagates internally through the chaining. However the price is that the DELETE rowops coming out of the output label will have the head-of-the-chain label in them:
collapse.idata.lbDel OP_DELETE local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="100"
collapse.idata.out OP_INSERT local_ip="3.3.3.3" remote_ip="7.7.7.7" bytes="300"
collapse.idata.lbDel OP_DELETE local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="100"
collapse.idata.out OP_INSERT local_ip="2.2.2.2" remote_ip="6.6.6.6" bytes="300"
collapse.idata.lbDel OP_DELETE local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="100"
collapse.idata.out OP_INSERT local_ip="4.4.4.4" remote_ip="8.8.8.8" bytes="300"
collapse.idata.lbDel OP_DELETE local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="100"
collapse.idata.out OP_INSERT local_ip="1.1.1.1" remote_ip="5.5.5.5" bytes="300"
The "ins" side can't be handled just by chaining because it has to replace the opcode in the rowops. A potential different way to handle this would be to define various label types in C++ for many primitive operations, like replacing the opcode, and then build by combining them.
The final item is that the code shown in this post involved a recursive call of the streaming function. Its output from the "del" label got fed back to the function, producing more output on the "ins" label. This worked because it invoked a different code path in the streaming function than the one that produced the "del" data. If it were to form a topological loop back to the same path with the same labels, that would have been an error. The recursion will be discussed in more detail later.
Hello streaming functions, or a functional Collapse
Coming up with the good examples of the streaming function usage in Triceps is surprisingly difficult. Ironically, the flexibility of Triceps is the problem. If all you have is SQL, the streaming functions become pretty much a must. But if you can write the procedural code, most things are easier that way. For a streaming function to become beneficial, it has to be written in SQLy primitives (such as tables, joins) and not be easily reducible to the procedural code.
The most distilled example I've come up is in te implementation of Collapse. The original implementation of Collapse is described in the manual section "Collapsed updates". The flush() there goes in a loop deleting the all rows from the state tables and sending them as rowops to the output.
The deletion of all the rows can nowadays be done easier with the Table method clear(). However by itself it doesn't solve the problem of sending the output. It sends the deleted rows to the table's output label but we can't just connect the output of the state tables to the Collapse output: then it would also pick up all the intermediate changes! The data needs to be picked up from the tables output selectively, only in flush().
This makes it a good streaming function: the body of the function consists of running clear() on the state tables, and its result is whatever comes on the output labels of the tables.
Since most of the logic remains unchanged, I've implemented this new version of Collapse as a subclass that extends and replaces some of the code with its own:
package FnCollapse;
our @ISA=qw(Triceps::Collapse);
sub new # ($class, $optName => $optValue, ...)
{
my $class = shift;
my $self = $class->SUPER::new(@_);
# Now add an FnReturn to the output of the dataset's tables.
# One return is enough for both.
# Also create the bindings for sending the data.
foreach my $dataset (values %{$self->{datasets}}) {
my $fret = Triceps::FnReturn->new(
name => $self->{name} . "." . $dataset->{name} . ".retTbl",
labels => [
del => $dataset->{tbDelete}->getOutputLabel(),
ins => $dataset->{tbInsert}->getOutputLabel(),
],
);
$dataset->{fret} = $fret;
# these variables will be compiled into the binding snippets
my $lbOut = $dataset->{lbOut};
my $unit = $self->{unit};
my $OP_INSERT = &Triceps::OP_INSERT;
my $OP_DELETE = &Triceps::OP_DELETE;
my $fbind = Triceps::FnBinding->new(
name => $self->{name} . "." . $dataset->{name} . ".bndTbl",
on => $fret,
unit => $unit,
labels => [
del => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->adopt($_[1]));
}
},
ins => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->makeRowop($OP_INSERT, $_[1]->getRow()));
}
},
],
);
$dataset->{fbind} = $fbind;
}
bless $self, $class;
return $self;
}
# Override the base-class flush with a different implementation.
sub flush # ($self)
{
my $self = shift;
foreach my $dataset (values %{$self->{datasets}}) {
# The binding takes care of producing and directing
# the output. AutoFnBind will unbind when the block ends.
my $ab = Triceps::AutoFnBind->new(
$dataset->{fret} => $dataset->{fbind}
);
$dataset->{tbDelete}->clear();
$dataset->{tbInsert}->clear();
}
}
new() adds the streaming function elements in each data set. They consist of two parts: FnReturn defines the return value of a streaming function (there is no formal definition of the body or the entry point since they are quite flexible), and FnBinding defines a call of the streaming function. In this case the function is called in only one place, so one FnBinding is defined. If called from multiple places, there would be multiple FnBindings.
When a normal procedural function is called, the return address provides the connection to get the result back from it to the caller. In a streaming function, the FnBinding connects the result labels to the caller's further processing of the returned data. Unlike the procedural functions, the data is not returned in one step: run the function, compute the value, return it. Instead the return value of a streaming function is a stream of rowops. As each of them is sent to a return label, it goes through the binding and to the caller's further processing. Then the streaming function continues, producing the next rowop, and so on.
If this sounds complicated, please realize that here we're dealing with the assembly language equivalent for streaming functions. I expect that over time it will become easier.
The second source of complexity is that the arguments of a streaming function are not computed in one step either. You don't normally have a full set of rows to send to a streaming function in one go. Instead you set up the streaming call to bind the result, then you pump the rowops to the function's input, creating them in whatever way.
Getting back to the definition of a streaming function, FnReturn defines a set of labels, each with a logical name. In this case the names are "del" and "ins". The labels inside FnReturn are a special variety of dummy labels, but they are chained to some real labels that send the result of the function. The snippet
del => $dataset->{tbDelete}->getOutputLabel(),
says "create a return label named 'del' and chain it from the tbDelete's output label". There is more details to the naming and label creation but let's not get bogged in it now.
The FnBinding defines a matching set of labels, with the same logical names. It's like a receptacle and plug: you put the plug into the receptacle and get the data flowing, you unplug it and the data flow stops. The Perl version of FnBinding provides a convenience: when it gets a code reference instead of a label, it automatically creates a label with that code for its handler.
In this case both binding labels forward the data to the Collapse's output label. Only the one for the insert table has to change the opcodes to OP_INSERT. The check
if ($_[1]->isDelete()) ...
is really redundant, to be on the safe side, since we know that when the data will be flowing, all of it will be coming from the table clearing and have the opcodes of OP_DELETE.
The actual call happens in flush(): Triceps::AutoFnBind does the "plug into receptable" thing, with automatic unplugging when leaving the block scope. If you want to do things manually, FnReturn has the methods push() and pop() but the scoped binding is safer and easier. Once the binding is done, the data is sent through the function by calling clear() on both tables. And then the block ends, AutoFnBind undoes the binding, and the life goes on.
The result produced by this version of Collapse is exactly the same as by the original version. And even when we get down to grits, it's produced with the exact same logical sequence: the rows are sent out as they are deleted from the state tables. But it's structured differently: instead of the procedural deletion and sending of the rows, the internal machinery of the tables gets invoked, and the results of that machinery are then converted to the form suitable for the collapse results and propagated to the output.
Philosophically, it could be argued: what is the body of this function? Is it just the internal logic of the table delection, that gets triggered by clear() in the caller? Or are the clear() calls also a part of the function body? But it practice it just doesn't matter, whatever.
The most distilled example I've come up is in te implementation of Collapse. The original implementation of Collapse is described in the manual section "Collapsed updates". The flush() there goes in a loop deleting the all rows from the state tables and sending them as rowops to the output.
The deletion of all the rows can nowadays be done easier with the Table method clear(). However by itself it doesn't solve the problem of sending the output. It sends the deleted rows to the table's output label but we can't just connect the output of the state tables to the Collapse output: then it would also pick up all the intermediate changes! The data needs to be picked up from the tables output selectively, only in flush().
This makes it a good streaming function: the body of the function consists of running clear() on the state tables, and its result is whatever comes on the output labels of the tables.
Since most of the logic remains unchanged, I've implemented this new version of Collapse as a subclass that extends and replaces some of the code with its own:
package FnCollapse;
our @ISA=qw(Triceps::Collapse);
sub new # ($class, $optName => $optValue, ...)
{
my $class = shift;
my $self = $class->SUPER::new(@_);
# Now add an FnReturn to the output of the dataset's tables.
# One return is enough for both.
# Also create the bindings for sending the data.
foreach my $dataset (values %{$self->{datasets}}) {
my $fret = Triceps::FnReturn->new(
name => $self->{name} . "." . $dataset->{name} . ".retTbl",
labels => [
del => $dataset->{tbDelete}->getOutputLabel(),
ins => $dataset->{tbInsert}->getOutputLabel(),
],
);
$dataset->{fret} = $fret;
# these variables will be compiled into the binding snippets
my $lbOut = $dataset->{lbOut};
my $unit = $self->{unit};
my $OP_INSERT = &Triceps::OP_INSERT;
my $OP_DELETE = &Triceps::OP_DELETE;
my $fbind = Triceps::FnBinding->new(
name => $self->{name} . "." . $dataset->{name} . ".bndTbl",
on => $fret,
unit => $unit,
labels => [
del => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->adopt($_[1]));
}
},
ins => sub {
if ($_[1]->isDelete()) {
$unit->call($lbOut->makeRowop($OP_INSERT, $_[1]->getRow()));
}
},
],
);
$dataset->{fbind} = $fbind;
}
bless $self, $class;
return $self;
}
# Override the base-class flush with a different implementation.
sub flush # ($self)
{
my $self = shift;
foreach my $dataset (values %{$self->{datasets}}) {
# The binding takes care of producing and directing
# the output. AutoFnBind will unbind when the block ends.
my $ab = Triceps::AutoFnBind->new(
$dataset->{fret} => $dataset->{fbind}
);
$dataset->{tbDelete}->clear();
$dataset->{tbInsert}->clear();
}
}
new() adds the streaming function elements in each data set. They consist of two parts: FnReturn defines the return value of a streaming function (there is no formal definition of the body or the entry point since they are quite flexible), and FnBinding defines a call of the streaming function. In this case the function is called in only one place, so one FnBinding is defined. If called from multiple places, there would be multiple FnBindings.
When a normal procedural function is called, the return address provides the connection to get the result back from it to the caller. In a streaming function, the FnBinding connects the result labels to the caller's further processing of the returned data. Unlike the procedural functions, the data is not returned in one step: run the function, compute the value, return it. Instead the return value of a streaming function is a stream of rowops. As each of them is sent to a return label, it goes through the binding and to the caller's further processing. Then the streaming function continues, producing the next rowop, and so on.
If this sounds complicated, please realize that here we're dealing with the assembly language equivalent for streaming functions. I expect that over time it will become easier.
The second source of complexity is that the arguments of a streaming function are not computed in one step either. You don't normally have a full set of rows to send to a streaming function in one go. Instead you set up the streaming call to bind the result, then you pump the rowops to the function's input, creating them in whatever way.
Getting back to the definition of a streaming function, FnReturn defines a set of labels, each with a logical name. In this case the names are "del" and "ins". The labels inside FnReturn are a special variety of dummy labels, but they are chained to some real labels that send the result of the function. The snippet
del => $dataset->{tbDelete}->getOutputLabel(),
says "create a return label named 'del' and chain it from the tbDelete's output label". There is more details to the naming and label creation but let's not get bogged in it now.
The FnBinding defines a matching set of labels, with the same logical names. It's like a receptacle and plug: you put the plug into the receptacle and get the data flowing, you unplug it and the data flow stops. The Perl version of FnBinding provides a convenience: when it gets a code reference instead of a label, it automatically creates a label with that code for its handler.
In this case both binding labels forward the data to the Collapse's output label. Only the one for the insert table has to change the opcodes to OP_INSERT. The check
if ($_[1]->isDelete()) ...
is really redundant, to be on the safe side, since we know that when the data will be flowing, all of it will be coming from the table clearing and have the opcodes of OP_DELETE.
The actual call happens in flush(): Triceps::AutoFnBind does the "plug into receptable" thing, with automatic unplugging when leaving the block scope. If you want to do things manually, FnReturn has the methods push() and pop() but the scoped binding is safer and easier. Once the binding is done, the data is sent through the function by calling clear() on both tables. And then the block ends, AutoFnBind undoes the binding, and the life goes on.
The result produced by this version of Collapse is exactly the same as by the original version. And even when we get down to grits, it's produced with the exact same logical sequence: the rows are sent out as they are deleted from the state tables. But it's structured differently: instead of the procedural deletion and sending of the rows, the internal machinery of the tables gets invoked, and the results of that machinery are then converted to the form suitable for the collapse results and propagated to the output.
Philosophically, it could be argued: what is the body of this function? Is it just the internal logic of the table delection, that gets triggered by clear() in the caller? Or are the clear() calls also a part of the function body? But it practice it just doesn't matter, whatever.
Thursday, April 12, 2012
An update on Collapse
While cleaning up the joins, I've also added a feature on Collapse: now if you specify the option "fromLabel" in a data set, you don't have to specify the option "unit" any more. You can but you don't have to. By default the unit will be taken from that label.
Saturday, March 31, 2012
The guts of Collapse
The Collapse implementation is fairly small, and is another worthy example for the docs. It's a template, and a "normal" one too: no code generation whatsoever, just a combination of ready components. As with SimpleAggregator, the current Collapse is quite simple and will grow more features over time, so I've copied the original simple version into t/xCollapse.t to stay there unchanged.
The most notable thing about Collapse is that it took just about an hour to write the first version of it and another three or so hours to test it. Which is a lot less than the similar code in the Aleri or Coral8 code base took. The reason for this is that Triceps provides the fairly flexible base data structures that can be combined easily directly in a scripting language. There is no need to redo a lot from scratch every time, just take something and add a little bit on top.
So here it is, with the interspersed comments.
The options parsing goes as usual. The option "data" is parsed again for the options inside it, and those are places into the hash %$dataset.
The dataset options "rowType" and "fromLabel" are both optional but exactly one of them must be present, to be sufficient and non-conflicting. So the code makes sure of it.
If "fromLabel" is used, the row type is found from it. This looks like a pretty good pattern that I plan to spread to the other elements in the future. The unit could also be found from it.
The state is kept in two tables. The reason for them is this: after collapsing, the Collapse may send for each key either a single INSERT rowop, if the row was not there before and became inserted, DELETE rowop if the row was there before and then became deleted, or a DELETE followed by an INSERT if the row was there but then changed its value. Accordingly, this state is kept in two tables: one contains the DELETE part, another the INSERT part for each key, and either part may be empty (or both, if the row at that key has not been changed). After each flush both tables become empty, and then start collecting the modifications again.
The input and output labels get created. The input label has the function with the processing logic set as its handler. The output label is just a dummy. Note that the tables don't get connected anywhere, they are just used as storage, without any immediate reactions to their modifications.
And if the fromLabel is used, the Collapse gets connected to it. After that there is no good reason to keep a separate reference to that label, especially considering that it creates a reference loop and would mess with the memory management. So it gets deleted.
The final blessing is boilerplate. The constructor creates the data structures but doesn't implement any logic. The logic goes next:
The Collapse element knows nothing about the data that went through it before. After each flush it starts again from scratch. It expects that the stream of rows is self-consistent, and makes the conclusions about the previous data based on the new data it sees. An INSERT rowop may mean one of two things: either there was no previous record with this key, or there was a previous record with this key and then it got deleted. The Delete table can be use to differentiate these situations: if there was a row that was then deleted, the Delete table would contain that row. But for the INSERT it doesn't matter: in either case it just inserts the new row into the Insert table.If there was no such row before, it would be the new INSERT. If there was such a row before, it would be an INSERT following a DELETE.
Incidentally, this logic happens to work for the insert-only streams of data too, when the rows get replaced by simply sending another row with the same key. Then if there was a previous row in the Insert table, it would simply get replaced by a new one, and eventually at the flush time the last row would go through.
The DELETE case is more interesting. If we see a DELETE rowop, this means that either there was an INSERT sent before the last flush and now that INSERT becomes undone, or that there was an INSERT after the flush, which also becomes undone. The actions for these cases are different: if the INSERT was before the flush, this row should go into the Delete table, and eventually propagate as a DELETE during the next flush. If the last INSERT was after the flush, then its row would be stored in the Insert table, and now we just need to delete that row and pretend that it never was.
That's what the logic does: first it tries to remove from the Insert table. If succeeded, then it was an INSERT after the flush, that became undone now, and there is nothing more to do. If there was no row to delete, this means that the INSERT must have happened before the last flush, and we need to remember this row in the Delete table and pass it on in the next flush.
Note that this logic is not resistant to an incorrect data sequences. If there ever are two DELETEs for the same key in a row (which should never happen in a correct sequence), the second DELETE will end up in the Delete table.
The flushing is fairly straightforward: first it sends on all the DELETEs, then all the INSERTs, clearing the tables along the way. At first I've though of matching the DELETEs and INSERTs together, sending them next to each other in case if both are available for some key. It's not that difficult to do. But then I've realized that it doesn't matter and just did it the simple way.
The getter functions are fairly simple. The only catch is that the code has to check for exists before it reads the value of $self->{datasets}{$dsetname}{lbOut}. Otherwise, if an incorrect $dsetname is used, the reading would return an undef but along the way would create an unpopulated $self->{datasets}{$dsetname}. Which would then cause a crash when flush() tries to iterate through it and finds the dataset options missing.
That's it, Collapse in a nutshell!
The most notable thing about Collapse is that it took just about an hour to write the first version of it and another three or so hours to test it. Which is a lot less than the similar code in the Aleri or Coral8 code base took. The reason for this is that Triceps provides the fairly flexible base data structures that can be combined easily directly in a scripting language. There is no need to redo a lot from scratch every time, just take something and add a little bit on top.
So here it is, with the interspersed comments.
package Triceps::Collapse;
use Carp;
use strict;
sub new # ($class, $optName => $optValue, ...)
{
my $class = shift;
my $self = {};
&Triceps::Opt::parse($class, $self, {
unit => [ undef, sub { &Triceps::Opt::ck_mandatory(@_); &Triceps::Opt::ck_ref(@_, "Triceps::Unit") } ],
name => [ undef, \&Triceps::Opt::ck_mandatory ],
data => [ undef, sub { &Triceps::Opt::ck_mandatory(@_); &Triceps::Opt::ck_ref(@_, "ARRAY") } ],
}, @_);
# parse the data element
my $dataref = $self->{data};
my $dataset = {};
# dataref->[1] is the best guess for the dataset name, in case if the option "name" goes first
&Triceps::Opt::parse("$class data set (" . $dataref->[1] . ")", $dataset, {
name => [ undef, \&Triceps::Opt::ck_mandatory ],
key => [ undef, sub { &Triceps::Opt::ck_mandatory(@_); &Triceps::Opt::ck_ref(@_, "ARRAY", "") } ],
rowType => [ undef, sub { &Triceps::Opt::ck_ref(@_, "Triceps::RowType"); } ],
fromLabel => [ undef, sub { &Triceps::Opt::ck_ref(@_, "Triceps::Label"); } ],
}, @$dataref);
The options parsing goes as usual. The option "data" is parsed again for the options inside it, and those are places into the hash %$dataset.
# save the dataset for the future
$self->{datasets}{$dataset->{name}} = $dataset;
# check the options
confess "The data set (" . $dataset->{name} . ") must have only one of options rowType or fromLabel"
if (defined $dataset->{rowType} && defined $dataset->{fromLabel});
confess "The data set (" . $dataset->{name} . ") must have exactly one of options rowType or fromLabel"
if (!defined $dataset->{rowType} && !defined $dataset->{fromLabel});
The dataset options "rowType" and "fromLabel" are both optional but exactly one of them must be present, to be sufficient and non-conflicting. So the code makes sure of it.
my $lbFrom = $dataset->{fromLabel};
if (defined $lbFrom) {
confess "The unit of the Collapse and the unit of its data set (" . $dataset->{name} . ") fromLabel must be the same"
unless ($self->{unit}->same($lbFrom->getUnit()));
$dataset->{rowType} = $lbFrom->getType();
}
If "fromLabel" is used, the row type is found from it. This looks like a pretty good pattern that I plan to spread to the other elements in the future. The unit could also be found from it.
# create the tables
$dataset->{tt} = Triceps::TableType->new($dataset->{rowType})
->addSubIndex("primary",
Triceps::IndexType->newHashed(key => $dataset->{key})
);
$dataset->{tt}->initialize()
or confess "Collapse table type creation error for dataset '" . $dataset->{name} . "':\n$! ";
$dataset->{tbInsert} = $self->{unit}->makeTable($dataset->{tt}, "EM_CALL", $self->{name} . "." . $dataset->{name} . ".tbInsert")
or confess "Collapse internal error: insert table creation for dataset '" . $dataset->{name} . "':\n$! ";
$dataset->{tbDelete} = $self->{unit}->makeTable($dataset->{tt}, "EM_CALL", $self->{name} . "." . $dataset->{name} . ".tbInsert")
or confess "Collapse internal error: delete table creation for dataset '" . $dataset->{name} . "':\n$! ";
The state is kept in two tables. The reason for them is this: after collapsing, the Collapse may send for each key either a single INSERT rowop, if the row was not there before and became inserted, DELETE rowop if the row was there before and then became deleted, or a DELETE followed by an INSERT if the row was there but then changed its value. Accordingly, this state is kept in two tables: one contains the DELETE part, another the INSERT part for each key, and either part may be empty (or both, if the row at that key has not been changed). After each flush both tables become empty, and then start collecting the modifications again.
# create the labels
$dataset->{lbIn} = $self->{unit}->makeLabel($dataset->{rowType}, $self->{name} . "." . $dataset->{name} . ".in",
undef, \&_handleInput, $self, $dataset)
or confess "Collapse internal error: input label creation for dataset '" . $dataset->{name} . "':\n$! ";
$dataset->{lbOut} = $self->{unit}->makeDummyLabel($dataset->{rowType}, $self->{name} . "." . $dataset->{name} . ".out")
or confess "Collapse internal error: output label creation for dataset '" . $dataset->{name} . "':\n$! ";
The input and output labels get created. The input label has the function with the processing logic set as its handler. The output label is just a dummy. Note that the tables don't get connected anywhere, they are just used as storage, without any immediate reactions to their modifications.
# chain the input label, if any
if (defined $lbFrom) {
$lbFrom->chain($dataset->{lbIn})
or confess "Collapse internal error: input label chaining for dataset '" . $dataset->{name} . "' to '" . $lbFrom->getName() . "' failed:\n$! ";
delete $dataset->{fromLabel}; # no need to keep the reference any more
}
And if the fromLabel is used, the Collapse gets connected to it. After that there is no good reason to keep a separate reference to that label, especially considering that it creates a reference loop and would mess with the memory management. So it gets deleted.
bless $self, $class; return $self; }
The final blessing is boilerplate. The constructor creates the data structures but doesn't implement any logic. The logic goes next:
sub _handleInput # ($label, $rop, $self, $dataset)
{
my $label = shift;
my $rop = shift;
my $self = shift;
my $dataset = shift;
if ($rop->isInsert()) {
$dataset->{tbInsert}->insert($rop->getRow())
or confess "Collapse " . $self->{name} . " internal error: dataset '" . $dataset->{name} . "' failed an insert-table-insert:\n$! ";
The Collapse element knows nothing about the data that went through it before. After each flush it starts again from scratch. It expects that the stream of rows is self-consistent, and makes the conclusions about the previous data based on the new data it sees. An INSERT rowop may mean one of two things: either there was no previous record with this key, or there was a previous record with this key and then it got deleted. The Delete table can be use to differentiate these situations: if there was a row that was then deleted, the Delete table would contain that row. But for the INSERT it doesn't matter: in either case it just inserts the new row into the Insert table.If there was no such row before, it would be the new INSERT. If there was such a row before, it would be an INSERT following a DELETE.
Incidentally, this logic happens to work for the insert-only streams of data too, when the rows get replaced by simply sending another row with the same key. Then if there was a previous row in the Insert table, it would simply get replaced by a new one, and eventually at the flush time the last row would go through.
} elsif($rop->isDelete()) {
if (! $dataset->{tbInsert}->deleteRow($rop->getRow())) {
confess "Collapse " . $self->{name} . " internal error: dataset '" . $dataset->{name} . "' failed an insert-table-delete:\n$! "
if ($! ne "");
$dataset->{tbDelete}->insert($rop->getRow())
or confess "Collapse " . $self->{name} . " internal error: dataset '" . $dataset->{name} . "' failed a delete-table-insert:\n$! ";
}
}
}
The DELETE case is more interesting. If we see a DELETE rowop, this means that either there was an INSERT sent before the last flush and now that INSERT becomes undone, or that there was an INSERT after the flush, which also becomes undone. The actions for these cases are different: if the INSERT was before the flush, this row should go into the Delete table, and eventually propagate as a DELETE during the next flush. If the last INSERT was after the flush, then its row would be stored in the Insert table, and now we just need to delete that row and pretend that it never was.
That's what the logic does: first it tries to remove from the Insert table. If succeeded, then it was an INSERT after the flush, that became undone now, and there is nothing more to do. If there was no row to delete, this means that the INSERT must have happened before the last flush, and we need to remember this row in the Delete table and pass it on in the next flush.
Note that this logic is not resistant to an incorrect data sequences. If there ever are two DELETEs for the same key in a row (which should never happen in a correct sequence), the second DELETE will end up in the Delete table.
sub flush # ($self)
{
my $self = shift;
my $unit = $self->{unit};
my $OP_INSERT = &Triceps::OP_INSERT;
my $OP_DELETE = &Triceps::OP_DELETE;
foreach my $dataset (values %{$self->{datasets}}) {
my $tbIns = $dataset->{tbInsert};
my $tbDel = $dataset->{tbDelete};
my $lbOut = $dataset->{lbOut};
my $next;
# send the deletes always before the inserts
for (my $rh = $tbDel->begin(); !$rh->isNull(); $rh = $next) {
$next = $rh->next(); # advance the irerator before removing
$tbDel->remove($rh);
$unit->call($lbOut->makeRowop($OP_DELETE, $rh->getRow()));
}
for (my $rh = $tbIns->begin(); !$rh->isNull(); $rh = $next) {
$next = $rh->next(); # advance the irerator before removing
$tbIns->remove($rh);
$unit->call($lbOut->makeRowop($OP_INSERT, $rh->getRow()));
}
}
}
The flushing is fairly straightforward: first it sends on all the DELETEs, then all the INSERTs, clearing the tables along the way. At first I've though of matching the DELETEs and INSERTs together, sending them next to each other in case if both are available for some key. It's not that difficult to do. But then I've realized that it doesn't matter and just did it the simple way.
sub getInputLabel($$) # ($self, $dsetname)
{
my ($self, $dsetname) = @_;
confess "Unknown dataset '$dsetname'"
unless exists $self->{datasets}{$dsetname};
return $self->{datasets}{$dsetname}{lbIn};
}
sub getOutputLabel($$) # ($self, $dsetname)
{
my ($self, $dsetname) = @_;
confess "Unknown dataset '$dsetname'"
unless exists $self->{datasets}{$dsetname};
return $self->{datasets}{$dsetname}{lbOut};
}
sub getDatasets($) # ($self)
{
my $self = shift;
return keys %{$self->{datasets}};
}
The getter functions are fairly simple. The only catch is that the code has to check for exists before it reads the value of $self->{datasets}{$dsetname}{lbOut}. Otherwise, if an incorrect $dsetname is used, the reading would return an undef but along the way would create an unpopulated $self->{datasets}{$dsetname}. Which would then cause a crash when flush() tries to iterate through it and finds the dataset options missing.
That's it, Collapse in a nutshell!
Friday, March 30, 2012
Collapsed updates
Sometimes the exact sequence of how a row at a particular key was updated does not matter, the only interesting part is the end result. Like the OUTPUT EVERY statement in CCL or the pulsed subscription in Aleri. It doesn't have to be time-driven either: if the data comes in as batches, it makes sense to collapse the modifications from the whole batch into one, and send it at the end of the batch.
To do this in Triceps, I've made a template. Here is an example of its use with interspersed comments:
The meaning of the rows is not particularly important for this example. It just uses a pair of the IP addresses as the collapse key. The collapse absolutely needs a primary key, since it has to track and collapse multiple updates to the same row.
Most of the options are self-explanatory. The dataset is defined with nested options to make the API extensible, to allow multiple datasets to be defined in the future. But at the moment only one is allowed. A dataset collapses the data at one label: an input label and an output label get defined for it, just as for the table. The data arrives at the input label, gets collapsed by the primary key, and then stays in the Collapse until the flush. When the Collapse gets flushed, the data is sent out of its output label. After the flush, the Collapse has no data it, and starts collecting the updates again from scratch. The labels gets named by connecting the names of the Collapse element, of the dataset, and "in" or "out". For this Collapse, the label names will be "collapse.idata.in" and "collapse.idata.out".
Note that the dataset options are specified in a referenced array, not a hash! If you try to use a hash, it will fail. When specifying the dataset options, put the "name" first. It's used in the error messages about any issues in the dataset, and the code really expects the name to go first.
To print the result, a print label is created in this example in the same way as in the previous ones. The print label gets connected to the Collapse's output label. The method to get the collapse's output label is very much like table's. Only it gets the dataset name as an argument.
There will be a second example, so I've placed the main look into a function. It works in the same way as in the examples before: extracts the data from the CSV format and sends it to a label. The first column is used as a command: "data" sends the data, and "flush" performs the flush from the Collapse. The flush marks the end of the batch. Here is an example of a run, with the input lines shown as usual in italics:
You can trace and make sure that the flushed data is the cumulative result of the data that went it.
The Collapse also allows to specify the row type and the input connection for a dataset in a different way:
Normally $lbInput would be not a dummy label but the output label of some element. The option "fromLabel" tells that the dataset input will be coming from that label. So the Collapse can automatically both copy its row type for the dataset, and also chain the dataset's input label to that label. It's a pure convenience, allowing to skip the manual steps. In the future it should probably take a whole list of source labels and chain itself to all of them, but for now only one.
This example produces exactly the same output as the previous one, so there is no use in copying it again.
For the last item that hasn't been shown yet, you can get the list of dataset names (well, currently only one name):
And the very last thing to tell about the use of Collapse, when something goes wrong, it will die (and confess). No need to follow its methods with "or die".
To do this in Triceps, I've made a template. Here is an example of its use with interspersed comments:
my $unit = Triceps::Unit->new("unit") or die "$!";
our $rtData = Triceps::RowType->new(
# mostly copied from the traffic aggregation example
local_ip => "string",
remote_ip => "string",
bytes => "int64",
) or die "$!";
The meaning of the rows is not particularly important for this example. It just uses a pair of the IP addresses as the collapse key. The collapse absolutely needs a primary key, since it has to track and collapse multiple updates to the same row.
my $collapse = Triceps::Collapse->new( unit => $unit, name => "collapse", data => [ name => "idata", rowType => $rtData, key => [ "local_ip", "remote_ip" ], ], ) or die "$!";
Most of the options are self-explanatory. The dataset is defined with nested options to make the API extensible, to allow multiple datasets to be defined in the future. But at the moment only one is allowed. A dataset collapses the data at one label: an input label and an output label get defined for it, just as for the table. The data arrives at the input label, gets collapsed by the primary key, and then stays in the Collapse until the flush. When the Collapse gets flushed, the data is sent out of its output label. After the flush, the Collapse has no data it, and starts collecting the updates again from scratch. The labels gets named by connecting the names of the Collapse element, of the dataset, and "in" or "out". For this Collapse, the label names will be "collapse.idata.in" and "collapse.idata.out".
Note that the dataset options are specified in a referenced array, not a hash! If you try to use a hash, it will fail. When specifying the dataset options, put the "name" first. It's used in the error messages about any issues in the dataset, and the code really expects the name to go first.
my $lbPrint = makePrintLabel("print", $collapse->getOutputLabel("idata"));
To print the result, a print label is created in this example in the same way as in the previous ones. The print label gets connected to the Collapse's output label. The method to get the collapse's output label is very much like table's. Only it gets the dataset name as an argument.
sub mainloop($$$) # ($unit, $datalabel, $collapse)
{
my $unit = shift;
my $datalabel = shift;
my $collapse = shift;
while(<STDIN>) {
chomp;
my @data = split(/,/); # starts with a command, then string opcode
my $type = shift @data;
if ($type eq "data") {
my $rowop = $datalabel->makeRowopArray(@data)
or die "$!";
$unit->call($rowop) or die "$!";
$unit->drainFrame(); # just in case, for completeness
} elsif ($type eq "flush") {
$collapse->flush();
}
}
}
&mainloop($unit, $collapse->getInputLabel("idata"), $collapse);
There will be a second example, so I've placed the main look into a function. It works in the same way as in the examples before: extracts the data from the CSV format and sends it to a label. The first column is used as a command: "data" sends the data, and "flush" performs the flush from the Collapse. The flush marks the end of the batch. Here is an example of a run, with the input lines shown as usual in italics:
data,OP_INSERT,1.2.3.4,6.7.8.9,1000 data,OP_DELETE,1.2.3.4,6.7.8.9,1000 flush collapse.idata.out OP_INSERT local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="100" data,OP_DELETE,1.2.3.4,5.6.7.8,100 data,OP_INSERT,1.2.3.4,5.6.7.8,200 data,OP_INSERT,1.2.3.4,6.7.8.9,2000 flush collapse.idata.out OP_DELETE local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="100" collapse.idata.out OP_INSERT local_ip="1.2.3.4" remote_ip="5.6.7.8" bytes="200" collapse.idata.out OP_INSERT local_ip="1.2.3.4" remote_ip="6.7.8.9" bytes="2000" data,OP_DELETE,1.2.3.4,6.7.8.9,2000 data,OP_INSERT,1.2.3.4,6.7.8.9,3000 data,OP_DELETE,1.2.3.4,6.7.8.9,3000 data,OP_INSERT,1.2.3.4,6.7.8.9,4000 data,OP_DELETE,1.2.3.4,6.7.8.9,4000 flush collapse.idata.out OP_DELETE local_ip="1.2.3.4" remote_ip="6.7.8.9" bytes="2000"
You can trace and make sure that the flushed data is the cumulative result of the data that went it.
The Collapse also allows to specify the row type and the input connection for a dataset in a different way:
my $lbInput = $unit->makeDummyLabel($rtData, "lbInput"); my $collapse = Triceps::Collapse->new( unit => $unit, name => "collapse", data => [ name => "idata", fromLabel => $lbInput, key => [ "local_ip", "remote_ip" ], ], ) or die "$!"; &mainloop($unit, $lbInput, $collapse);
Normally $lbInput would be not a dummy label but the output label of some element. The option "fromLabel" tells that the dataset input will be coming from that label. So the Collapse can automatically both copy its row type for the dataset, and also chain the dataset's input label to that label. It's a pure convenience, allowing to skip the manual steps. In the future it should probably take a whole list of source labels and chain itself to all of them, but for now only one.
This example produces exactly the same output as the previous one, so there is no use in copying it again.
For the last item that hasn't been shown yet, you can get the list of dataset names (well, currently only one name):
@names = $collapse->getDatasets();
And the very last thing to tell about the use of Collapse, when something goes wrong, it will die (and confess). No need to follow its methods with "or die".
Subscribe to:
Posts (Atom)