odds and ends

While working on threads support, I've added a few small features here and there. Some of them have been already described, some will be described now. I've also done a few more small clean-ups.

First, the historic methods setName() are now gone everywhere. This means Unit and Label classes, and in C++ also the Gadget. The names can now only be specified during the object construction.

FnReturn has the new method:

$res = fret->isFaceted();

bool isFaceted() const;

It returns true (or 1 in Perl) if this FnReturn object is a part of a Facet.

Unit has gained a couple of methods:

$res = $unit->isFrameEmpty();

bool isFrameEmpty() const;

Check whether the current frame is empty. This is different from the method empty() that checks whether the the whole unit is empty. This method is useful if you run multiple units in the same thread, with some potentially complicated cross-unit scheduling. It's what nextXtray() does with a multi-unit Triead, repeatedly calling drainFrame() for all the units that are found not empty. In this situation the simple empty() can not be used because the current inner frame might not be the outer frame, and draining the inner frame can be repeated forever while the outer frame will still contain rowops. The more precise check of isFrameEmpty() prevents the possibility of such endless loops.

$res = $unit->isInOuterFrame();

bool isInOuterFrame() const;

Check whether the unit's current inner frame is the same as its outer frame, which means that the unit is not in the middle of a call.


In Perl the method Rowop::printP() has gained an optional argument for the printed label name:

$text = $rop->printP();
$text = $rop->printP($lbname);

The reason for that is to make the printing of rowops in the chained labels more convenient. A chained label's execution handler receives the original unchanged rowop that refers to the first label in the chain. So when it gets printed, it will print the name of the first label in the chain, which might be very surprising. The explicit argument allows to override it to the name of the chained label (or to any other value).


 In C++ the Autoref has gained the method swap():

void swap(Autoref &other);

It swaps the values of two references without changing the reference counts in the referred values. This is a minor optimization for such a special situation. One or both references may contain NULL.


In C++ the Table has gained the support for sticky errors. The table internals contain a few places where the errors can't just throw an Exception because it will mess up the logic big time, most specifically the comparator functions for the indexes. The Triceps built-in indexes can't encounter any errors in the comparators but the user-defined ones, such as the Perl Sorted Index, can. Previously there was no way to report these errors other than print the error message and then either continue pretending that nothing happened or abort the program.

The sticky errors provide a way out of this sticky situation. When an index comparator encounters an error, it reports it as a sticky error in the table and then returns false. The table logic then unrolls like nothing happened for a while, but before returning from the user-initiated method it will find this sticky error and throw an Exception at a safe time. Obviously, the incorrect comparison means that the table enters some messed-up state, so all the further operations on the table will keep finding this sticky error and throw an Exception right away, before doing anything. The sticky error can't be unstuck. The only way out of it is to just discard the table and move on.

void setStickyError(Erref err);

Set the sticky error from a location where an exception can not be thrown, such as from the comparators in the indexes. Only the first error sticks, all the others are ignored since (a) the table will be dead and throwing this error in exceptions from this point on anyway and (b) the comparator is likely to report the same error repeatedly and there is no point in seeing multiple copies.

Errors *getStickyError() const;

Get the table's sticky error. Normally there is no point in doing this manually, but just in case.

void checkStickyError() const;

If the sticky error has been set, throw an Exception with it.

No more enqueueing mode for table creation

I've finally got around to get rid of that obsolete enqueuing mode argument for the table creation, which always ended up as EM_CALL nowadays anyway. So, now in Perl the call becomes:

$uint->makeTable($tabType, $name);

In C++ the Table constructor becomes:

Table(Unit *unit, const string &name, const TableType *tt, const RowType *rowt, const RowHandleType *handt);

And the convenience wrapper in the TableType:

Onceref<Table> makeTable(Unit *unit, const string &name) const;

Yeah, it's kind of weird that in Perl the method makeTable() is defined on Unit, and in C++ on TableType. But if I remember correctly, it has to do with avoiding the circular dependency in the C++ header files.

how to export a table, or the guts of TQL join exposed

Now to the point of why the multithreaded TQl example got written: the export of a table between two threads.

It all starts in the Tql initialization method. In the multithreaded mode it builds the nexuses for communication. I'll skip the input nexus and show the building of only the output and request-dump nexuses:

    # row type for dump requests and responses
    my $rtRequest = Triceps::RowType->new(
      client => "string", #requesting client
      id => "string", # request id
      name => "string", # the table name, for convenience of requestor
      cmd => "string", # for convenience of requestor, the command that it is executing
    ) or confess "$!";

The request row type is used by the client writer thread to request the table dumps from the core logic, and to get back the notifications about the dumps.

    # build the output side
    for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
      my $name = $self->{tableNames}[$i];
      my $table = $self->{tables}[$i];

      push @tabtypes, $name, $table->getType()->copyFundamental();
      push @labels, "t.out." . $name, $table->getOutputLabel();
      push @labels, "t.dump." . $name, $table->getDumpLabel();
    }
    push @labels, "control", $rtControl; # pass-through from in to out
    push @labels, "beginDump", $rtRequest; # framing for the table dumps
    push @labels, "endDump", $rtRequest;

    $self->{faOut} = $owner->makeNexus(
      name => $self->{nxprefix} . "out",
      labels => [ @labels ],
      tableTypes => [ @tabtypes ],
      import => "writer",
    );
    $self->{beginDump} = $self->{faOut}->getLabel("beginDump");
    $self->{endDump} = $self->{faOut}->getLabel("endDump");

On the output side each table is represented by 3 elements:

• its fundamental table type (stripped down to the primary key);
• its output label for normal updates;
• its dump label for the responses to the dump requests.

There also are the "beginDump" and "endDump" labels that frame each response to a dump request.

The row type $rtControl and label "control" is used to pass the commands from the client reader to client writer, but it's exact contents is not important here.

The dump request nexus is built in a similar way:

    # build the dump requests, will be coming from below
    undef @labels;
    for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
      my $name = $self->{tableNames}[$i];
      my $table = $self->{tables}[$i];

      push @labels, "t.rqdump." . $name, $rtRequest;
    }
    $self->{faRqDump} = $owner->makeNexus(
      name => $self->{nxprefix} . "rqdump",
      labels => [ @labels ],
      reverse => 1, # avoids making a loop, and gives priority
      import => "reader",
    );
    # tie together the labels
    for (my $i = 0; $i <= $#{$self->{tables}}; $i++) {
      my $name = $self->{tableNames}[$i];
      my $table = $self->{tables}[$i];

      $self->{faRqDump}->getLabel("t.rqdump." . $name)->makeChained(
        $self->{nxprefix} . "rqdump." . $name, undef,
        \&_dumpTable, $self, $table
      );
    }

The dumps are executed in the function _dumpTable:

sub _dumpTable # ($label, $rowop, $self, $table)
{
  my ($label, $rop, $self, $table) = @_;
  my $unit = $label->getUnit();
  # pass through the client id to the dump
  $unit->call($self->{beginDump}->adopt($rop));
  $table->dumpAll();
  $unit->call($self->{endDump}->adopt($rop));
  $self->{faOut}->flushWriter();
}

The data gets framed around by the "beginDump" and "endDump" labels getting the copies of the original request. This helps the client writer thread keep track of its current spot. The flushing of the writer is not strictly needed. Just in case if multiple dump requests are received in a single tray, it breaks up the responses into a separate tray for each dump, keeping the size of the trays lower. Not that this situation could actually happen yet.

This part taken care of, let's jump around and see how the client writer thread processes a "querysub" request:

      } elsif ($cmd eq "querysub") {
        if ($id eq "" || exists $queries{$id}) {
          printOrShut($app, $fragment, $sock,
            "error,$id,Duplicate id '$id': query ids must be unique,bad_id,$id\n");
          next;
        }
        my $ctx = compileQuery(
          qid => $id,
          qname => $args[0],
          text => $args[1],
          subError => sub {
            chomp $_[2];
            $_[2] =~ s/\n/\\n/g; # no real newlines in the output
            $_[2] =~ s/,/;/g; # no confusing commas in the output
            printOrShut($app, $fragment, $sock, "error,", join(',', @_), "\n");
          },
          faOut => $faOut,
          faRqDump => $faRqDump,
          subPrint => sub {
            printOrShut($app, $fragment, $sock, @_);
          },
        );
        if ($ctx) { # otherwise the error is already reported
          $queries{$id} = $ctx;
          &$runNextRequest($ctx);
        }
      }

The query id is used to keep track of the outstanding queries, so the code makes sure that it's unique, and you can see an example of the query response. The bulk of the work is done in the method compileQuery(). The arguments to it give the details of the query and also provide the closures for the functionality that differs between the single-threaded and multi-threaded versions. The option "subError" is used to send the errors to the client, and "subPrint" is used to send the output to the client, it gets used for building the labels in the "print" command of the query.

compileQuery() returns the query context, which contains a compiled sub-model that executes the query and a set of requests that tell the writer how to connect the query to the incoming data. Or on error it reports the error using subError and returns an undef. If the compilation succeeded, the writer remembers the query and starts the asynchronous execution of the requests. More about the requests later, now let's look at the query compilation and context.

The context is created in compileQuery() thusly:

  my $ctx = {};
  $ctx->{qid} = $opts->{qid};
  $ctx->{qname} = $opts->{qname};

  # .. skipped the parts related to single-threadde TQL

  $ctx->{faOut} = $opts->{faOut};
  $ctx->{faRqDump} = $opts->{faRqDump};
  $ctx->{subPrint} = $opts->{subPrint};
  $ctx->{requests} = []; # dump and subscribe requests that will run the pipeline
  $ctx->{copyTables} = []; # the tables created in this query
    # (have to keep references to the tables or they will disappear)

  # The query will be built in a separate unit
  $ctx->{u} = Triceps::Unit->new($opts->{nxprefix} . "${q}.unit");
  $ctx->{prev} = undef; # will contain the output of the previous command in the pipeline
  $ctx->{id} = 0; # a unique id for auto-generated objects
  # deletion of the context will cause the unit in it to clean
  $ctx->{cleaner} = $ctx->{u}->makeClearingTrigger();

It has some parts common and some parts differing for the single- and multi-threaded varieties, here I've skipped over the single-threaded parts.

One element that is left undefined here is $ctx->{prev}. It's the label created as the output of the previous stage of the query pipeline. As each command in the pipeline builds its piece of processing, it chains its logic from $ctx->{prev} and leaves its result label in $ctx->{next}. Then compileQuery() moves "next" to "prev" and calls the compilation of the next command in the pipeline. The only command that accepts an undefined "prev" (and it must be undefined for it) is "read", that reads the table at the start of the pipeline.

$ctx->{copyTables} also has an important point behind it. When you create a label, it's OK to discard the original reference after you chain the label into the logic, that chaining will keep a reference and the label will stay alive. Not so with a table: if you create a table, chain its input label and then drop the reference to a table, the table will be discarded. Then when the input label will try to send any data to the table, it will die (and unless very recently it outright crashed). So it's important to keep the table reference alive, and that's what this array is for.

$ctx->{id} is used to generate the unique names for the objects build in a query.

Each query is built in its own unit. This is convenient, after the query is done or the compilation encounters an error, the unit with its whole contents gets easily discarded. The clearing trigger placed in the context makes sure that the unit gets properly cleared and discarded.

Next goes the compilation of the join query command, I'll go through it in chunks.

sub _tqlJoin # ($ctx, @args)
{
  my $ctx = shift;
  die "The join command may not be used at the start of a pipeline.\n"
    unless (defined($ctx->{prev}));
  my $opts = {};
  &Triceps::Opt::parse("join", $opts, {
    table => [ undef, \&Triceps::Opt::ck_mandatory ],
    rightIdxPath => [ undef, undef ],
    by => [ undef, undef ],
    byLeft => [ undef, undef ],
    leftFields => [ undef, undef ],
    rightFields => [ undef, undef ],
    type => [ "inner", undef ],
  }, @_);

  my $tabname = bunquote($opts->{table});
  my $unit = $ctx->{u};
  my $table;

  &Triceps::Opt::checkMutuallyExclusive("join", 1, "by", $opts->{by}, "byLeft", $opts->{byLeft});
  my $by = split_braced_final($opts->{by});
  my $byLeft = split_braced_final($opts->{byLeft});

  my $rightIdxPath;
  if (defined $opts->{rightIdxPath}) { # propagate the undef
    $rightIdxPath = split_braced_final($opts->{rightIdxPath});
  }

It starts by parsing the options and converting them to the internal representation, removing the braced quotes.

  if ($ctx->{faOut}) {
    # Potentially, the tables might be reused between multiple joins
    # in the query if the required keys match. But for now keep things
    # simpler by creating a new table from scratch each time.

    my $tt = eval {
      # copy to avoid adding an index to the original type
      $ctx->{faOut}->impTableType($tabname)->copy();
    };
    die ("Join found no such table '$tabname'\n") unless ($tt);

    if (!defined $rightIdxPath) {
      # determine or add the index automatically
      my @workby;
      if (defined $byLeft) { # need to translate
        my @leftfld = $ctx->{prev}->getRowType()->getFieldNames();
        @workby = &Triceps::Fields::filterToPairs("Join option 'byLeft'",
          \@leftfld, [ @$byLeft, "!.*" ]);
      } else {
        @workby = @$by;
      }

      my @idxkeys; # extract the keys for the right side table
      for (my $i = 1; $i <= $#workby; $i+= 2) {
        push @idxkeys, $workby[$i];
      }
      $rightIdxPath = [ $tt->findOrAddIndex(@idxkeys) ];
    }

    # build the table from the type
    $tt->initialize() or confess "$!";
    $table = $ctx->{u}->makeTable($tt, "EM_CALL", "tab" . $ctx->{id} . $tabname);
    push @{$ctx->{copyTables}}, $table;

    # build the request that fills the table with data and then
    # keeps it up to date;
    # the table has to be filled before the query's main flow starts,
    # so put the request at the front
    &_makeQdumpsub($ctx, $tabname, 1, $table->getInputLabel());
  } else {
    die ("Join found no such table '$tabname'\n")
      unless (exists $ctx->{tables}{$tabname});
    $table = $ctx->{tables}{$tabname};
  }

The presence of $ctx->{faOut} means that the query is compiled in the multithreaded context.

The command handles may freely die, and the error messages will be caught by compileQuery() and nicely (at least, sort-of) reported back to the user.

If an explicit rightIdxPath was not requested, it gets found or added automatically. On the way there the index fields need to be determined. Which can be specified as either explicit pairs in the option "by" or the in the name translation syntax in the option "byLeft". If we've got a "byLeft", first it gets translated to the same format as "by", and then the right-side fields are extracted from the format of "by". After that $tt->findOrAddIndex() takes care of all the heavy lifting. It either finds a matching index type in the table type or creates a new one from the specified fields, and either way returns the index path. (An invalid field will make it confess).

It looks a bit anti-climactic, but the three lines of exporting with copyFundamental(), impTableType() and findOrAddIndex() is what this large example is all about.

You might wonder, how come the explicit rightIdxPath is not checked in any way? It will be checked later by LookupJoin(), so not much point in doing the check twice.

After that the table is created in a straightforward way, and rememebered in copyTables. And the requests list gets prepended with a request to dump and subscribe to this table. I'll get back to that, for now let's finish up with _tqlJoin().

  my $isLeft = 0; # default for inner join
  my $type = $opts->{type};
  if ($type eq "inner") {
    # already default
  } elsif ($type eq "left") {
    $isLeft = 1;
  } else {
    die "Unsupported value '$type' of option 'type'.\n"
  }

  my $leftFields = split_braced_final($opts->{leftFields});
  my $rightFields = split_braced_final($opts->{rightFields});

  my $join = Triceps::LookupJoin->new(
    name => "join" . $ctx->{id},
    unit => $unit,
    leftFromLabel => $ctx->{prev},
    rightTable => $table,
    rightIdxPath => $rightIdxPath,
    leftFields => $leftFields,
    rightFields => $rightFields,
    by => $by,
    byLeft => $byLeft,
    isLeft => $isLeft,
    fieldsDropRightKey => 1,
  );

  $ctx->{next} = $join->getOutputLabel();
}

The rest of the options get parsed, and then all the collected data gets forwarded to the LookupJoin constructor. Finally the "next" label is assigned from the join's result.

Now jumping to the _makeQdumpsub(). It's used by both the "read" and "join" query commands to initiate the joins and subscriptions.

sub _makeQdumpsub # ($ctx, $tabname, [$front, $lbNext])
{
  my $ctx = shift;
  my $tabname = shift;
  my $front = shift;
  my $lbNext = shift;

  my $unit = $ctx->{u};

  my $lbrq = eval {
    $ctx->{faRqDump}->getLabel("t.rqdump.$tabname");
  };
  my $lbsrc = eval {
    $ctx->{faOut}->getLabel("t.out.$tabname");
  };
  die ("Found no such table '$tabname'\n") unless ($lbrq && $lbsrc);

  # compute the binding for the data dumps, that would be a cross-unit
  # binding to the original faOut but it's OK
  my $fretOut = $ctx->{faOut}->getFnReturn();
  my $dumpname = "t.dump.$tabname";
  # the dump and following subscription data will merge on this label
  if (!defined $lbNext) {
    $lbNext = $unit->makeDummyLabel(
      $lbsrc->getRowType(), "lb" . $ctx->{id} . "out_$tabname");
  }

  my $bindDump = Triceps::FnBinding->new(
    on => $fretOut,
    name => "bind" . $ctx->{id} . "dump",
    labels => [ $dumpname => $lbNext ],
  );

First it finds all the proper labels. The label $lbNext will accept the merged dump contents and the following subscription, and it might be either auto-generated or received as an argument. A join pass it as an argument, $table->getInputLabel(), so all the data goes to the copied table.

The binding is used to receive the dump. It's a bit of an optimization. Remember, the dump labels are shared between all the clients. Whenever any client requests a dump, all the clients will get the response. A client finds that the incoming dump is destined for it by processing the "beginDump" label. If it contains this client's name, the dump is destined here, and the client reacts by pushing the appropriate binding onto the facet's FnReturn, and the data flows. The matching "endDump" label then pops the binding and the data stops flowing. The binding allows to avoid checking every rowop for whethere it's supposed to be accepted and if yes then where exactly (rememeber, the same table may be dumped independently multiple times by multiple queries). Just check once at the start of the bundle and then let the data flow in bulk.

  # qdumpsub:
  #   * label where to send the dump request to
  #   * source output label, from which a subscription will be set up
  #     at the end of the dump
  #   * target label in the query that will be tied to the source label
  #   * binding to be used during the dump, which also directs the data
  #     to the same target label
  my $request = [ "qdumpsub", $lbrq, $lbsrc, $lbNext, $bindDump ];
  if ($front) {
    unshift @{$ctx->{requests}}, $request;
  } else {
    push @{$ctx->{requests}}, $request;
  }
  return $lbNext;
}

Finally, the created bits and pieces get packaged into a request and added to the list of requests in the query context. The last tricky part is that the request can be added at the back or the front of the list. The "normal" way is to add to the back, however the dimension tables for the joins have to be populated before the main data flow of the query starts. So for them the argument $front is set to 1, and they get added in the front.

Now jumping back to the writer thread logic, after it called compileQuery, it starts the query execution by calling &$runNextRequest(). Which is a closure function defined inside the client writer function, and knows how to process the "qdumpsub"s we've just seen created.

  my $runNextRequest = sub { # ($ctx)
    my $ctx = shift;
    my $requests = $ctx->{requests};
    undef $ctx->{curRequest}; # clear the info of the previous request
    my $r = shift @$requests;
    if (!defined $r) {
      # all done, now just need to pump the data through
      printOrShut($app, $fragment, $sock,
        "querysub,$ctx->{qid},$ctx->{qname}\n");
      return;
    }

First it clears the information about the previous request, if any. This function will be called after each request, to send on the next one, so on all its calls except the first one for a query it will have something to clear.

Then it checks if all the requests are already done. If so, it sends the query confirmation to the client and returns. The subscription part of the query will continue running on its own.

    $ctx->{curRequest} = $r; # remember until completed
    my $cmd = $$r[0];
    if ($cmd eq "qdumpsub") {
      # qdumpsub:
      #   * label where to send the dump request to
      #   * source output label, from which a subscription will be set up
      #     at the end of the dump
      #   * target label in the query that will be tied to the source label
      #   * binding to be used during the dump, which also directs the data
      #     to the same target label
      my $lbrq = $$r[1];
      $unit->makeHashCall($lbrq, "OP_INSERT",
        client => $fragment, id => $ctx->{qid}, name => $ctx->{qname}, cmd => $cmd);

The "qdumpsub" gets forwarded to the core logic. The responses will be processed in the handlers or "beginDump" and "endDump". One of the great pains of this "actor" architecture is that the linear logic gets broken up into many disjointed pieces in the separate handlers.

    } else {
      printOrShut($app, $fragment, $sock,
        "error,", $ctx->{qid}, ",Internal error: unknown request '$cmd',internal,", $cmd, "\n");
      $ctx->{requests} = [];
      undef $ctx->{curRequest};
      # and this will leave the query partially initialized,
      # but it should never happen
      return;
    }
  };

And a catch-all just in case if the query compiler ever decides to produce an invalid request.

Next goes the handling of the dump labels (again, this gets set up during the build of the client reader threads, and then the nature is left to run its course, reacting to the rowops as they come in).

  $faOut->getLabel("beginDump")->makeChained("lbBeginDump", undef, sub {
    my $row = $_[1]->getRow();
    my ($client, $id, $name, $cmd) = $row->toArray();
    return unless ($client eq $fragment);
    if ($cmd eq "qdumpsub") {
      return unless(exists $queries{$id});
      my $ctx = $queries{$id};
      $fretOut->push($ctx->{curRequest}[4]); # the binding for the dump
    } else {
      # .. skipped the handling of dump/dumpsub
    }
  });

All it does is checks if this is the destination client, and if there is an active request with this id, then it pushes the appropriate binding.

  $faOut->getLabel("endDump")->makeChained("lbEndDump", undef, sub {
    my $row = $_[1]->getRow();
    my ($client, $id, $name, $cmd) = $row->toArray();
    return unless ($client eq $fragment);

    if ($cmd eq "qdumpsub") {
      return unless(exists $queries{$id});
      my $ctx = $queries{$id};
      $fretOut->pop($ctx->{curRequest}[4]); # the binding for the dump
      # and chain together all the following updates
      $ctx->{curRequest}[2]->makeChained(
        "qsub$id." . $ctx->{curRequest}[3]->getName(), undef,
        sub {
          # a cross-unit call
          $_[2]->call($_[3]->adopt($_[1]));
        },
        $ctx->{u}, $ctx->{curRequest}[3]
      );

      &$runNextRequest($ctx);
    } else {
      # .. skipped the handling of dump/dumpsub
    }
  });

Same things as the "beginDump", checks if this is the right client, and if it has an outstanding dump request, then pops the binding. After the dump is completed, the subscription has to be set up, so it sets up a label that forwards the normal output of this table to the label specified in the request. Since each query is defined in its own unit, this forwarding is done as a cross-unit call.

And then the next request of this query can be started.

By the way, the cross-unit adopt() didn't work in Perl until I wrote this example. There was a check against it (the C++ API never bothered with this check). But the adoption between the units has turned out to be quite convenient, so I've removed that check.

And that's it. Long and winding but finally completed. It's kind of about only three lines of code, but I think the rest of it also shows the useful techniques of the work with threads.

TQL server with multithreading

The next big example I've been talking about is finally ready. It's the adaptation of the TQL to work with the multithreaded server framework. The big reason for this example is the export of the table types through a nexus and creation of tables from them. And we'll get to that, but first let's look at the new abilities of the TQL.

TQL is still not of a production quality, in either single- or multi-threaded variety, and contains a large number of simplifying assumptions in its code. As the single-threaded version works symbiotically with the SimpleServer, the multithreaded version works with the ThreadedServer.

One thread created by the programmer contains the "core logic" of the model. It doesn't technically have to be all in a single thread: the data can be forwarded to the other threads and then the results forwarded back from them. But a single core logic thread is a convenient simplification. This thread has some input labels, to receive data from the outside, and some tables with the computed results that can be read by TQL. Of course, it's entirely realistic to have also just the output labels without tables, sending a stream or computed rowops, but again for simplicity I've left this out for now.

This core logic thread creates a TQL instance, which listens on a socket, accepts the connections, forwards the input data to the core logic, performs queries on the tables from the core logic and sends the results back to the client. To this end, the TQL instance creates a few nexuses in the core logic thread and uses them to communicate between all the fragments. The input labels and tables in the core thread also get properly connected to these nexuses. The following figure shows the thread architecture, I'll use it for the reference throughout the discussion:

Fig. 1. TQL application.

The core logic thread then goes into its main loop and performs as its name says the core logic computations.

Here is a very simple example of a TQL application:

sub appCoreT # (@opts)
{
  my $opts = {};
  &Triceps::Opt::parse("appCoreT", $opts, {@Triceps::Triead::opts,
    socketName => [ undef, \&Triceps::Opt::ck_mandatory ],
  }, @_);
  undef @_; # avoids a leak in threads module
  my $owner = $opts->{owner};
  my $app = $owner->app();
  my $unit = $owner->unit();

  # build the core logic

  my $rtTrade = Triceps::RowType->new(
    id => "int32", # trade unique id
    symbol => "string", # symbol traded
    price => "float64",
    size => "float64", # number of shares traded
  ) or confess "$!";

  my $ttWindow = Triceps::TableType->new($rtTrade)
    ->addSubIndex("byId",
      Triceps::SimpleOrderedIndex->new(id => "ASC")
    )
    or confess "$!";
  $ttWindow->initialize() or confess "$!";

  # Represents the static information about a company.
  my $rtSymbol = Triceps::RowType->new(
    symbol => "string", # symbol name
    name => "string", # the official company name
    eps => "float64", # last quarter earnings per share
  ) or confess "$!";

  my $ttSymbol = Triceps::TableType->new($rtSymbol)
    ->addSubIndex("bySymbol",
      Triceps::SimpleOrderedIndex->new(symbol => "ASC")
    )
    or confess "$!";
  $ttSymbol->initialize() or confess "$!";

  my $tWindow = $unit->makeTable($ttWindow, "EM_CALL", "tWindow")
    or confess "$!";
  my $tSymbol = $unit->makeTable($ttSymbol, "EM_CALL", "tSymbol")
    or confess "$!";

  # export the endpoints for TQL (it starts the listener)
  my $tql = Triceps::X::Tql->new(
    name => "tql",
    trieadOwner => $owner,
    socketName => $opts->{socketName},
    tables => [
      $tWindow,
      $tSymbol,
    ],
    tableNames => [
      "window",
      "symbol",
    ],
    inputs => [
      $tWindow->getInputLabel(),
      $tSymbol->getInputLabel(),
    ],
    inputNames => [
      "window",
      "symbol",
    ],
  );

  $owner->readyReady();

  $owner->mainLoop();
}

{
  my ($port, $thread) = Triceps::X::ThreadedServer::startServer(
      app => "appTql",
      main => \&appCoreT,
      port => 0,
      fork => -1, # create a thread, not a process
  );
}

This core logic is very simple: all it does is create two tables and then send the input data into them. The server gets started in a background thread (fork => -1) because this code is taken from a test that then goes and runs the expect with the SimpleClient.

The specification of inputs and tables for TQL is somewhat ugly but I kept it as it was historic (it was done this way to keep the parsing of the options simpler).  The new options compared to the single-threaded TQL are the "threadOwner", "inputs" and "inputNames". The "threadOwner" is how TQL knows that it must run in the multithreaded mode, and it's used to create the nexuses for communication between the core logic and the rest of TQL. The inputs are needed because the multithreaded TQL parses and forwards the input data, unlike the single-threaded version that relies on the SimpleServer to do that according to the user-defined dispatch table.

The names options don't have to be used: if you name your labels and tables nicely and suitable for the external vieweing, the renaming-for-export can be skipped.

Similar to the single-threaded version, if any of the options "tables" or "inputs" is used, the TQL object gets initialized automatically, otherwise the tables and inputs can be added piecemeal with addTable(), addNamedTable(), addInput(), addNamedInput(), and then the whole thing initialized manually.

Then the clients can establish the connections with the TQL server, send in the data and the queries. To jump in, here is a trace of a simple session that sends some data, then does some table dumps and subscribes, not touching the queries yet. I'll go through it fragment by fragment and explain the meaning. The dumps and subscribes were the warm-up exercises before writing the full queries, but they're useful in their own right, and here they serve as the warm-up exercises for the making of the queries!


> connect c1
c1|ready

The "connect" is not an actual command send but just the indication in the trace that the connection was set up by the client "c1" (it's a trace from the SimpleClient, so it follows the usual conventions). The "ready" response is set when the connection is opened, similar to the chat server shown before.

> c1|subscribe,s1,symbol
c1|subscribe,s1,symbol

This is a subscription request. It means "I'm not interested in the current state of a table but send me all the updates". The response is the mirror of the request, so that the client knows that the request has been processed. "s1" is the unique identifier of the request, so that the client can match together the responses it received to the requests it sent (and keeping the uniqueness is up to the client, the server may refuse the requests with duplicate identifiers).  And "symbol" is the name of the table. Once a subscription is in place, there is no way to unsubscribe other than by disconnecting the client (it's doable but adds complications, and I wanted to skip over the nonessential parts). Subscribing multiple times to the same table will send a confirmation every time but the repeated confirmations will have no effect: only one copy of the data will be sent anyway.

> c1|d,symbol,OP_INSERT,ABC,ABC Corp,1.0
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1

This sends the data into the model. And since it propagates through the subscription, the data gets sent back too. The "symbol" here means two different things: on the input side it's the name of the label where the data is sent, on the output side it's the name of the table that has been subscribed to.

The data lines start with the command "d" (since the data is sent much more frequently than the commands, I've picked a short one-letter "command name" for it), then the label/table name, opcode and the row fields in CSV format.

> c1|confirm,cf1
c1|confirm,cf1,,,

The "confirm" command provides a way for the client to check that the data it send had propagated through the model. And it doesn't have to subscribe back to the data and read them. Send some data lines, then send the "confirm" command and wait for it to come back (again, the unique id allows to keep multiple confirmations in flight if you please). This command doesn't guarantee that all the clients have seen the results from that data. It only guarantees that the core logic had seen the data, and more weakly guarantees that the data has been processed by the core logic, and this particular client had already seen all the results from it.

Why weakly? It has to do with the way it works inside, and it depends on the core logic. If the core logic consists of one thread, the guarantee is quite strong. But if the core logic farms out the work from the main thread to the other threads and then collects the results back, the guarantee breaks.

On the Fig. 1 you can see that unlike the chat server shown before, TQL doesn't have any private nexuses for communication between the reader and writer threads of a client. Instead it relies on the same input and output nexuses, adding a control label to them, to forward the commands from the reader to the writer. The TQL object in the core logic thread creates a short-circuit connection between the control labels in the input and output nexuses, forwarding the commands. And if the core logic all runs in one thread, this creates a natural pipeline: the data comes in, gets processed, comes out, the "confirm" command comes in, comes out after the data. But if the core logic farms out the work to more threads, the confirmation can "jump the line" because its path is a direct short circuit.

> c1|drain,dr1
c1|drain,dr1,,,

The "drain" is an analog of "confirm" but more reliable and slower:  the reader thread drains the whole model before sending the command on. This guarantees that all the processing is done, and all the output from it has been sent to all the clients.

> c1|dump,d2,symbol
c1|startdump,d2,symbol
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
c1|dump,d2,symbol

The "dump" command dumps the current contents of a table. Its result starts with "startdump", and the same id and table name as in the request, then goes the data (all with OP_INSERT), finishing with the completion confirmation echoing the original command. The dump is atomic, the contents of the table doesn't change in the middle of the dump. However if a subscription on this table is active, the data rows from that subscription may come before and after the dump.

I'm not going to describe the error reporting, but it's worth mentioning that if a command contains errors, its "confirmation" will be an error line with the same identifier.

> c1|dumpsub,ds3,symbol
c1|startdump,ds3,symbol
c1|d,symbol,OP_INSERT,ABC,ABC Corp,1
c1|dumpsub,ds3,symbol

The "dumpsub" command is a combination of a dump and subscribe: get the initial state and then get all the updates.  The confirmation of "dumpsub" marks the boundary between the original dump and the following updates.

> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
c1|d,symbol,OP_INSERT,DEF,Defense Corp,2

Send some more data, and it comes back only once, even though the subscription was done twice: once in "subscribe" and once in "dumpsub". The repeated subscription requests simply get consumed into one subscription.

> c1|d,window,OP_INSERT,1,ABC,101,10

This sends a row to the other table but nothing comes back because there is no subscription to that table.

> c1|dumpsub,ds4,window
c1|startdump,ds4,window
c1|d,window,OP_INSERT,1,ABC,101,10
c1|dumpsub,ds4,window
> c1|d,window,OP_INSERT,2,ABC,102,12
c1|d,window,OP_INSERT,2,ABC,102,12

This demonstrates the pure dump-and-subscribe without any interventions.

> c1|shutdown
c1|shutdown,,,,
c1|__EOF__

And the shutdown command works the same as in the chat server, draning and then shutting down the whole server.

Now on to the queries.

 > connect c1
c1|ready
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,1.0

Starts a client connection and sends some data.


> c1|querysub,q1,query1,{read table symbol}{print tokenized 0}
c1|d,query1,OP_INSERT,ABC,ABC Corp,1
c1|querysub,q1,query1

The "querysub" command does the "query-and-subscribe": reads the initial state of the table, processed through the query, and then subscribes to any future updates. The single-threaded variety of TQL doesn't do this, it does just the one-time queries. The multithreaded TQL could also do the one-time queries, and also just the subscribes without the initial state, but I've been cutting corners for this example and the only thing that's actually available is the combination of two, the "querysub".

"q1" is similar to the other command, the command identifier. The next field "query1" is the name for the query, it's the name that will be shown for the data lines coming out of the query. And then goes the query in the brace-quoted format, same as the single-threaded TQL (and there is no further splitting by commas, so the commas can be used freely in the query).

The identified and the name for the query sound kind of redundant. But the client may generate them in different ways and need both. The name has the more symbolic character. The identifier can be generated as a sequence of numbers,  so that the client can keep track of its progress more easily. And the error reports include the identifier but not the query name in them.

For the query, there is no special line coming out before the initial dump. Supposedly, there would not be more than one query in flight with the same name, so this could be easily told apart based on the name in the data lines. There is also an underlying consideration that when the query involves a join, in the future the initial dump might be happening in multiple chunks, requiring to either surround every chunk with the start-end lines or just let them go without the extra notifications, as they are now.
 
And the initial dump ends as usual with getting the echo of the command (without the query part) back.

This particular query is very simple and equivalent to a "dumpsub".


> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
c1|d,query1,OP_INSERT,DEF,Defense Corp,2

Send more data and it will come out of the query.


> c1|querysub,q2,query2,{read table symbol}{where istrue {$%symbol =~ /^A/}}{project fields {symbol eps}}
c1|t,query2,query2 OP_INSERT symbol="ABC" eps="1"
c1|querysub,q2,query2

This query is more complicated, doing a selection (the "where" query command) and projection. It also prints the results in the tokenized format (the "print" command gets added automatically if it wasn't used explicitly, and the default options for it enable the tokenized format).


The tokenized lines come out with the command "t", query name and then the contents of the row. The query name happens to be sent twice, and I'm not sure yet if it's a feature or a bug.

> c1|d,symbol,OP_INSERT,AAA,Absolute Auto Analytics Inc,3.0
c1|d,query1,OP_INSERT,AAA,Absolute Auto Analytics Inc,3
c1|t,query2,query2 OP_INSERT symbol="AAA" eps="3"
> c1|d,symbol,OP_DELETE,DEF,Defense Corp,2.0
c1|d,query1,OP_DELETE,DEF,Defense Corp,2

More examples of the data sent, getting processed by both queries.  In the second case the "where" filters out the row from query2, so only query1 produces the result.

> c1|shutdown
c1|shutdown,,,,
c1|__EOF__

And the shutdown as usual.

Now the "piece de resistance": queries with joins.

> connect c1
c1|ready
> c1|d,symbol,OP_INSERT,ABC,ABC Corp,2.0
> c1|d,symbol,OP_INSERT,DEF,Defense Corp,2.0
> c1|d,symbol,OP_INSERT,AAA,Absolute Auto Analytics Inc,3.0
> c1|d,window,OP_INSERT,1,AAA,12,100

Connect and send some starting data.

> c1|querysub,q1,query1,{read table window}{join table symbol byLeft {symbol} type left}
c1|t,query1,query1 OP_INSERT id="1" symbol="AAA" price="12" size="100" name="Absolute Auto Analytics Inc" eps="3"
c1|querysub,q1,query1

A left join of the tables "window" and "symbol", by the field "symbol" as join condition.

Note that unlike the previous single-threaded TQL examples, the index type path for the table "symbol" is not explicitly specified. It's the result of the new method TableType::findIndexPathForKeys() described before, now the index gets found automatically. And the single-threaded TQL now has this feature too. If you really want, you can still specify the index path but usually there is no need to.

The TQL joins, even in the multithreaded mode, are still implemented internally as LookupJoin, driven only by the main flow of the query. So the changes to the joined dimension tables will not update the query results, and will be visible only when a change on the main flow picks them up, potentially creating inconsistencies in the output. This is wrong, but fixing it presents complexities that I've left alone until some later time.

> c1|d,window,OP_INSERT,2,ABC,13,100
c1|t,query1,query1 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2"
> c1|d,window,OP_INSERT,3,AAA,11,200
c1|t,query1,query1 OP_INSERT id="3" symbol="AAA" price="11" size="200" name="Absolute Auto Analytics Inc" eps="3"

Sending data updates the results of the query.

> c1|d,symbol,OP_DELETE,AAA,Absolute Auto Analytics Inc,3.0
> c1|d,symbol,OP_INSERT,AAA,Alcoholic Abstract Aliens,3.0

As described above, the modifications of the dimension table are mot visible in the query directly.

> c1|d,window,OP_DELETE,1
c1|t,query1,query1 OP_DELETE id="1" symbol="AAA" price="12" size="100" name="Alcoholic Abstract Aliens" eps="3"

But an update on the main flow brings them up (an in this case inconsistently, the row getting deleted is not exactly the same as the row inserted before).

> c1|querysub,q2,query2,{read table window}{join table symbol byLeft {symbol} type left}{join table symbol byLeft {eps} type left rightFields {symbol/symbol2}}
c1|t,query2,query2 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2" symbol2="ABC"
c1|t,query2,query2 OP_INSERT id="2" symbol="ABC" price="13" size="100" name="ABC Corp" eps="2" symbol2="DEF"
c1|t,query2,query2 OP_INSERT id="3" symbol="AAA" price="11" size="200" name="Alcoholic Abstract Aliens" eps="3" symbol2="AAA"
c1|querysub,q2,query2

This is a more complicated query, involving two joins, with the same dimension table "symbol". The second join by "eps" makes no real-world sense whatsoever but it's interesting from the technical perspective: if you check the table type of this table at the start of the post, you'll find that it has no index on the field "eps". The join adds this index on demand!

The way it works, all the dimension tables are copied into the client's writer thread, created from the table types exported by the core logic throuhg the output nexus. (And if a table is used in the same query twice, it's currently also copied twice). This provides a nice opportunity to amend the table type by adding any necessary secondary index before creating the table, and TQL makes a good use of it.

The details are forthcoming in the next post.

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.

Table in C++

The Table is defined in table/Table.h. It inherits from Gadget, with the table's output label being the gadget's output label. Naturally, it's an Starget and usable from one thread only.

 It's constructor is not public, and it's created from the TableType with its  method makeTable():

Autoref<Table> t = tabType->makeTable(unit, Table::EM_CALL, "t");

The arguments are the unit where the table will belong, the enqueueing mode for its output label (this is a legacy argument and will go away soon), and the name of the table.

For the reference, that TableType method is:

Onceref<Table> makeTable(Unit *unit, Gadget::EnqMode emode, const string &name) const;

The table has a large number of methods, grouped into multiple subsets.

EnqMode getEnqMode() const;
const string &getName() const;
Unit *getUnit() const;
Label *getLabel() const;

These methods are inherited from the Gadget. The only special thing to remember is that getLabel() returns the table's output label. Technically, getName() has an overriding implementation in the Table, to return the table's name while its output label has the suffix ".out" appended to it.


const TableType *getType() const;
const RowType *getRowType() const;
const RowHandleType *getRhType() const;

Get back the type of the table, of its rows, and its row handles.

Label *getInputLabel() const;
Label *getPreLabel() const;
Label *getDumpLabel() const;
Label *getAggregatorLabel(const string &agname) const;

Get the assorted labels. The aggregator label getter takes the name of the aggregator (as was defined in the TableType) as an argument.

FnReturn *fnReturn() const;

Get the FnReturn of this table's outputs. It gets created and remembered on the first call, and the subsequent calls return the same object.  It has a few labels with the fixed names: "out", "pre" and "dump", and a label for each aggregator with the aggregator's name. It could throw an Exception if you name an aggregator to conflict with one of the fixed labels, which you should not. The return's name will be "tableName.fret".

Next go the operations on the table (and of course the table may also be modified by sending rowops to its input label).
RowHandle *makeRowHandle(const Row *row) const;

Create a row handle for a row. Remember, the row handles are reference-counted, and also have the special kind of references with Rhref. So the returned pointer should be stored in an Rhref. The row handle created will not be inserted into the table yet.

bool insert(RowHandle *rh);

Insert a row handle into the table. This invokes all the row replacement policies along the way. If the handle is already in the table, does nothing and returns false. May also return false if a replacement policy refuses the row, but in practice there are no such refusing policies yet. Otherwise returns true.

It may throw an Exception. It may throw by itself if the row handle doesn't belong to this table or propagate the exception up: since the execution involves calling the output labels and such, an exception might be thrown from there.

bool insertRow(const Row *row);

The version that combines the row handle construction and insertion. Unlike Perl, in C++ this method is named differently instead of overloading. The comments about the replacement policies and return code, and about exceptions apply here too.

void remove(RowHandle *rh);

Remove a row handle from the table. If the handle is not in the table, silently does nothing. May throw an Exception.

bool deleteRow(const Row *row);

Find a matching row and delete it. Returns true if the row was found and removed, false if not found. May throw an Exception.

void clear(size_t limit = 0);

Clear the table by removing all the rows from it. The removed rows are sent as usual to the "pre" and "out" labels. If the limit is not 0, no more than that number of the rows will be removed. The rows are removed in the usual order of the first leaf index.

Next go the iteration methods. The rule of thumb is that for them a NULL row handle pointer means "end of iteration" or "not found" (or sometimes "bad arguments"). And they can handle the NULL row handles OK on the input, just returning NULL on the output.

RowHandle *begin() const;

Get the first row handle in the default order of the first leaf index. If the table is empty, returns NULL.

RowHandle *beginIdx(IndexType *ixt) const;

Get the first handle in the order of a particular index. The index type must belong to this table's type. For an incorrect index type it returns NULL (perhaps in the future this will be changed to an exception).

RowHandle *next(const RowHandle *cur) const;
RowHandle *nextIdx(IndexType *ixt, const RowHandle *cur) const;

Get the next row handle in the order of  the default or specific index. Returns NULL after the last handle. It's safe to pass the current row handle as NULL, the result will be NULL, and also on any other error.

RowHandle *firstOfGroupIdx(IndexType *ixt, const RowHandle *cur) const;
RowHandle *lastOfGroupIdx(IndexType *ixt, const RowHandle *cur) const;

Get the first or last row handle in the same group as the current row according to a non-leaf index. The NULL current handle will cause NULL returned. See the details in the description of the Perl API.

RowHandle *nextGroupIdx(IndexType *ixt, const RowHandle *cur) const;

Get the first row handle of the next group. The return will be NULL if the current group was the last one, or if the current handle is NULL.

Next go the size operations:

size_t size() const;


Get the number of rows currently in the table.


size_t groupSizeIdx(IndexType *ixt, const RowHandle *what) const;

Get the size of the group where the handle belongs according to a non-leaf index. If any arguments are wrong, returns 0. The row handle doesn't have to be in the table. If it isn't in the table, the method will find the group where the row would belong if it were inserted and return its current size.

size_t groupSizeRowIdx(IndexType *ixt, const Row *what) const;

A convenience version that makes a row handle from a row, finds the group size and disposes of the handle.

Next go the finding methods:

RowHandle *find(const RowHandle *what) const;
RowHandle *findIdx(IndexType *ixt, const RowHandle *what) const;

Find the handle of a matching row according to the default (first leaf) or the specific index, or return NULL if not found.

RowHandle *findRow(const Row *what) const;
RowHandle *findRowIdx(IndexType *ixt, const Row *what) const;

The convenience versions that create a temporary row handle and then perform the search.

Next goes the dump API that sends the whole contents of the table to the "dump" label, thus making any labels connected to it perform an implicit iteration over the table.

void dumpAll(Rowop::Opcode op = Rowop::OP_INSERT) const;
void dumpAllIdx(IndexType *ixt, Rowop::Opcode op = Rowop::OP_INSERT) const;

The dump can go in the order of default or specific index. The opcode argument is used for the rowops sent on the dump label. Using the argument index type of NULL makes dumpAllIdx() use the default index and work just like DumpAll(). In the furute there probably will be methods that dump only a group of records.

As usual, the general logic of the methods matches the Perl API unless said otherwise. Please refer to the Perl API description for the details and examples.

Table dump

Another intermediate step for the example I'm working on is the table dumping. It allows to iterate on a table in a functional manner.

A new label "dump" is added to the table and its FnReturn. Whenever the method dumpAll() is called, it sends the whole contents of the table to that label. Then you can set a binding on the table's FnReturn, call dumpAll(), and the binding will iterate through the whole table's contents.

The grand plan is also to add the dumping by a a condition that selects a sub-index, but it's not implemented yet.

It's also possible to dump in an alternative order: dumpAllIdx() can send the rows in the order of any index, rather than the default first leaf index.

If you want to get the dump label explicitly, you can do it with

my $dlab = $table->getDumpLabel();

Normally the only reason to do that would be to add it to another FnReturn (besides the table's FnReturn). Chaining anything else directly to this label would not make much sense, because the dump of the table can be called from many places, and the directly chained label will receive data every time the dump is called.

The typical usage looks like this:

    Triceps::FnBinding::call(
        name => "iterate",
        on => $table->fnReturn(),
        unit => $unit,
        labels => [
            dump => sub { ... },        ],
        code => sub {
            $table->dumpAll();
        },
    );

It's less efficient than the normal iteration but sometimes comes handy.

Normally the rowops are sent with the opcode OP_INSERT. But the opcode can also be specified explicitly:

$table->dumpAll($opcode);

The alternative order can be achieved with:

$table->dumpAllIdx($indexType);
$table->dumpAllIdx($indexType, $opcode);

As usual, the index type must belong to the exact type of this table. For example:

$table->dumpAllIdx($table->getType()->findIndexPath("cb"), "OP_NOP");

And some more interesting examples will be forthcoming later.

Streaming functions and tables

The Copy Tray used in the tables in the version 1.0 was really a precursor to the streaming functions. Now when the full-blown streaming functions became worked out, there is no sense in keeping the copy trays any more, so I've removed them.

Instead, I've added a Table method that gets the FnReturn for that table:

$fret = $table->fnReturn();

The return contains the labels "pre", "out", and the named labels for each aggregators. The FnReturn object is created on the first call of this method and is kept in the table. All the following calls return the same object. This has some interesting consequences for the "pre" label: the rowop for the "pre" label doesn't get created at all if there is nothing chained from that label. But when the FnReturn gets created, one of its labels gets chained from the "pre" label. Which means that once, you call $table->fnReturn() for the first time, you will see that table's "pre" label called in all the traces. It's not a huge extra overhead, but still something to keep in mind and not be surprised when calling fnReturn() changes all your traces.

The produced FnReturn then gets used as any other one. If you use it with an FnBinding that has withTrace => 1, you get an improved equivalent of the Copy Tray. For example:

$fret2 = $t2->fnReturn();
$fbind2 = Triceps::FnBinding->new(
    unit => $u1,
    name => "fbind2",
    on => $fret2,
    withTray => 1,
    labels => [
        out => sub { }, # another way to make a dummy
    ],
);

$fret2->push($fbind2);
$t2->insert($r2);
$fret2->pop($fbind2);

# $ctr is the Copy Tray analog
$ctr = $fbind2->swapTray(); # get the updates on an insert

Of course, most of the time you would not want to make a dummy label and then iterate manually through the copy tray. You would want to create bindings to the actual next logical labels and simply execute them, immediately or delayed with a tray.

Table clearing

The table clearing has been coming up repeatedly, so I've made a convenience method for it. In Perl it's called as:

$table->clear();
$table->clear($limit);

If $limit is absent or 0, the whole table gets cleared. If it's greater than 0, no more than this number of records will be deleted. A negative limit is an error. The deletion happens in the usual order of the first leaf index, and the rowops are sent to the table's output label as usual. It's really the same thing as running a loop over all the row handles and removing them, only in C++ it's more efficient than in Perl. There is no return value, the errors cause a confess().

The C++ API has this method in the Table as well:

void clear(size_t limit = 0);

even more stuff for 1.0

I've added the function Triceps::Fields::makeTranslation() that makes the output field filtering and renaming, similar to the one in joins, easy to insert into the random user-defined templates. Again, look in the upcoming docs, the chapter on templates for the description.

The build instructions have been completely rewritten and greatly clarified. The build configuration settings became easier too.

The Table methods for row finding and iteration now confess on errors, since checking their error codes otherwise is completely impractical.

A lot more methods have been converted to the new-style error handling, with confessing on errors.

more updates

Some more stuff has been getting cleaned up:

$unit->setTracer($tracer);

now confesses on errors, so the problem with its error checking is fixed.

The tables now allow to create rowops of rows of all matching types, not only of the equal types. The approach with matching types was not consistent with what the labels did, so I've changed it.

The TableType now has the method

$tt->getRowType() 

that has the name consistent with the tables and labels. The old method rowType() also still exists.

The Opt::ck_ref() now also accepts the subclasses of the defined classes.

The new helper Fields::isStringType() has been added.

There also have been some major addition of the examples:

I've added a pretty big example of the main loop that includes the full socket handling in the chapter on Scheduling. It's not exactly production-ready but gives some idea.

Another addition in the chapter on Scheduling is an example of a topological loop that computes the Fibonacci numbers.

Many new examples have been added to the chapter on Templates.

error handling in the Perl wrappers

Since the API started the transition to confessing on the fatal errors instead of just returning an undef and an error message, I've converted the Perl wrapper methods to do the same. This includes:

AggregatorContext::makeHashSend()
AggregatorContext::makeArraySend()

Label::makeRowopHash()
Label::makeRowopArray()

Table::findBy()Table::findIdxBy() 

Unit::makeHashCall()
Unit::makeArrayCall()
Unit::makeHashSchedule()
Unit::makeArraySchedule()
Unit::makeHashLoopAt()
Unit::makeArrayLoopAt() 

Large deletes in small chunks

If you have worked with Coral8 and similar CEP systems, you should be familiar with the situation when you ask it to delete a million rows from the table and the model goes into self-contemplation for half an hour, not reacting to any requests. It starts responding again only when the deletes are finished. That's because the execution is single-threaded, and deleting a million rows takes time.

Triceps is succeptible to the same issue. So, how to avoid it? Even better, how to make the deletes work "in background", at a low priority, kicking in only when there is no other pending requests?

The solution is do do it in smaller chunks. Delete a few rows (say, a thousand or so) then check if there are any other requests. Keep processing these other request until the model becomes idle. Then continue with deleting the next chunk of rows.

Let's make a small example of it. First, let's make a table.

our $uChunks = Triceps::Unit->new("uChunks") or confess "$!";

# data is just some dumb easily-generated filler
our $rtData = Triceps::RowType->new(
    s => "string",
    i => "int32",
) or confess "$!";

# the data is auto-generated by a sequence
our $seq = 0;

our $ttData = Triceps::TableType->new($rtData)
    ->addSubIndex("fifo", Triceps::IndexType->newFifo())
or confess "$!";
$ttData->initialize() or confess "$!";
our $tData = $uChunks->makeTable($ttData,
    &Triceps::EM_CALL, "tJoin1"
) or confess "$!";
makePrintLabel("lbPrintData", $tData->getOutputLabel());

The data in the table is completely silly, just something to put in there. Even the index is a simple FIFO, just something to keep the table together. And the data will be put in there by the main loop in an equally silly way:

while(<STDIN>) {
    chomp;
    my @data = split(/,/); # starts with a command, then string opcode
    my $type = shift @data;
    if ($type eq "data") {
        my $count = shift @data;
        for (; $count > 0; $count--) {
            ++$seq;
            $uChunks->makeHashCall($tData->getInputLabel(), "OP_INSERT",
                s => ("data_" . $seq),
                i => $seq,
            ) or confess "$!";
        }
    } elsif ($type eq "dump") {
        for (my $rhit = $tData->begin(); !$rhit->isNull(); $rhit = $rhit->next()) {
            print("dump: ", $rhit->getRow()->printP(), "\n");
        }
    }
    $uChunks->drainFrame(); # just in case, for completeness
}

When we send the command like "data,3", the mail loop will insert 3 new rows into the table. The contents is generated with sequential numbers, so the rows can be told apart. As the table gets changed, the updates get printed by the label lbPrintData. Also the contents of the table can be dumped with the main loop command "dump". Now let's add a main loop command to clear the table, initially by going through all the data and deleting it at once.

# notifications about the clearing
our $rtNote = Triceps::RowType->new(
    text => "string",
) or confess "$!";

our $lbReportNote = $uChunks->makeDummyLabel($rtNote, "lbReportNote"
) or confess "$!";
makePrintLabel("lbPrintNote", $lbReportNote);

# code that clears the table
our $lbClear = $uChunks->makeLabel($rtNote, "lbClear", undef, sub {
    my $next;
    for (my $rhit = $tData->begin(); !$rhit->isNull(); $rhit = $next) {
        $next = $rhit->next(); # advance before removal
        $tData->remove($rhit);
    }
    $uChunks->makeHashCall($lbReportNote, "OP_INSERT",
        text => "done clearing",
    ) or confess "$!";
}) or confess "$!";

It's done in a bit round-about way: the main loop will send the clearing notification row to the label lbClear. Which does the clearing, then sends a notification that the clearing has completed to the label lbReportNote. Which eventually gets printed.

In the real world not the whole table would be erased but only the old data, from before a certain date. I've shown it before in the traffic aggregation example, ordering the rows by a date field, and deleting until you see a newer row. Here for simplicity all the data get wiped out.

The part of the main loop responsible for the clearing command is:

    elsif ($type eq "clear") {
        $uChunks->makeHashCall($lbClear, "OP_INSERT",
            text => "clear",
        ) or confess "$!";
    } 

With the basic clearing done, time to add the chunking logic. First, add a tray to collect things that need to be done when the model is idle:

our $trayIdle = $uChunks->makeTray();

Then modify the lbClear code to work with the limited chunks:

# code that clears the table in small chunks
our $lbClear = $uChunks->makeLabel($rtNote, "lbClear", undef, sub {
    my $limit = 2; # no more than 2 rows per run
    my $next;
    for (my $rhit = $tData->begin(); !$rhit->isNull(); $rhit = $next) {
        if ($limit-- <= 0) {
            # request to be called again when the model becomes idle
            $trayIdle->push($_[0]->adopt($_[1]));
            return;
        }
        $next = $rhit->next(); # advance before removal
        $tData->remove($rhit);
    }
    $uChunks->makeHashCall($lbReportNote, "OP_INSERT",
        text => "done clearing",
    ) or confess "$!";
}) or confess "$!";

Since it's real inconvenient to play with a million rows, we'll play with just a few rows. And so the chunk size limit is also set smaller, to just two rows instead of a thousand. When the limit is reached, the code pushes the command row to the idle tray for later rescheduling and returns. The adoption part is not strictly necessary, and this small example would work fine without it. But it's a safeguard for the more complicated programs that may have the labels chained, with our clearing label being just one link in a chain. If the incoming rowop gets rescheduled as is, the whole chain will get executed again. which might not be desirable. Re-adopting it to our label will cause only our label (okay, and everything chained from it) to be executed.

How would the rowops in the idle tray get executed? In the real world, the main loop logic would be like this pseudocode:

while(1) {
  if (idle tray empty)
    timeout = infinity;
  else
    timeout = 0;
  poll(file descriptors, timeout);
  if (poll timed out)
    run the idle tray;
  else
    process the incoming data;
}

But it's hugely inconvenient for a toy demonstration, getting the timing right would be a major pain. So instead let's just add an extra command "idle" to the main loop, to trigger the idle logic at will:

    elsif ($type eq "idle") {
        $uChunks->schedule($trayIdle);
        $trayIdle->clear();
    }

And while at it, let's make the "dump" command show the contents of the idle tray as well:

        for my $r ($trayIdle->toArray()) {
            print("when idle: ", $r->printP(), "\n");
        }

All the pieces have been put together, let's run the code. As usual, the input lines are shown in italics:

data,1
tJoin1.out OP_INSERT s="data_1" i="1" 
clear
tJoin1.out OP_DELETE s="data_1" i="1" 
lbReportNote OP_INSERT text="done clearing" 

This is pretty much a dry run: put in one row (less than the chunk size), see it deleted on clearing. And see the completion reported afterwards.

data,5
tJoin1.out OP_INSERT s="data_2" i="2" 
tJoin1.out OP_INSERT s="data_3" i="3" 
tJoin1.out OP_INSERT s="data_4" i="4" 
tJoin1.out OP_INSERT s="data_5" i="5" 
tJoin1.out OP_INSERT s="data_6" i="6" 

Add more data, which will be enough for three chunks.

clear
tJoin1.out OP_DELETE s="data_2" i="2" 
tJoin1.out OP_DELETE s="data_3" i="3" 

Now the clearing  does one chunk and stops, waiting for the idle condition.

dump
dump: s="data_4" i="4" 
dump: s="data_5" i="5" 
dump: s="data_6" i="6" 
when idle: lbClear OP_INSERT text="clear" 

See what's inside: the remaining 3 rows, and a row in the idle tray saying that the clearing is in progress.

idle
tJoin1.out OP_DELETE s="data_4" i="4" 
tJoin1.out OP_DELETE s="data_5" i="5" 

The model goes idle once more, one more chunk of two rows gets deleted.

data,1
tJoin1.out OP_INSERT s="data_7" i="7"
dump
dump: s="data_6" i="6" 
dump: s="data_7" i="7" 
when idle: lbClear OP_INSERT text="clear" 

What will happen if we add more data in between the chunks of clearing? Let's see, let's add one more row. It shows up in the table as usual.

idle
tJoin1.out OP_DELETE s="data_6" i="6" 
tJoin1.out OP_DELETE s="data_7" i="7" 
lbReportNote OP_INSERT text="done clearing" 
dump
idle

And on the next idle condition the clearing picks up whatever was in the table for the next chunk. Since there were only two rows left, it's the last chunk, and the clearing reports a successful completion. And a dump shows that there is nothing left in the table nor in the idle tray. And the next idle condition does nothing, because the idle tray is empty.

The delete-by-chunks logic can be made into a template, just I'm not sure yet what is the best way to do it. It would have to have a lot of configurable parts.

 On another subject, scheduling the things to be done on idle adds an element of unpredictability to the model. It's impossible to predict the exact timing of the incoming requests, and the idle work may get inserted between any of them. Presumably it's OK because the data being deleted should not be participating in any logic at this time any more. For repeatability in the unit tests, make the chunk size adjustable and adjust it to a size larger than the biggest amount of data used in the unit tests.

A similar logic can also be used in querying the data. But it's more difficult. For deletion the continuation is easy: just take the first row in the index, and it will be the place to continue (because the index is ordered correctly, and because the previous rows are getting deleted). For querying you would have to remember the next row handle and continue from it. Which is OK if it can not get deleted in the meantime. But if it can get deleted, you'll have to keep track of that too, and advance to the next row handle when this happens. And if you want to receive a full snapshot with the following subscription to all updates, you'd have to check whether the modified rows are before or after the marked handle, and pass them through if they are before it, letting the user see the updates to the data already received. And since the data is being sent to the user, filling up the output buffer and stopping would stop the whole model too, and not restart until the user reads the buffered data. So there has to be a flow control logic that would stop the query when output buffer fills up, return to the normal operation, and then reschedule the idle job for the query only when the output buffer drains down. I've kind of started on doing an example of the chunked query too, but then because of all these complications decided to leave it for later.

The new error handling

I've been considering the change to the error handling for a while. The repeating checks of the calls for "or confess" are pretty annoying, just dying/confessing by default would be better. And then if anyone wants to catch that death, they can use eval {} around the call.

The need to check for the recursive modification attempts in the tables struck me as something that could particularly benefit from just dying rather than returning an error code that would likely be missed. This pushed me towards starting the shift towards this new error handling scheme.

With the interaction through the Perl and the native C++ code, unrolling the call stack is a pretty tricky proposition but I've got it worked out. If the error is not caught with eval, it keeps unrolling the call sequence through both the Perl call stack and the Triceps unit call stack. If the error is caught with eval, the message and the whole stack trace will be as usual in $@.

The bad news is that so far only a limited subset of the calls use the new scheme. The rest are still setting $! and returning an undef. So overall it's a mix and you need to remember, which calls work which way. In the future versions eventually everything will be converted to the new scheme. For a bit of backwards-compatibility, the error messages from the new-style dying calls are saved in both $@ and $!. This will eventually go away, and only $@ will be used.

The converted calls are:

• The Table modification methods:
• insert()
• remove()
• deleteRow()
• The Unit methods that deal with the scheduling:
• schedule()
• fork()
• call()
• enqueue()
• setMark()
• loopAt()
• callNext()
• drainFrame()
• clearLabels()

If the code dies in the middle of executing a Perl label's handler, the scheduler will catch it as before, but instead of just printing an error message on stderr and continuing, it will start unrolling the Unit call stack, collecting the label call trace until it eventually gets back to the Perl level, which in order will die. If not caught by eval, that would continue the unrolling, until the outermost Perl code dies and prints the whole stack trace.

Note though that right now this works only with the label handlers. The handlers for the tracers, aggregators, sorted indexes etc. still work in the old way, with an error message printed to stderr.

These errors from the label handlers are not to be treated lightly. Usually you can't just catch them be an eval and continue on your way. The reason is that as the Unit scheduling stack gets unrolled, any unprocessed rowops in it get thrown away. By the time you catch the error, the data is probably in an inconsistent state, and you can't just dust off and continue. You would have to reset your model to a good state first. Treat such errors as near-fatal. It could have been possible to keep going through the scheduled rowops, collecting the errors along the way and then returning the whole pile. But it seems more important to report the error as soon as possible. And anyway, if something has died somewhere, it has probably already left some state inconsistent, and continuing to run forward as normal would just pile up crap upon crap. If you want the errors to be handled early and lightly, make sure that your Perl code doesn't die in the first place.

Another added item is an explicit check that the labels are not called recursively. That is, if a label is called, it can not call itself again, directly or through the other labels, until it returns. Such recursive calls don't hurt anything by themselves but they are a bad design practice, and it seems more important to catch the accidental errors of this kind early than to leave the door open for the intentional use of them by design. If you want a label's processing to loop back to itself, the proper way it to arrange it with schedule() or loopAt().

Tables: no more bundling

I believe I've told before that the table first processes all the rows from an operation on it (only one row is the argument of the operation but it may trigger the deletion of multiple rows with the replacement policies) and only then sends all the results. This is essentially an implicit bundling of the rowops, and has all the issues of the bundling that have been described before. Since a join sees the rowops only after they come out of the table, the processing of the missing matches for the outer joins (described in the last post) could not work reliably. If there are multiple rows at the same join key affected by an operation, when the join looks in the table, it would see the state after all of them have been already applied, and would make the wrong decisions. So, how does it work?

The answer is that now I've changed the way the tables work. No more implicit bundling. Each row gets changed in the table, and a rowop is immediately called on the table's output label. The handler of that rowop can read the table and see it exactly in the state right after that rowop was applied, and none more. Nice, consistent, convenient.

Note though that any labels called from this point may only read the table, not modify it. The table is still in the middle of the previous modification, and starting a new modification at this point would corrupt it. So if you want to modify the table, you have to schedule it for later, after the current, modification is completed, using the Unit methods schedule() or loopAt(). But keep in mind that by the time that rowop gets called, many other changes may have already happened to the table. So it's best to schedule not the direct table changes but the more high-level operations which would look at the state of the table at their run time and decide the proper action.

The order of sending the aggregation results has also changed. It used to be the table changes, then all the aggregation results. Now first the aggregation handlers get called with AO_BEFORE_MOD and their results get sent through, then the table modifications work through as described, and then the aggregation handlers get called with AO_AFTER_* and their results go through. The aggregation modifications are still bundled: all the aggregators get called with their results remembered, and then all the result are sent through. This is not very pretty but not such a big deal either. The reason is that the aggregation code has to detect whether each modification is the last one for each aggregation group or not. And it's hard enough to do in the bundled way, and would be quite difficult to unbundle. Besides, the aggregators are specially designed to improve their efficiency by throwing away the intermediate updates on the same group, so there is not much use in unbundling that.

Another new feature of the table is the "pre" label. It can be found with:

$lb = $table->getPreLabel();

It has the name of "TableName.pre". A rowop on this label gets called right before applying the row to the table. Just as with the table's output label, the code that handles that rowop can't do any other direct modification to the table and can't prevent the ongoing modifications from happening. However if it reads the table, it will find it in exactly the state before that modification gets applied. Which comes useful sometimes, in particular for the self-joins that will be shown later. There is also a bit of optimization going on: since the "pre" label gets used fairly rarely, the table code first checks if there are any labels chained from it. If none, the "pre" label doesn't get called at all. If you do the unit tracing, you won't see the call of the "pre" label in the trace unless there are other labels chained from it.

To recap, the new high-level order of the table operation processing is:

• Execute the replacement policies on all the indexes, find all the rows that need to be deleted first.
• If any of the index policies forbid the modification, return 0.
• Call all the aggregators with AO_BEFORE_MOD on all the affected rows.
• Send these aggregator results.
• For each affected row:
• Call the "pre" label (if it has any labels chained to it).
• Modify the row in the table.
• Call the "out" label.
• Call all the aggregators with AO_AFTER_*, on all the affected rows.
• Send these aggregator results.

 And to let the join find whether some row is the only one in the group or not, I've added another call:

$size = $table->groupSizeIdx($idxType, $row_or_rh);

It works very similar to the AggregatorContext::groupSize(), only it has no context and has to get the index type and row or row handle as its arguments. It returns the count of rows in the group. If there is no such group in the table, the result will be 0. If the argument is a row handle, that handle may be in the table or not in the table, either will be handled transparently (though calling it for a row handle in the table is more efficient because the group would not need to be looked up first). If the argument is a row, it gets handled similarly to findIdx(): a temporary row handle gets created, used to find the result, and then destroyed.

The $idxType is the one that owns the split. Naturally, it must be a non-leaf index. (Using a non-leaf index type is not an error but it always returns 0, because there are no groups under it). It's basically the same index type as you would use in findIdx() to find the first row of the group. For example, if you have a table type defined as

our $ttPosition = Triceps::TableType->new($rtPosition)
  ->addSubIndex("primary",
    Triceps::IndexType->newHashed(key => [ "date", "customer", "symbol" ])
  ) 
  ->addSubIndex("currencyLookup", # for joining with currency conversion
    Triceps::IndexType->newHashed(key => [ "date", "currency" ])
    ->addSubIndex("grouping", Triceps::IndexType->newFifo())
  ) 
  ->addSubIndex("byDate", # for cleaning by date
    Triceps::SimpleOrderedIndex->new(date => "ASC")
    ->addSubIndex("grouping", Triceps::IndexType->newFifo())
  ) 
or die "$!";

Then it would make sense to call groupSizeIdx on the indexes "currencyLookup" or "byDate" but not on "primary", "currencyLookup/grouping" nor "byDate/grouping". Remember, a non-leaf index type defines the groups, and the nested index types under it define the order in those groups (and possibly further break them down into sub-groups).