As you can see from the previous descriptions, building a new Triead is a serious business, containing many moving part. Doing it every time from scratch would be hugely annoying and error prone. The class BasicPthread, defined in app/BasicPthread.h, takes care of wrapping all that complicated logic.
It originated as a subclass of pw::pwthread, and even though it ended up easier to copy and modify the code (okay, maybe this means that pwthread can be made more flexible), the usage is still very similar to it. You define a new subclass of BasicPthread, and define the virtual function execute() in it. Then you instantiate the object and call the method start() with the App argument.
For a very simple example:
class MainLoopPthread : public BasicPthread
{
public:
MainLoopPthread(const string &name):
BasicPthread(name)
{
}
// overrides BasicPthread::start
virtual void execute(TrieadOwner *to)
{
to->readyReady();
to->mainLoop();
}
};
...
Autoref<MainLoopPthread> pt3 = new MainLoopPthread("t3");
pt3->start(myapp);
It will properly create the Triead, TrieadOwner, register the thread joiner and start the execution. The TrieadOwner will pass through to the execute() method, and its field fi_ will contain the reference to the FileInterrupt object. After execute() returns, it will take care of marking the thread as dead.
It also wraps the call of execute() into a try/catch block, so any Exceptions thrown will be caught and cause the App to abort. In short, it's very similar to the Triead management in Perl.
You don't need to keep the reference to the thread object afterwards, you can even do the construction and start in one go:
(new MainLoopPthread("t3"))->start(myapp);
The internals of BasicPthread will make sure that the object will be dereferenced (and thus, in the absence of other references, destroyed) after the thread gets joined by the harvester.
Of course, if you need to pass more arguments to the thread, you can define them as fields in your subclass, set them in the constructor (or by other means between constructing the object and calling start()), and then execute() can access them. Remember, execute() is a method, so it receives not only the TrieadObject as an argument but also the BasicPthread object as this.
BasicPthread is implemented as a subclass of TrieadJoin, and thus is an Mtarget. It provides the concrete implementation of the joiner's virtual methods, join() and interrupt(). Interrupt() calls the method of the base class, then sends the signal SIGUSR2 to the target thread.
And finally the actual reference:
BasicPthread(const string &name);
Constructor. The name of the thread is passed through to App::makeTriead(). The Triead will be constructed in start(), the BasicPthread constructor just collects together the arguments.
void start(Autoref<App> app);
Construct the Triead, create the POSIX thread, and start the execution there.
void start(Autoref<TrieadOwner> to);
Similar to the other version of start() but uses a pre-constructed TrieadOwner object. This version is useful mostly for the tests, and should not be used much in the real life.
virtual void execute(TrieadOwner *to);
Method that must be redefined by the subclass, containing the threads's logic.
virtual void join();
virtual void interrupt();
Methods inherited from TrieadJoin, providing the proper implementations for the POSIX threads.
And unless, I've missed something, this concludes the description of the Triceps threads API.
Sunday, July 14, 2013
Saturday, July 13, 2013
FileInterrupt reference (C++)
FileInterrupt is the class that keeps track of a bunch of file descriptor and revokes them on demand, hopefully interrupting any ongoing operations on them (and if that doesn't do the job, a separately sent signal will). It's not visible in Perl, being intergrated into the TrieadOwner methods, but in C++ it's a separate class. It's defined in app/FileInterrupt.h, and is an Mtarget, since the descriptors are registered and revoked from different threads.
FileInterrupt();
The constructor, absolutely plain. Normally you would not want to construct it directly but use the object already constructed in TrieadJoin. The object keeps the state, whether the interruption had happened, and is obviously initialized to the non-interrupted state.
void trackFd(int fd);
Add a file descriptor to the tracked interruptable set. If the interruption was already done, the descriptor will instead be revoked right away by dupping over from /dev/null. If the attempt to open /dev/null fails, it will throw an Exception.
void forgetFd(int fd);
Remove a file descriptor to the tracked interruptable set. You must do it before closing the descriptor, or a race leading to the corruption of random file descriptors may occur. If this file descriptor was not registered, the call will be silently ignored.
void interrupt();
Perform the revocation of all the registered file descriptors by dupping over them from /dev/null. If the attempt to open /dev/null fails, it will throw an Exception.
This marks the FileInterrupt object state as interrupted, and any following trackFd() calls will lead to the immediate revocation of the file descriptors in them, thus preventing any race conditions.
bool isInterrupted() const;
Check whether this object has been interrupted.
FileInterrupt();
The constructor, absolutely plain. Normally you would not want to construct it directly but use the object already constructed in TrieadJoin. The object keeps the state, whether the interruption had happened, and is obviously initialized to the non-interrupted state.
void trackFd(int fd);
Add a file descriptor to the tracked interruptable set. If the interruption was already done, the descriptor will instead be revoked right away by dupping over from /dev/null. If the attempt to open /dev/null fails, it will throw an Exception.
void forgetFd(int fd);
Remove a file descriptor to the tracked interruptable set. You must do it before closing the descriptor, or a race leading to the corruption of random file descriptors may occur. If this file descriptor was not registered, the call will be silently ignored.
void interrupt();
Perform the revocation of all the registered file descriptors by dupping over them from /dev/null. If the attempt to open /dev/null fails, it will throw an Exception.
This marks the FileInterrupt object state as interrupted, and any following trackFd() calls will lead to the immediate revocation of the file descriptors in them, thus preventing any race conditions.
bool isInterrupted() const;
Check whether this object has been interrupted.
TrieadJoin reference (C++)
TrieadJoin is the abstract base class that tells the harvester how to join a thread after it had finished. Obviously, it's present only in the C++ API and not in Perl (and by the way, the reference of the Perl classes in 2.0 has been completed, the remaining classes are only in C++).
Currently TrieadJoin has two subclasses: PerlTrieadJoin for the Perl threads and BasicPthread for the POSIX threads in C++. I won't be describing PerlTriedJoin, since it's in the internals of the Perl implementation, never intended to be directly used by the application developers, and if you're interested, you can always look at its source code. I'll describe BasicPthread later.
Well, actually there is not a whole lot of direct use for TrieadJoin either: you need to worry about it only if you want to define a joiner for some other kind of threads, and this is not very likely. But since I've started, I'll complete the write-up of it.
So, if you want to define a joiner for some other kind of threads, you define a subclass of it, with an appropriately defined method join().
TrieadJoin is an Mtarget, naturally referenced from multiple threads (at the very least it's created in the thread to be joined or its parent, and then passed to the harvester thread by calling App::defineJoin()). The methods of TrieadJoin are:
TrieadJoin(const string &name);
The constrcutor. The name is the name of the Triead, used for the error messages. Due to the various syncronization reasons, this makes the life of the harvester much easier, than trying to look up the name from the Triead object.
virtual void join() = 0;
The Most Important joining method to be defined by the subclass. The subclass object must also hold the identity of the thread in it, to know which thread to join. The harvester will call this method.
virtual void interrupt();
The method that interrupts the target thread when it's requested to die. It's called in the context of the thread that triggers the App shutdown (or otherwise requests the target thread to die). By default the TrieadJoin carries a FileInterrupt object in it (it gets created on TrieadJoin construction, and then TrieadJoin keeps a reference to it), that will get called by this method to revoke the files. But everything else is a part of the threading system, and the base class doesn't know how to do it, the subclasses must define their own methods, wrapping the base class.
Both PerlTrieadJoin and BasicPthread add sending the signal SIGUSR2 to the target thread. For that they use the same target thread identity kept in the object as used by the join() call.
FileInterrupt *fileInterrupt() const;
Get a pointer to the FileInterrupt object defined in the TrieadJoin. The most typical use is to pass it to the TrieadOwner object, so that it can be easily found later:
to->fileInterrupt_ = fileInterrupt();
Though of course it could be kept in a separate local Autoref instead.
const string &getName() const;
Get back the name of the joiner's thread.
Currently TrieadJoin has two subclasses: PerlTrieadJoin for the Perl threads and BasicPthread for the POSIX threads in C++. I won't be describing PerlTriedJoin, since it's in the internals of the Perl implementation, never intended to be directly used by the application developers, and if you're interested, you can always look at its source code. I'll describe BasicPthread later.
Well, actually there is not a whole lot of direct use for TrieadJoin either: you need to worry about it only if you want to define a joiner for some other kind of threads, and this is not very likely. But since I've started, I'll complete the write-up of it.
So, if you want to define a joiner for some other kind of threads, you define a subclass of it, with an appropriately defined method join().
TrieadJoin is an Mtarget, naturally referenced from multiple threads (at the very least it's created in the thread to be joined or its parent, and then passed to the harvester thread by calling App::defineJoin()). The methods of TrieadJoin are:
TrieadJoin(const string &name);
The constrcutor. The name is the name of the Triead, used for the error messages. Due to the various syncronization reasons, this makes the life of the harvester much easier, than trying to look up the name from the Triead object.
virtual void join() = 0;
The Most Important joining method to be defined by the subclass. The subclass object must also hold the identity of the thread in it, to know which thread to join. The harvester will call this method.
virtual void interrupt();
The method that interrupts the target thread when it's requested to die. It's called in the context of the thread that triggers the App shutdown (or otherwise requests the target thread to die). By default the TrieadJoin carries a FileInterrupt object in it (it gets created on TrieadJoin construction, and then TrieadJoin keeps a reference to it), that will get called by this method to revoke the files. But everything else is a part of the threading system, and the base class doesn't know how to do it, the subclasses must define their own methods, wrapping the base class.
Both PerlTrieadJoin and BasicPthread add sending the signal SIGUSR2 to the target thread. For that they use the same target thread identity kept in the object as used by the join() call.
FileInterrupt *fileInterrupt() const;
Get a pointer to the FileInterrupt object defined in the TrieadJoin. The most typical use is to pass it to the TrieadOwner object, so that it can be easily found later:
to->fileInterrupt_ = fileInterrupt();
Though of course it could be kept in a separate local Autoref instead.
const string &getName() const;
Get back the name of the joiner's thread.
SIGUSR2
When a thread is requested to die, its registered file descriptors become revoked, and the signal SIGUSR2 is sent to it to interrupt any ongoing system calls. For this to work correctly, there must be a signal handler defined on SIGUSR2, because otherwise the default reaction to it is to kill the process. It doesn't matter what signal handler, just some handler must be there. The Triceps library defines an empty signal handler but you can also define your own instead.
In Perl, the empty handler for SIGUSR2 is set when the module Triceps.pm is loaded. You can change it afterwards.
In C++ Triceps provides a class Sigusr2, defined in app/Sigusr2.h, to help with this. If you use the class BasicPthread, you don't need to deal with Sigusr2 directly: BasicPthread takes care of it. All the methods of Sigusr2 are static.
static void setup();
Set up an empty handler for SIGUSR2 if it hasn't been done yet. This class has a static flag (synchronized by a mutex) showing that the handler had been set up. On the first call it sets the handler and sets the flag. On the subsequent calls it checks the flag and does nothing.
static void markDone();
Just set the flag that the setup has been done. This allows to set your own handler instead and still cooperate with the logic of Sigusr2 and BasicPthread.
If you set your custom handler before any threads have been started, then set up your handler and then call markDone(), telling Sigusr2 that there is no need to set the handler any more.
If you set your custom handler when the Triceps threads are already running (not the best idea but still a possibility), there is a possibility of a race with another thread calling setup(). To work around that race, set up your handler, call markDone(), then set up your handler again.
static void reSetup();
This allows to replace the custom handler with the empty one. It always forcibly sets the empty handler (and also the flag).
In Perl, the empty handler for SIGUSR2 is set when the module Triceps.pm is loaded. You can change it afterwards.
In C++ Triceps provides a class Sigusr2, defined in app/Sigusr2.h, to help with this. If you use the class BasicPthread, you don't need to deal with Sigusr2 directly: BasicPthread takes care of it. All the methods of Sigusr2 are static.
static void setup();
Set up an empty handler for SIGUSR2 if it hasn't been done yet. This class has a static flag (synchronized by a mutex) showing that the handler had been set up. On the first call it sets the handler and sets the flag. On the subsequent calls it checks the flag and does nothing.
static void markDone();
Just set the flag that the setup has been done. This allows to set your own handler instead and still cooperate with the logic of Sigusr2 and BasicPthread.
If you set your custom handler before any threads have been started, then set up your handler and then call markDone(), telling Sigusr2 that there is no need to set the handler any more.
If you set your custom handler when the Triceps threads are already running (not the best idea but still a possibility), there is a possibility of a race with another thread calling setup(). To work around that race, set up your handler, call markDone(), then set up your handler again.
static void reSetup();
This allows to replace the custom handler with the empty one. It always forcibly sets the empty handler (and also the flag).
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.
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.
AutoDrain reference, C++
The scoped drain in C++ has more structure than in Perl. It consists of the base class AutoDrain and two subclasses: AutoDrainShared and AutoDrainExclusive. The base AutoDrain can not be created directly but is convenient for keeping a reference to any kind of drain:
Autoref<AutoDrain> drain = new AutoDrainShared(app);
The constructor of the subclass determines whether the drain is shared or exclusive, and the rest of the methods are defined in the base class.
The AutoDrain is an Starget, and naturally can be used in only one thread. It's also possible to use the a direct local variable:
{
AutoDrainShared drain(app);
...
}
Just remember not to mix the metaphors, if you create a local variable, don't try to create the references to it.
The constructors are:
AutoDrainShared(App *app, bool wait = true);
AutoDrainShared(TrieadOwner *to, bool wait = true);
Create a shared drain from an App or TrieadOwner. The wait flag determines if the constructor will wait for the drain to complete, otherwise it will return immediately. The shared drain requests the draining of all the Trieads, and multiple threads may have their shared drains active at the same time (the release will happen when all these drains become released). A shared drain will wait for all the preceding exclusive drains to be released before it gets created.
AutoDrainExclusive(TrieadOwner *to, bool wait = true);
Create an exclusive drain from a TrieadOwner. The wait flag makes the constructor wait for the completion of the drain. Only one exclusive drain at a time may be active, from only one thread, an exclusive drain will wait for all the preceding shared and exclusive drains to be released before it gets created.
The common method is:
void wait();
Wait for the drain to complete. Can be called repeatedly. If more data has been injected into the model through the excluded Triead, will wait for that data to drain.
I'm not sure if I mentioned this yet, but if the new Trieads are created while a drain is active, these Trieads will be notified of the drain. This means that the input-only Trieads won't be able to send any data until the drain is released. However the Trieads in the middle of the model will follow the normal protocol for such threads: the drain will become incomplete after the Triead is marked as ready and until it blocks on the following nextXtray(). Normally this should be a very short amount of time. However such Trieads should take care to never send any rowops on their own before reading from nextXtray(), if they find that TrieadOwner::isRdDrain() returns true, because this may introduce the data into the model at a very inconvenient moment, when some logic expects that no data is changing, and cause a corruption. This is the same caveat as for using nextXtray() varieties with the timeouts: if you want to send data on a timeout, always check isRqDrain(), and never send any data on timeouts if isRqDrain() returns true.
I've also realized that I might have used the word "undrain"/"undrained" for two different meanings: one is for the drain release, the other is for the condition when there is some data being processed in the model (whether a drain is active or not, but if it's active, it will not be completed). Perhaps I should make it less ambiguous. But for now just please keep this ambiguity in mind.
Autoref<AutoDrain> drain = new AutoDrainShared(app);
The constructor of the subclass determines whether the drain is shared or exclusive, and the rest of the methods are defined in the base class.
The AutoDrain is an Starget, and naturally can be used in only one thread. It's also possible to use the a direct local variable:
{
AutoDrainShared drain(app);
...
}
Just remember not to mix the metaphors, if you create a local variable, don't try to create the references to it.
The constructors are:
AutoDrainShared(App *app, bool wait = true);
AutoDrainShared(TrieadOwner *to, bool wait = true);
Create a shared drain from an App or TrieadOwner. The wait flag determines if the constructor will wait for the drain to complete, otherwise it will return immediately. The shared drain requests the draining of all the Trieads, and multiple threads may have their shared drains active at the same time (the release will happen when all these drains become released). A shared drain will wait for all the preceding exclusive drains to be released before it gets created.
AutoDrainExclusive(TrieadOwner *to, bool wait = true);
Create an exclusive drain from a TrieadOwner. The wait flag makes the constructor wait for the completion of the drain. Only one exclusive drain at a time may be active, from only one thread, an exclusive drain will wait for all the preceding shared and exclusive drains to be released before it gets created.
The common method is:
void wait();
Wait for the drain to complete. Can be called repeatedly. If more data has been injected into the model through the excluded Triead, will wait for that data to drain.
I'm not sure if I mentioned this yet, but if the new Trieads are created while a drain is active, these Trieads will be notified of the drain. This means that the input-only Trieads won't be able to send any data until the drain is released. However the Trieads in the middle of the model will follow the normal protocol for such threads: the drain will become incomplete after the Triead is marked as ready and until it blocks on the following nextXtray(). Normally this should be a very short amount of time. However such Trieads should take care to never send any rowops on their own before reading from nextXtray(), if they find that TrieadOwner::isRdDrain() returns true, because this may introduce the data into the model at a very inconvenient moment, when some logic expects that no data is changing, and cause a corruption. This is the same caveat as for using nextXtray() varieties with the timeouts: if you want to send data on a timeout, always check isRqDrain(), and never send any data on timeouts if isRqDrain() returns true.
I've also realized that I might have used the word "undrain"/"undrained" for two different meanings: one is for the drain release, the other is for the condition when there is some data being processed in the model (whether a drain is active or not, but if it's active, it will not be completed). Perhaps I should make it less ambiguous. But for now just please keep this ambiguity in mind.
Friday, July 12, 2013
AutoDrain reference, Perl
The AutoDrain class creates the drains on an App with the automatic scoping. When the returned AutoDrain object gets destroyed, the drain becomes released. So placing the object into a lexically-scoped variable in a block with cause the undrain on the block exit. Placing it into another object will cause the undrain on deletion of that object. And just not storing the object anywhere works as a barrier: the drain gets completed and then immediately undrained, guaranteeing that all the previously sent data is processed and then continuing with the processing of the new data.
All the drain caveats described in the App class apply to the automatic drains too.
$ad = Triceps::AutoDrain::makeShared($app);
$ad = Triceps::AutoDrain::makeShared($to);
Create a shared drain and wait for it to complete. A drain may be created from either an App or a TrieadOwner object. Returns the AutoDrain object.
$ad = Triceps::AutoDrain::makeSharedNoWait($app);
$ad = Triceps::AutoDrain::makeSharedNoWait($to);
Same as makeShared() but doesn't wait for the drain to complete before returning. May still sleep if an exclusive drain is currently active.
$ad = Triceps::AutoDrain::makeExclusive($to);
Create an exclusive drain on a TrieadOwner and wait for it to complete. Returns the AutoDrain object. Normally the excluded thread should be input-only. Such an input-only thread is allowed to send more data in without blocking (to wait for the app become drained again after that, use the method wait()).
$ad = Triceps::AutoDrain::makeExclusiveNoWait($to);
Same as makeExclusive() but doesn't wait for the drain to complete before returning. May still sleep if a shared or another exclusive drain is currently active.
$ad->wait();
Wait for the drain to complete. Particularly useful after the NoWait creation, but can also be used to wait for the App to become drained again after injecting some rowops through the excluded Triead of the exclusive drain.
$ad->same($ad2);
Check that two AutoDrain references point to the same object.
All the drain caveats described in the App class apply to the automatic drains too.
$ad = Triceps::AutoDrain::makeShared($app);
$ad = Triceps::AutoDrain::makeShared($to);
Create a shared drain and wait for it to complete. A drain may be created from either an App or a TrieadOwner object. Returns the AutoDrain object.
$ad = Triceps::AutoDrain::makeSharedNoWait($app);
$ad = Triceps::AutoDrain::makeSharedNoWait($to);
Same as makeShared() but doesn't wait for the drain to complete before returning. May still sleep if an exclusive drain is currently active.
$ad = Triceps::AutoDrain::makeExclusive($to);
Create an exclusive drain on a TrieadOwner and wait for it to complete. Returns the AutoDrain object. Normally the excluded thread should be input-only. Such an input-only thread is allowed to send more data in without blocking (to wait for the app become drained again after that, use the method wait()).
$ad = Triceps::AutoDrain::makeExclusiveNoWait($to);
Same as makeExclusive() but doesn't wait for the drain to complete before returning. May still sleep if a shared or another exclusive drain is currently active.
$ad->wait();
Wait for the drain to complete. Particularly useful after the NoWait creation, but can also be used to wait for the App to become drained again after injecting some rowops through the excluded Triead of the exclusive drain.
$ad->same($ad2);
Check that two AutoDrain references point to the same object.
Thursday, July 11, 2013
no more explicit confessions
It's official: all the code has been converted to the new error handling. Now if anything goes wrong, the Triceps Perl calls just confess right away. No more need for the pattern 'or confess "$!"' that was used throughout the code (though of course you can still use it for handling the other errors).
It also applies to the error checks done by the XS typemaps, these will also confess automatically.
I've also added one more method that doesn't confess: IndexType::getTabtypeSafe(). If the index type is not set into a table type, it will silently return an undef without any error indications.
On a related note, the construction of the Type subclasses has been made nicer in the C++: instead of calling abort() on the major errors, they now throw Exceptions. Mind you, these exceptions are thrown not in the constructors as such but in the chainable methods that set the contents of the types. And they try to be smart enough to preserve the reference count correctness: if the object was not assigned into any reference yet (as is typical for the chained calls), they take care to temporarily increase and decrease the reference count, thus freeing the object, before throwing. Of course, the default reaction to Exceptions is still to dump core, but need be, these exceptions can be caught.
It also applies to the error checks done by the XS typemaps, these will also confess automatically.
I've also added one more method that doesn't confess: IndexType::getTabtypeSafe(). If the index type is not set into a table type, it will silently return an undef without any error indications.
On a related note, the construction of the Type subclasses has been made nicer in the C++: instead of calling abort() on the major errors, they now throw Exceptions. Mind you, these exceptions are thrown not in the constructors as such but in the chainable methods that set the contents of the types. And they try to be smart enough to preserve the reference count correctness: if the object was not assigned into any reference yet (as is typical for the chained calls), they take care to temporarily increase and decrease the reference count, thus freeing the object, before throwing. Of course, the default reaction to Exceptions is still to dump core, but need be, these exceptions can be caught.
Sunday, July 7, 2013
safe functions in RowHandle
As I'm updating the error reporting in the Perl methods, there is one more class that has grown the safe (non-confessing functions). In RowHandle now the method
$row = $rh->getRow();
confesses if the RowHandle is NULL. The method
$row = $rh->getRowSafe();
returns an undef in this situation, just like getRow() used to, only now it doesn't set the text in $! any more. A consequence is that some of the Aggregator examples that branch directly on checking whether a row handle contains NULL, now had to be changed to use getRowSafe().
The method
$result = $rh->isInTable();
has also been updated for the case when it contains a NULL: now it simply returns 0 (instead of undef) and doesn't set the text in $!.
$row = $rh->getRow();
confesses if the RowHandle is NULL. The method
$row = $rh->getRowSafe();
returns an undef in this situation, just like getRow() used to, only now it doesn't set the text in $! any more. A consequence is that some of the Aggregator examples that branch directly on checking whether a row handle contains NULL, now had to be changed to use getRowSafe().
The method
$result = $rh->isInTable();
has also been updated for the case when it contains a NULL: now it simply returns 0 (instead of undef) and doesn't set the text in $!.
Saturday, July 6, 2013
findSubIndexSafe
The scalar leakage in Carp::confess was causing an unpleasant issue with the functions that were trying to look up the nested indexes and catch when they went missing. So, similarly to the string conversions, I've added the method findSubIndexSafe() to the TableType and IndexType:
$ixt = $tt->findSubIndexSafe($name);
$ixt = $ixt_parent->findSubIndexSafe($name);
If the name is not found, it silently returns an undef, without setting any error codes.
Eventually the issue with the leakage in confess() would have to be fixed, but for now it's a good enough plug for the most typical cases. I'll need to think about other methods that could use the safe treatment.
$ixt = $tt->findSubIndexSafe($name);
$ixt = $ixt_parent->findSubIndexSafe($name);
If the name is not found, it silently returns an undef, without setting any error codes.
Eventually the issue with the leakage in confess() would have to be fixed, but for now it's a good enough plug for the most typical cases. I'll need to think about other methods that could use the safe treatment.
Friday, July 5, 2013
carp carpity carp
I've found that calling Carp::confess (more exactly, even the lower-level function Carp::longmess that gets called by Carp::confess and by Triceps's error reporting from the C++ code) in a threaded program leaks the scalars, apparently by leaving garbage on the Perl stack.
The problem seems to be in the line "package DB;" in the middle of one of its internal functions. Perhaps changing the package in the middle of a function is not such a great idea, leaving some garbage on the stack. The most interesting part is that this line can be removed altogether, with no adverse effects, and then the leak stops.
Oh, well, looks like the homebrewn Triceps::confess will be coming soon. And I'd like to find some contact to get the stock Carp fixed.
The problem seems to be in the line "package DB;" in the middle of one of its internal functions. Perhaps changing the package in the middle of a function is not such a great idea, leaving some garbage on the stack. The most interesting part is that this line can be removed altogether, with no adverse effects, and then the leak stops.
Oh, well, looks like the homebrewn Triceps::confess will be coming soon. And I'd like to find some contact to get the stock Carp fixed.
Thursday, July 4, 2013
string-and-constant conversion errors
As outlined before, I'm going through the the Perl API, changing to the new way of the error reporting by confessing. On of the consequences is that the functions for parsing the string name of a constant into the value got converted too. They now confess on an invalid string. However sometimes it's still convenient to check the string for validity without causing an error, so for that I've added another version of them, with the suffix Safe attached, that still work the old way, returning an undef for an unknown string.
For example, &Triceps::humanStringTracerWhen and Triceps::humanStringTracerWhenSafe.
The same applies the other way around to: the functions converting the integer enumeration value to string will confess when the value is not in the proper enum. The functions with the suffix Safe will still return an undef. Such as &Triceps::tracerWhenHumanString and &Triceps::tracerWhenHumanStringSafe.
The only special case is &Triceps::opcodeString and &Triceps::opcodeStringSafe: neither of them confesses nor returns undef, they print the unknown strings as a combination of bits, but for consistency both names are present anyway, doing the same thing.
For example, &Triceps::humanStringTracerWhen and Triceps::humanStringTracerWhenSafe.
The same applies the other way around to: the functions converting the integer enumeration value to string will confess when the value is not in the proper enum. The functions with the suffix Safe will still return an undef. Such as &Triceps::tracerWhenHumanString and &Triceps::tracerWhenHumanStringSafe.
The only special case is &Triceps::opcodeString and &Triceps::opcodeStringSafe: neither of them confesses nor returns undef, they print the unknown strings as a combination of bits, but for consistency both names are present anyway, doing the same thing.
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.
$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.
Tuesday, July 2, 2013
Facet reference, C++
The general functioning of a facet is the same in C++ as in Perl, so please refer to the part 1 of the Perl description for this information.
However the construction of a Facet is different in C++. The import is still the same: you call TrieadOwner::importNexus() method or one of its varieties and it returns a Facet. However the export is different: first you construct a Facet from scratch and then give it to TrieadOwner::exportNexus() to create a Nexus from it.
In the C++ API the Facet has a notion of being imported, very much like say the FnReturn has a notion of being initialized. When a Facet is first constructed, it's not imported. Then exportNexus() creates the nexus from the facet, exports it into the App, and also imports the nexus information back into the facet, marking the facet as imported. It returns back a reference to the exact same Facet object, only now that object becomes imported. Obviously, you can use either the original or returned reference, they point to the same object. Once a Facet has been imported, it can not be modified any more. The Facet object returned by the importNexus() is also marked as imported, so it can not be modified either. And also an imported facet can not be exported again.
However exportNexus() has an exception. If the export is done with the argument import set to false, the facet object will be left unchanged and not marked as imported. A facet that is not marked as imported can not be used to send or receive data. Theoretically, it can be used to export another nexus, but practically this would not work because that would be an attempt to export another nexus with the same name from the same thread. In reality such a Facet can only be thrown away, and there is not much use for it. You can read its components and use them to construct another Facet but that's about it.
It might be more convenient to use the TrieadOwner::makeNexus*() methods to build a Facet object rather than building it directly. In either case, the methods are the same and accept the same arguments, just the Facet methods return a Facet pointer while the nexus maker methods return a NexusMaker pointer.
The Facet class is defined in app/Facet.h. It inherits from Mtarget for an obscure reason that has to do with App topology analysis but it's intended to be used from one thread only.
You don't have to keep your own references to all your Facets. The TrieadOwner will keep a reference to all the imported Facets, and they will not be destroyed while the TrieadOwner exists (and this applies to Perl as well).
enum {
DEFAULT_QUEUE_LIMIT = 500,
};
The default value used for the nexus queue limit (as a count of Xtrays, not rowops). Since the reading from the nexus involves double-buffering, the real queue size might grow up to twice that amount.
static string buildFullName(const string &tname, const string &nxname);
Build the full nexus name from its components.
Facet(Onceref<FnReturn> fret, bool writer);
Create a Facet, initially non-imported (and non-exported). The FnReturn object fret defines the set of labels in the facet (and nexus), and the name of FnReturn also becomes the name of the Facet, and of the Nexus. The writer flag determines whether this facet will become a writer (if true) or a reader (if false) when a nexus is created from it. If the nexus gets created without importing the facet back, the writer flag doesn't matter and can be set either way.
The FnReturn should generally be not initialized yet. The Facet constructor will check if FnReturn already has the labels _BEGIN_ and _END_ defined, and if either is missing, will add it to the FnReturn, then initialize it. If both _BEGIN_ and _END_ are already present, then the FnReturn can be used even if it's already initialized. But if any of them is missing, FnReturn must be not initialized yet, otherwise the Facet constructor will fail to add these labels.
The same FnReturn object may be used to create only one Facet object. And no, you can not import a Facet, get an FnReturn from it, then use it to create another Facet.
If anything goes wrong, the constructor will not throw but will remember the error, and later the exportNexus() will find it and throw an Exception from it.
static Facet *make(Onceref<FnReturn> fret, bool writer);
Same as the constructor, used for the more convenient operator priority for the chained calls.
static Facet *makeReader(Onceref<FnReturn> fret);
static Facet *makeWriter(Onceref<FnReturn> fret);
Syntactic sugar around the constructor, hardcoding the writer flag.
Normally the facets are constructed and exported with the chained calls, like:
Autoref<Facet> myfacet = to->exportNexus(
Facet::makeWriter(FnReturn::make("My")->...)
->setReverse()
->exportTableType(Table::make(...)->...)
);
Because of this, the methods that are used for post-construction return the pointer to the original Facet object. They also almost never throw the Exceptions, to prevent the memory leaks through the orphaned Facet objects. The only way an Exception might get thrown is on an attempt to use these methods on an already imported Facet. Any errors get collected, and eventually exportNexus() will find them and properly throw an Exception, making sure that the Facet object gets properly disposed of.
Facet *exportRowType(const string &name, Onceref<RowType> rtype);
Add a row type to the Facet. May throw an Exception if the the facet is already imported. On other errors remembers them to be thrown on an export attempt.
Facet *exportTableType(const string &name, Autoref<TableType> tt);
Add a table type to the Facet. May throw an Exception if the the facet is already imported. On other errors remembers them to be thrown on an export attempt. The table type must also be deep-copyable and contain no errors. Not sure if I described this before, but if the deep copy can not proceed (say, a table type involves a Perl sort condition with a direct reference to the compiled Perl code) the deepCopy() method must still return a newly created object but remember the error inside it. Later when the table type is initialized, that object's initialization must return this error. The exportTableType() does a deep copy then initializes the copied table type. If this detects any errors, they get remembered and cause an Exception later in exportNexus().
Facet *setReverse(bool on = true);
Set (or clear) the nexus reverse flag. May throw an Exception if the the facet is already imported.
Facet *setQueueLimit(int limit);
Set the nexus queue limit. May throw an Exception if the the facet is already imported.
Erref getErrors() const;
Get the collected errors, so that they can be found without an export attempt.
bool isImported() const;
Check whether this facet is imported.
The rest of the methods are the same as in Perl. They can be used even if the facet is not imported.
bool isWriter() const;
Check whether this is a writer facet (or if returns false, a reader facet).
bool isReverse() const;
Check whether the underlying nexus is reverse.
int queueLimit() const;
Get the queue size limit of the nexus. Until the facet is exported, this will always return the last value set by setQueueLimit(). However if the nexus is reverse, on import the value will be changed to a very large integer value, currently INT32_MAX, and on all the following calls this value will be returned. Technically speaking, the queue size of the reverse nexuses is not unlimited, it's just very large, but in practice it amounts to the same thing.
FnReturn *getFnReturn() const;
Get the FnReturn object. If you plan to destroy the Facet object soon after this method is called, make sure that you put the FnReturn pointer into an Autoref first.
const string &getShortName() const;
Get the short name, AKA "as-name", which is the same as the FnReturn's name.
const string &getFullName() const;
Get the full name of the nexus imported through this facet. If the facet is not imported, will return an empty string.
typedef map<string, Autoref<RowType> > RowTypeMap;const RowTypeMap &rowTypes() const;
Get the map of the defined row types. Returns the reference to the Facet's internal map object.
typedef map<string, Autoref<TableType> > TableTypeMap;
const TableTypeMap &tableTypes() const;
Get the map of defined table types. Returns the reference to the Facet's internal map object.
RowType *impRowType(const string &name) const;
Find a single row type by name. If the name is not known, returns NULL.
TableType *impTableType(const string &name) const;
Find a single table type by name. If the name is not known, returns NULL.
Nexus *nexus() const;
Get the nexus of this facet. If the facet is not imported, returns NULL.
int beginIdx() const;
int endIdx() const;
Return the indexes (as in "integer offset") of the _BEGIN_ and _END_ labels in FnReturn.
bool flushWriter();
Flush the collected rowops into the nexus as a single Xtray. If there is no data collected, does nothing. Returns true on a successful flush (even if there was no data collected), false if the Triead was requested to die and thus all the data gets thrown away.
However the construction of a Facet is different in C++. The import is still the same: you call TrieadOwner::importNexus() method or one of its varieties and it returns a Facet. However the export is different: first you construct a Facet from scratch and then give it to TrieadOwner::exportNexus() to create a Nexus from it.
In the C++ API the Facet has a notion of being imported, very much like say the FnReturn has a notion of being initialized. When a Facet is first constructed, it's not imported. Then exportNexus() creates the nexus from the facet, exports it into the App, and also imports the nexus information back into the facet, marking the facet as imported. It returns back a reference to the exact same Facet object, only now that object becomes imported. Obviously, you can use either the original or returned reference, they point to the same object. Once a Facet has been imported, it can not be modified any more. The Facet object returned by the importNexus() is also marked as imported, so it can not be modified either. And also an imported facet can not be exported again.
However exportNexus() has an exception. If the export is done with the argument import set to false, the facet object will be left unchanged and not marked as imported. A facet that is not marked as imported can not be used to send or receive data. Theoretically, it can be used to export another nexus, but practically this would not work because that would be an attempt to export another nexus with the same name from the same thread. In reality such a Facet can only be thrown away, and there is not much use for it. You can read its components and use them to construct another Facet but that's about it.
It might be more convenient to use the TrieadOwner::makeNexus*() methods to build a Facet object rather than building it directly. In either case, the methods are the same and accept the same arguments, just the Facet methods return a Facet pointer while the nexus maker methods return a NexusMaker pointer.
The Facet class is defined in app/Facet.h. It inherits from Mtarget for an obscure reason that has to do with App topology analysis but it's intended to be used from one thread only.
You don't have to keep your own references to all your Facets. The TrieadOwner will keep a reference to all the imported Facets, and they will not be destroyed while the TrieadOwner exists (and this applies to Perl as well).
enum {
DEFAULT_QUEUE_LIMIT = 500,
};
The default value used for the nexus queue limit (as a count of Xtrays, not rowops). Since the reading from the nexus involves double-buffering, the real queue size might grow up to twice that amount.
static string buildFullName(const string &tname, const string &nxname);
Build the full nexus name from its components.
Facet(Onceref<FnReturn> fret, bool writer);
Create a Facet, initially non-imported (and non-exported). The FnReturn object fret defines the set of labels in the facet (and nexus), and the name of FnReturn also becomes the name of the Facet, and of the Nexus. The writer flag determines whether this facet will become a writer (if true) or a reader (if false) when a nexus is created from it. If the nexus gets created without importing the facet back, the writer flag doesn't matter and can be set either way.
The FnReturn should generally be not initialized yet. The Facet constructor will check if FnReturn already has the labels _BEGIN_ and _END_ defined, and if either is missing, will add it to the FnReturn, then initialize it. If both _BEGIN_ and _END_ are already present, then the FnReturn can be used even if it's already initialized. But if any of them is missing, FnReturn must be not initialized yet, otherwise the Facet constructor will fail to add these labels.
The same FnReturn object may be used to create only one Facet object. And no, you can not import a Facet, get an FnReturn from it, then use it to create another Facet.
If anything goes wrong, the constructor will not throw but will remember the error, and later the exportNexus() will find it and throw an Exception from it.
static Facet *make(Onceref<FnReturn> fret, bool writer);
Same as the constructor, used for the more convenient operator priority for the chained calls.
static Facet *makeReader(Onceref<FnReturn> fret);
static Facet *makeWriter(Onceref<FnReturn> fret);
Syntactic sugar around the constructor, hardcoding the writer flag.
Normally the facets are constructed and exported with the chained calls, like:
Autoref<Facet> myfacet = to->exportNexus(
Facet::makeWriter(FnReturn::make("My")->...)
->setReverse()
->exportTableType(Table::make(...)->...)
);
Because of this, the methods that are used for post-construction return the pointer to the original Facet object. They also almost never throw the Exceptions, to prevent the memory leaks through the orphaned Facet objects. The only way an Exception might get thrown is on an attempt to use these methods on an already imported Facet. Any errors get collected, and eventually exportNexus() will find them and properly throw an Exception, making sure that the Facet object gets properly disposed of.
Facet *exportRowType(const string &name, Onceref<RowType> rtype);
Add a row type to the Facet. May throw an Exception if the the facet is already imported. On other errors remembers them to be thrown on an export attempt.
Facet *exportTableType(const string &name, Autoref<TableType> tt);
Add a table type to the Facet. May throw an Exception if the the facet is already imported. On other errors remembers them to be thrown on an export attempt. The table type must also be deep-copyable and contain no errors. Not sure if I described this before, but if the deep copy can not proceed (say, a table type involves a Perl sort condition with a direct reference to the compiled Perl code) the deepCopy() method must still return a newly created object but remember the error inside it. Later when the table type is initialized, that object's initialization must return this error. The exportTableType() does a deep copy then initializes the copied table type. If this detects any errors, they get remembered and cause an Exception later in exportNexus().
Facet *setReverse(bool on = true);
Set (or clear) the nexus reverse flag. May throw an Exception if the the facet is already imported.
Facet *setQueueLimit(int limit);
Set the nexus queue limit. May throw an Exception if the the facet is already imported.
Erref getErrors() const;
Get the collected errors, so that they can be found without an export attempt.
bool isImported() const;
Check whether this facet is imported.
The rest of the methods are the same as in Perl. They can be used even if the facet is not imported.
bool isWriter() const;
Check whether this is a writer facet (or if returns false, a reader facet).
bool isReverse() const;
Check whether the underlying nexus is reverse.
int queueLimit() const;
Get the queue size limit of the nexus. Until the facet is exported, this will always return the last value set by setQueueLimit(). However if the nexus is reverse, on import the value will be changed to a very large integer value, currently INT32_MAX, and on all the following calls this value will be returned. Technically speaking, the queue size of the reverse nexuses is not unlimited, it's just very large, but in practice it amounts to the same thing.
FnReturn *getFnReturn() const;
Get the FnReturn object. If you plan to destroy the Facet object soon after this method is called, make sure that you put the FnReturn pointer into an Autoref first.
const string &getShortName() const;
Get the short name, AKA "as-name", which is the same as the FnReturn's name.
const string &getFullName() const;
Get the full name of the nexus imported through this facet. If the facet is not imported, will return an empty string.
typedef map<string, Autoref<RowType> > RowTypeMap;const RowTypeMap &rowTypes() const;
Get the map of the defined row types. Returns the reference to the Facet's internal map object.
typedef map<string, Autoref<TableType> > TableTypeMap;
const TableTypeMap &tableTypes() const;
Get the map of defined table types. Returns the reference to the Facet's internal map object.
RowType *impRowType(const string &name) const;
Find a single row type by name. If the name is not known, returns NULL.
TableType *impTableType(const string &name) const;
Find a single table type by name. If the name is not known, returns NULL.
Nexus *nexus() const;
Get the nexus of this facet. If the facet is not imported, returns NULL.
int beginIdx() const;
int endIdx() const;
Return the indexes (as in "integer offset") of the _BEGIN_ and _END_ labels in FnReturn.
bool flushWriter();
Flush the collected rowops into the nexus as a single Xtray. If there is no data collected, does nothing. Returns true on a successful flush (even if there was no data collected), false if the Triead was requested to die and thus all the data gets thrown away.
Facet reference, Perl, part 2
As mentioned before, a Facet object is returned from either a nexus creation or nexus import. Then the owner thread can work with it.
$result = $fa->same($fa2);
Check whether two references point to the same Facet object.
$name = $fa->getShortName();
Get the short name of the facet (AKA "as-name", with which it has been imported).
$name = $fa->getFullName();
Get the full name of the nexus represented by this facet. The name consists of two parts separated by a slash, "$thread/$nexus".
$result = $fa->isWriter();
Check whether this is a writer facet (i.e. writes to the nexus). Each facet is either a writer or a reader, so if this method returns 0, it means that this is a reader facet.
$result = $fa->isReverse();
Check whether this facet represents a reverse nexus.
$limit = $fa->queueLimit();
Get the queue size limit of the facet's nexus. I think I forgot to mention it in the Nexus reference, but for a reverse nexus the returned value will be a large integer (currently INT32_MAX but the exact value might change in the future). And if some different limit value was specified during the creation of the reverse nexus, it will be ignored.
$limit = &Triceps::Facet::DEFAULT_QUEUE_LIMIT();
The constant of the default queue size limit that is used for the nexus creation, unless explicitly overridden.
$fret = $fa->getFnReturn();
Get the FnReturn object of this facet. This FnReturn will have the same name as the facet's short name, and it has a special symbiotic relation with the Facet object. Its use depends on whether this is a reader or writer facet. For a writer facet, sending rowops to the labels in FnReturn (directly or by chaining them off the other labels) causes these rowops to be buffered for sending into the nexus. For a reader facet, you can either chain your logic directly off the FnReturn's labels, or push an FnBinding onto it as usual.
$nexus = $fa->nexus();
Get the facet's nexus. There is not a whole lot that can be done with the nexus object, just get the introspection information, and the same information can be obtained directly with the facet's methods.
$idx = $fa->beginIdx();
Index (as in "integer offset", not a table index) of the _BEGIN_ label in the FnReturn's set of labels. There probably isn't much use for this method, and its name is somewhat confusing.
$idx = $fa->endIdx();
Index (as in "integer offset", not a table index) of the _END_ label in the FnReturn's set of labels. There probably isn't much use for this method, and its name is somewhat confusing.
$label = $fa-> getLabel($labelName);
Get a label from FnReturn by name. This is a convenience method, equivalent to $fa->getFnReturn()->getLabel($labelName). Confesses if the label with this name is not found.
@rowTypes = $fa->impRowTypesHash();
Get ("import") the whole set of row types exported through the nexus. The result is an array containing the name-value pairs, values being the imported row types. This array can be assigned into a hash to populate it. As it happens, the pairs will be ordered by name in the lexicographical order but there are no future guarantees about it.
The actual import of the types is done only once, when the nexus is imported to create the facet, and the repeated calls of the imp* methods will return the same objects.
$rt = $fa->impRowType($rtName);
Get ("import") one row type by name. If the name is not known, will confess.
@tableTypes = $fa->impTableTypesHash();
Get ("import") the whole set of table types exported through the nexus. The result is an array containing the name-value pairs, values being the imported table types. This array can be assigned into a hash to populate it. As it happens, the pairs will be ordered by name in the lexicographical order but there are no future guarantees about it.
The actual import of the types is done only once, when the nexus is imported to create the facet, and the repeated calls of the imp* methods will return the same objects.
$tt = $fa->impTableType($ttName);
Get ("import") one table type by name. If the name is not known, will confess.
$result = $fa-> flushWriter();
Flush the collected buffered rowops to the nexus as a single Xtray. If there are no collected rowops, does nothing. Returns 1 if the flush succeeded (even if there was no data to send), 0 if this thread was requested to die and thus all the collected data gets thrown away, same as for the TrieadOwner::flushWriters(). The rules for when this method may be called is also the same: only after calling readyReady(), or it will confess.
If this facet is in an input-only Triead, this call may sleep if a drain is currently active, until the drain is released.
$result = $fa->same($fa2);
Check whether two references point to the same Facet object.
$name = $fa->getShortName();
Get the short name of the facet (AKA "as-name", with which it has been imported).
$name = $fa->getFullName();
Get the full name of the nexus represented by this facet. The name consists of two parts separated by a slash, "$thread/$nexus".
$result = $fa->isWriter();
Check whether this is a writer facet (i.e. writes to the nexus). Each facet is either a writer or a reader, so if this method returns 0, it means that this is a reader facet.
$result = $fa->isReverse();
Check whether this facet represents a reverse nexus.
$limit = $fa->queueLimit();
Get the queue size limit of the facet's nexus. I think I forgot to mention it in the Nexus reference, but for a reverse nexus the returned value will be a large integer (currently INT32_MAX but the exact value might change in the future). And if some different limit value was specified during the creation of the reverse nexus, it will be ignored.
$limit = &Triceps::Facet::DEFAULT_QUEUE_LIMIT();
The constant of the default queue size limit that is used for the nexus creation, unless explicitly overridden.
$fret = $fa->getFnReturn();
Get the FnReturn object of this facet. This FnReturn will have the same name as the facet's short name, and it has a special symbiotic relation with the Facet object. Its use depends on whether this is a reader or writer facet. For a writer facet, sending rowops to the labels in FnReturn (directly or by chaining them off the other labels) causes these rowops to be buffered for sending into the nexus. For a reader facet, you can either chain your logic directly off the FnReturn's labels, or push an FnBinding onto it as usual.
$nexus = $fa->nexus();
Get the facet's nexus. There is not a whole lot that can be done with the nexus object, just get the introspection information, and the same information can be obtained directly with the facet's methods.
$idx = $fa->beginIdx();
Index (as in "integer offset", not a table index) of the _BEGIN_ label in the FnReturn's set of labels. There probably isn't much use for this method, and its name is somewhat confusing.
$idx = $fa->endIdx();
Index (as in "integer offset", not a table index) of the _END_ label in the FnReturn's set of labels. There probably isn't much use for this method, and its name is somewhat confusing.
$label = $fa-> getLabel($labelName);
Get a label from FnReturn by name. This is a convenience method, equivalent to $fa->getFnReturn()->getLabel($labelName). Confesses if the label with this name is not found.
@rowTypes = $fa->impRowTypesHash();
Get ("import") the whole set of row types exported through the nexus. The result is an array containing the name-value pairs, values being the imported row types. This array can be assigned into a hash to populate it. As it happens, the pairs will be ordered by name in the lexicographical order but there are no future guarantees about it.
The actual import of the types is done only once, when the nexus is imported to create the facet, and the repeated calls of the imp* methods will return the same objects.
$rt = $fa->impRowType($rtName);
Get ("import") one row type by name. If the name is not known, will confess.
@tableTypes = $fa->impTableTypesHash();
Get ("import") the whole set of table types exported through the nexus. The result is an array containing the name-value pairs, values being the imported table types. This array can be assigned into a hash to populate it. As it happens, the pairs will be ordered by name in the lexicographical order but there are no future guarantees about it.
The actual import of the types is done only once, when the nexus is imported to create the facet, and the repeated calls of the imp* methods will return the same objects.
$tt = $fa->impTableType($ttName);
Get ("import") one table type by name. If the name is not known, will confess.
$result = $fa-> flushWriter();
Flush the collected buffered rowops to the nexus as a single Xtray. If there are no collected rowops, does nothing. Returns 1 if the flush succeeded (even if there was no data to send), 0 if this thread was requested to die and thus all the collected data gets thrown away, same as for the TrieadOwner::flushWriters(). The rules for when this method may be called is also the same: only after calling readyReady(), or it will confess.
If this facet is in an input-only Triead, this call may sleep if a drain is currently active, until the drain is released.
Saturday, June 29, 2013
Facet reference, Perl, part 1
A Facet represents a Nexus endpoint imported into a Triead. A facet is either a reader (reading from the nexus) or a writer (writing into the nexus).
In the Perl API the Facets are created by TrieadOwner::makeNexus() or TrieadOwner::importNexus(). After that the metadata in the Facet is fixed and is available for reading only. Of course, the rowops can then be read or written.
The reading of data from a Facet is done by TrieadOwner::nextXtray(). There is no way to read a tray from a particular facet, nextXtray() reads from all the Triead's imported reader facets, alternating in a fair fashion if more than one of them has data available.
Each Facet has an FnReturn connected to it. The reading from a reader facet happens by forwarding the incoming rowops to that FnReturn. To actually process the data, you can either chain your handler labels directly to the FnReturn labels, or push an FnBinding onto that FnReturn. An incoming Xtray is always processed as a unit, with no intermixing with the other Xtrays.
The writing to a writer facet happens by calling the labels of its FnReturn. Which then has the logic that collects all these rowops into a buffer. Then when the facet is flushed, that buffer becomes an indivisible Xtray that gets sent to the nexus as a unit, and then read by the reader facets as a unit.
The facet metadata consists of:
There are two special labels, named _BEGIN_ and _END_. They may be defined explicitly, but if they aren't, they will be always added implicitly, with an empty row type (i.e. a row type with no fields).
When reading an Xtray, the _BEGIN_ label will always be called first, and _END_ last, thus framing the rest of the data. There are optimizations that skip the calling if there is nothing chained to these labels in FnReturn nor to the top FnBinding, and the rowop as such carries no extra data. The optimization is actually a bit deeper: the _BEGIN_ and _END_ rowops that have no extra data in them aren't even carried in the Xtray through the nexus. They are generated on the fly if there is an interest in them, or otherwise the generation is skipped.
What is meant by the "extra data"? It's either the opcode is not OP_INSERT or there are some non-NULL fields (or both). If the _BEGIN_ and _END_ labels were auto-generated, their row type will contain no fields, so the only way to sent the non-default data in them will be the non-default opcode. But if you define them explicitly with a different row type, you can also send the data in them.
When sending the data into a writer Facet, you don't have to send the _BEGIN_ and _END_ rowops, if you don't, they will be generated automatically as needed, with the default contents (opcode OP_INSERT and NULLs in all the fields). Moreover, they will really be generated automatically on the reader side, thus saving the overhead of passing them through the nexus. Another consequence of this optimization is that it's impossible to create an Xtray consisting of only a default _BEGIN_, a default _END_ and no payload rowops between them. It would be an empty Xtray, that would never be sent through the nexus. Even if you create these _BEGIN_ and _END_ rowops manually (but with the default contents), they will be thrown away when they reach the writer facet. If you want an Xtray to get through, you've got to either send the payload or put something non-default into at least one of the _BEGIN_ or _END_ rowops, at the very minimum a different opcode.
Sending the _BEGIN_ and _END_ rowops into a writer facet also has the effect of flushing it. Even if these rowops have the default contents and become thrown away by the facet, the flushing effect still works. The _BEGIN_ rowop flushes any data that has been collected in the buffer before it. The _END_ rowop gets added to the buffer (or might get thrown away) and then flushes the buffer. If the buffer happens to contain anything at the flush time, that contents forms an Xtray and gets forwarded to the nexus.
It's a long and winding explanation, but really it just does what is intuitively expected.
A Facet has two names, the "full" one and the "short" one:
In the Perl API the Facets are created by TrieadOwner::makeNexus() or TrieadOwner::importNexus(). After that the metadata in the Facet is fixed and is available for reading only. Of course, the rowops can then be read or written.
The reading of data from a Facet is done by TrieadOwner::nextXtray(). There is no way to read a tray from a particular facet, nextXtray() reads from all the Triead's imported reader facets, alternating in a fair fashion if more than one of them has data available.
Each Facet has an FnReturn connected to it. The reading from a reader facet happens by forwarding the incoming rowops to that FnReturn. To actually process the data, you can either chain your handler labels directly to the FnReturn labels, or push an FnBinding onto that FnReturn. An incoming Xtray is always processed as a unit, with no intermixing with the other Xtrays.
The writing to a writer facet happens by calling the labels of its FnReturn. Which then has the logic that collects all these rowops into a buffer. Then when the facet is flushed, that buffer becomes an indivisible Xtray that gets sent to the nexus as a unit, and then read by the reader facets as a unit.
The facet metadata consists of:
- a set of labels, same as for an FnReturn, used to build the facet's internal FnReturn; these labels define the data that can be carried through the nexus;
- a set of row types that gets exported through the nexus;
- a set of table types that gets exported through the nexus.
There are two special labels, named _BEGIN_ and _END_. They may be defined explicitly, but if they aren't, they will be always added implicitly, with an empty row type (i.e. a row type with no fields).
When reading an Xtray, the _BEGIN_ label will always be called first, and _END_ last, thus framing the rest of the data. There are optimizations that skip the calling if there is nothing chained to these labels in FnReturn nor to the top FnBinding, and the rowop as such carries no extra data. The optimization is actually a bit deeper: the _BEGIN_ and _END_ rowops that have no extra data in them aren't even carried in the Xtray through the nexus. They are generated on the fly if there is an interest in them, or otherwise the generation is skipped.
What is meant by the "extra data"? It's either the opcode is not OP_INSERT or there are some non-NULL fields (or both). If the _BEGIN_ and _END_ labels were auto-generated, their row type will contain no fields, so the only way to sent the non-default data in them will be the non-default opcode. But if you define them explicitly with a different row type, you can also send the data in them.
When sending the data into a writer Facet, you don't have to send the _BEGIN_ and _END_ rowops, if you don't, they will be generated automatically as needed, with the default contents (opcode OP_INSERT and NULLs in all the fields). Moreover, they will really be generated automatically on the reader side, thus saving the overhead of passing them through the nexus. Another consequence of this optimization is that it's impossible to create an Xtray consisting of only a default _BEGIN_, a default _END_ and no payload rowops between them. It would be an empty Xtray, that would never be sent through the nexus. Even if you create these _BEGIN_ and _END_ rowops manually (but with the default contents), they will be thrown away when they reach the writer facet. If you want an Xtray to get through, you've got to either send the payload or put something non-default into at least one of the _BEGIN_ or _END_ rowops, at the very minimum a different opcode.
Sending the _BEGIN_ and _END_ rowops into a writer facet also has the effect of flushing it. Even if these rowops have the default contents and become thrown away by the facet, the flushing effect still works. The _BEGIN_ rowop flushes any data that has been collected in the buffer before it. The _END_ rowop gets added to the buffer (or might get thrown away) and then flushes the buffer. If the buffer happens to contain anything at the flush time, that contents forms an Xtray and gets forwarded to the nexus.
It's a long and winding explanation, but really it just does what is intuitively expected.
A Facet has two names, the "full" one and the "short" one:
- The full name is copied from the nexus and consists of the name of the thread that exported the nexus and the name of the nexus itself separated by a slash, such as "t1/nx1".
- The short name is the name with which the facet was imported. By default it's taken from the short name of the nexus. But it can also be given a different explicit name during the import, which is known as the "as-name" (because it's similar to the SQL "AS" clause). So if the full name is "t1/nx1", the default short name will be "nx1", but it can be overridden. The facet's FnReturn is named with the facet's short name.
Friday, June 28, 2013
Nexus reference, Perl and C++
The Nexus class is pretty much opaque. It's created and managed entirely inside the App infrastructure, and even the public API for importing a nexus doesn't deal with the Nexus object itself, but only with its name. The only public use of the Nexus object is for the introspection and entertainment value, to see what Trieads export and import what Nexuses: pretty much the only way to get a Nexus reference is by listing the exports or imports of a Triead.
The Perl API of a Nexus is very limited and is very much an afterthought:
$nxname = $nx->getName();
Get the name of the nexus (the short name, inside the thread).
$tname = $nx->getTrieadName();
Get the name of the Triead that exported this nexus.
$result = $nx->same($nx2);
Check whether two Nexus references point to the same object.
$result = $nx->isReverse();
Check whether the Nexus is reverse.
$limit = $nx->queueLimit();
Get the queue limit of the nexus.
In C++ the corresponding methods are:
const string &getName() const;
const string &getTrieadName() const;
bool isReverse() const;
int queueLimit() const;
In C++ the Nexus object is an Mtarget, and safe to access from multiple threads. It's defined in app/Nexus.h.
The Perl API of a Nexus is very limited and is very much an afterthought:
$nxname = $nx->getName();
Get the name of the nexus (the short name, inside the thread).
$tname = $nx->getTrieadName();
Get the name of the Triead that exported this nexus.
$result = $nx->same($nx2);
Check whether two Nexus references point to the same object.
$result = $nx->isReverse();
Check whether the Nexus is reverse.
$limit = $nx->queueLimit();
Get the queue limit of the nexus.
In C++ the corresponding methods are:
const string &getName() const;
const string &getTrieadName() const;
bool isReverse() const;
int queueLimit() const;
In C++ the Nexus object is an Mtarget, and safe to access from multiple threads. It's defined in app/Nexus.h.
TrieadOwner reference, C++
The TrieadOwner is defined in app/TrieadOwner.h. Its constructor is protected, the normal way of constructing is by calling App::makeTriead(). This is also called "defining a Triead".
TrieadOwner is an Starget, and must be accessed only from the thread that owns it (though it's possible to create it in the parent thread and then pass to the actual owner thread, as long as the synchronization between the threads is done properly).
Triead *get() const;
Get the public side if the Triead. In C++, unlike Perl, the Triead methods are not duplicated in TrieadOwner. So they are accessed through get(), and for example to get the Triead name, call to->get()->getName(). The TrieadOwner holds a reference to Triead, so the Triead object will never get destroyed as long as the TrieadOwner is alive.
App *app() const;
Get the App where this Triead belongs. The TrieadOwner holds a reference to App, so the App object will never get destroyed as long as the TrieadOwner is alive.
bool isRqDead() const;
Check whether this triead has been requested to die.
void requestMyselfDead();
Request this thread itself to die (see the reasoning for the in the Perl reference).
Unit *unit() const;
Get the main unit of this Triead. It has the same name as the Triead itself.
void addUnit(Autoref<Unit> u);
Keep track of an additional unit. Adding a unit multiple times has no effect. See the other implications in the Perl reference.
bool forgetUnit(Unit *u);
Forget about an additional unit. If the unit is already unknown, has no effect. The main unit can not be forgotten.
typedef list<Autoref<Unit> > UnitList;
const UnitList &listUnits() const;
List the tracked units. The main unit is always included at the front of the list.
void markConstructed();
Advance the Triead state to Constructed. Repeated calls have no effect.
void markReady();
Advance the Triead state to Ready. Repeated calls have no effect. The advancement is cumulative: if the Triead was not constructed yet, it will be automatically advanced first to Constructed and then to Ready. If this is the last Triead to become ready, it will trigger the App topology check, and if the check fails, abort the App and throw an Exception.
void readyReady();
Advance the Triead to the Ready state, and wait for all the Trieads to become ready. The topology check applies in the same way as in markReady().
void markDead();
Advance the Triead to the Dead state. If this Triead was not Ready yet and it's the last one to become so, the topology check will run. If the topology check fails, the App will be aborted but markDead() will not throw an exception.
This method is automatically called from the TrieadOwner destructor, so most of the time there is no need to call it explicitly.
It also clears all the tracked units.
void abort(const string &msg) const;
Abort the App. The name of this thread will be forwarded to the App along with the error message. The error message may be multi-line.
Onceref<Triead> findTriead(const string &tname, bool immed = false);
Find a Triead in the App by name. The flag immed controls whether this method may wait. If immed is false, and the target thread is not constructed yet (but at least declared), the method will sleep until it becomes constructed, and then returns it. If the target thread is not declared, it will throw an Exception. If immed is true, the look-up is immediate: it will return the thread even if it's not constructed but is at least defined. If it's not defined (even if it's declared), an Exception will be thrown. The look-up of this Triead itself is always immediate, irrespective of the immed flag.
An Exception may also be thrown if a circular sequence of Trieads deadlocks waiting for each other.
Onceref<Facet> exportNexus(Autoref<Facet> facet, bool import = true);
Export a Nexus. The Nexus definition is constructed as a Facet object, which is then used by this method to construct and export the Nexus. The same argument Facet reference is then returned back as the result. The import flag tells, whether the Nexus is to be also imported back by connecting the same original Facet object to it. If import is false, the original Facet reference is still returned back but it can't be used for anything, and can only be thrown away. The direction of the import (reading or writing) is defined in the Facet, as well as the nexus name and all the other information.
Throws an Exception on any errors, in particular on the duplicate facet names within the Triead.
Onceref<Facet> exportNexusNoImport(Autoref<Facet> facet);
A convenience wrapper around exportNexus() with import=false.
Onceref<Facet> importNexus(const string &tname, const string &nexname, const string &asname, bool writer, bool immed = false);
Import a Nexus from another Triead. tname is the name of the other thread, nexname is the name of nexus exported from it, asname is the local name for the imported facet ("" means "same as nexname"), the writer flag determines if the import is for writing, and the immed flag has the same meaning as in findTriead(). The import of a nexus involves finding its exporting thread, and the immed flag controls, how this finding is done.
Throws an Exception if anything is not found, or the local import name conflicts with another imported facet.
Onceref<Facet> importNexusImmed(const string &tname, const string &nexname, const string &asname, bool writer);
Onceref<Facet> importReader(const string &tname, const string &nexname, const string &asname = "", bool immed=false);
Onceref<Facet> importWriter(const string &tname, const string &nexname, const string &asname = "", bool immed=false);
Onceref<Facet> importReaderImmed(const string &tname, const string &nexname, const string &asname = "");
Onceref<Facet> importWriterImmed(const string &tname, const string &nexname, const string &asname = "");
Convenience wrappers for importNexus(), providing the default arguments and the more mnemonic names.
typedef map<string, Autoref<Nexus> > NexusMap;
void exports(NexusMap &ret) const;
Get the nexuses exported here. The map argument will be cleared and refilled with the new values.
typedef map<string, Autoref<Facet> > FacetMap;
void imports(FacetMap &ret) const;
Get the facets imported here. The map argument will be cleared and refilled with the new values.
NexusMaker *makeNexusReader(const string &name);
NexusMaker *makeNexusWriter(const string &name);
NexusMaker *makeNexusNoImport(const string &name);
A convenient way to build the nexuses for export in a chained fashion. The name argument is the nexus name. The NexusMaker is an opaque class that has the same building methods as a Facet, plus the method complete() that finishes the export. This call sequence is more convenient than building a Facet and then exporting it. For example:
Autoref<Facet> myfacet = ow->makeNexusReader("my")
->addLabel("one", rt1)
->addFromLabel("two", lb2)
->setContext(new MyFnContext)
->setReverse()
->complete();
Only one nexus may be built like this at a time, since there is only one instance of NexusMaker per TrieadOwner that gets reused over and over. It keeps the Facet instance being built in it. If you don't complete the build, that Facet instance will be left sitting around until another makeNexus*() calls when it will get thrown away. But in general if you do the calling in the sequence as shown, you can't forget to call complete() at the end, since otherwise the return type would not match and the compiler will fail.
bool flushWriters();
Flush all the writer facets. Returns true on the successfull completion, false if the thread was requested to die, and thus all the output was thrown away.
bool nextXtray(bool wait = true, const struct timespec &abstime = *(const struct timespec *)NULL);
Read and process the next Xtray. Automatically calls the flushWriters() after processing. The rest works in the same way as described in Perl, however this is a combination of Perl's nextXtray(), nextXtrayNoWait() and nextXtrayTimeLimit(). If wait is false, this method will never wait. If wait is true and the abstime reference is not NULL, it might wait but not past that absolute time. Otherwise it will wait until the data becomes available of the thread is requested to die.
Returns true if an Xtray has been processed, false if it wasn't (for any reason, a timeout expiring or thread being requested to die).
bool nextXtrayNoWait();
A convenience wrapper over nextXtray(false).
bool nextXtrayTimeout(int64_t sec, int32_t nsec);
Another convenience wrapper over nextXtray(): read and process an Xtray, with a timeout limit. The timeout value consists of the seconds and nanoseconds parts.
void mainLoop();
Run the main loop, calling nextXtray() repeatedly until the thread is requested to die.
bool isRqDrain();
Check if a drain is currently requested by any thread (and applies to this thread). In case if an exclusive drain is requested with the exclusion of this thread, this method will return false.
void requestDrainShared();
void requestDrainExclusive();
void waitDrain();
bool isDrained();
void drainShared();
void drainExclusive();
void undrain();
The drain control, same as in Perl. These methods are really wrappers over the corresponding App methods. And generally a better idea is to do the scoped drains with AutoDrain rather than to call these methods directly.
The C++ API provides no methods for the file descriptor tracking as such. Instead these methods are implemented in the class FileInterrupt. TrieadOwner has a public field
Autoref<FileInterrupt> fileInterrupt_;
to keep an instance of the interruptor. TrieadOwner itself has no use for it, nor does any other part of Triceps, it's just a location to keep this reference for the convenience of the application developer. The Perl API makes use of this location.
But how does the thread then get interrupted when it's requested to die? The answer is that the TrieadJoin object also has a reference to the FileInterrupt object, and even creates that object in its own constructor. So when a joiner method is defined for a thread, that supplies the App with the access to its interruptor as well. And to put the file descriptors into the FileInterrupt object, you can either keep a direct reference to it somewhere in your code, or copy that reference into the TrieadOwner object.
Here is for example how the Perl thread joiner is created for the TrieadOwner inside the Perl implementation:
string tn(tname);
Autoref<TrieadOwner> to = appv->makeTriead(tn, fragname);
PerlTrieadJoin *tj = new PerlTrieadJoin(appv->getName(), tname,
SvIOK(tid)? SvIV(tid): -1,
SvIOK(handle)? SvIV(handle): 0,
testfail);
to->fileInterrupt_ = tj->fileInterrupt();
appv->defineJoin(tn, tj);
This is somewhat cumbersome, but the native C++ programs can create their threads using the class BasicPthread that takes care of this and more.
TrieadOwner is an Starget, and must be accessed only from the thread that owns it (though it's possible to create it in the parent thread and then pass to the actual owner thread, as long as the synchronization between the threads is done properly).
Triead *get() const;
Get the public side if the Triead. In C++, unlike Perl, the Triead methods are not duplicated in TrieadOwner. So they are accessed through get(), and for example to get the Triead name, call to->get()->getName(). The TrieadOwner holds a reference to Triead, so the Triead object will never get destroyed as long as the TrieadOwner is alive.
App *app() const;
Get the App where this Triead belongs. The TrieadOwner holds a reference to App, so the App object will never get destroyed as long as the TrieadOwner is alive.
bool isRqDead() const;
Check whether this triead has been requested to die.
void requestMyselfDead();
Request this thread itself to die (see the reasoning for the in the Perl reference).
Unit *unit() const;
Get the main unit of this Triead. It has the same name as the Triead itself.
void addUnit(Autoref<Unit> u);
Keep track of an additional unit. Adding a unit multiple times has no effect. See the other implications in the Perl reference.
bool forgetUnit(Unit *u);
Forget about an additional unit. If the unit is already unknown, has no effect. The main unit can not be forgotten.
typedef list<Autoref<Unit> > UnitList;
const UnitList &listUnits() const;
List the tracked units. The main unit is always included at the front of the list.
void markConstructed();
Advance the Triead state to Constructed. Repeated calls have no effect.
void markReady();
Advance the Triead state to Ready. Repeated calls have no effect. The advancement is cumulative: if the Triead was not constructed yet, it will be automatically advanced first to Constructed and then to Ready. If this is the last Triead to become ready, it will trigger the App topology check, and if the check fails, abort the App and throw an Exception.
void readyReady();
Advance the Triead to the Ready state, and wait for all the Trieads to become ready. The topology check applies in the same way as in markReady().
void markDead();
Advance the Triead to the Dead state. If this Triead was not Ready yet and it's the last one to become so, the topology check will run. If the topology check fails, the App will be aborted but markDead() will not throw an exception.
This method is automatically called from the TrieadOwner destructor, so most of the time there is no need to call it explicitly.
It also clears all the tracked units.
void abort(const string &msg) const;
Abort the App. The name of this thread will be forwarded to the App along with the error message. The error message may be multi-line.
Onceref<Triead> findTriead(const string &tname, bool immed = false);
Find a Triead in the App by name. The flag immed controls whether this method may wait. If immed is false, and the target thread is not constructed yet (but at least declared), the method will sleep until it becomes constructed, and then returns it. If the target thread is not declared, it will throw an Exception. If immed is true, the look-up is immediate: it will return the thread even if it's not constructed but is at least defined. If it's not defined (even if it's declared), an Exception will be thrown. The look-up of this Triead itself is always immediate, irrespective of the immed flag.
An Exception may also be thrown if a circular sequence of Trieads deadlocks waiting for each other.
Onceref<Facet> exportNexus(Autoref<Facet> facet, bool import = true);
Export a Nexus. The Nexus definition is constructed as a Facet object, which is then used by this method to construct and export the Nexus. The same argument Facet reference is then returned back as the result. The import flag tells, whether the Nexus is to be also imported back by connecting the same original Facet object to it. If import is false, the original Facet reference is still returned back but it can't be used for anything, and can only be thrown away. The direction of the import (reading or writing) is defined in the Facet, as well as the nexus name and all the other information.
Throws an Exception on any errors, in particular on the duplicate facet names within the Triead.
Onceref<Facet> exportNexusNoImport(Autoref<Facet> facet);
A convenience wrapper around exportNexus() with import=false.
Onceref<Facet> importNexus(const string &tname, const string &nexname, const string &asname, bool writer, bool immed = false);
Import a Nexus from another Triead. tname is the name of the other thread, nexname is the name of nexus exported from it, asname is the local name for the imported facet ("" means "same as nexname"), the writer flag determines if the import is for writing, and the immed flag has the same meaning as in findTriead(). The import of a nexus involves finding its exporting thread, and the immed flag controls, how this finding is done.
Throws an Exception if anything is not found, or the local import name conflicts with another imported facet.
Onceref<Facet> importNexusImmed(const string &tname, const string &nexname, const string &asname, bool writer);
Onceref<Facet> importReader(const string &tname, const string &nexname, const string &asname = "", bool immed=false);
Onceref<Facet> importWriter(const string &tname, const string &nexname, const string &asname = "", bool immed=false);
Onceref<Facet> importReaderImmed(const string &tname, const string &nexname, const string &asname = "");
Onceref<Facet> importWriterImmed(const string &tname, const string &nexname, const string &asname = "");
Convenience wrappers for importNexus(), providing the default arguments and the more mnemonic names.
typedef map<string, Autoref<Nexus> > NexusMap;
void exports(NexusMap &ret) const;
Get the nexuses exported here. The map argument will be cleared and refilled with the new values.
typedef map<string, Autoref<Facet> > FacetMap;
void imports(FacetMap &ret) const;
Get the facets imported here. The map argument will be cleared and refilled with the new values.
NexusMaker *makeNexusReader(const string &name);
NexusMaker *makeNexusWriter(const string &name);
NexusMaker *makeNexusNoImport(const string &name);
A convenient way to build the nexuses for export in a chained fashion. The name argument is the nexus name. The NexusMaker is an opaque class that has the same building methods as a Facet, plus the method complete() that finishes the export. This call sequence is more convenient than building a Facet and then exporting it. For example:
Autoref<Facet> myfacet = ow->makeNexusReader("my")
->addLabel("one", rt1)
->addFromLabel("two", lb2)
->setContext(new MyFnContext)
->setReverse()
->complete();
Only one nexus may be built like this at a time, since there is only one instance of NexusMaker per TrieadOwner that gets reused over and over. It keeps the Facet instance being built in it. If you don't complete the build, that Facet instance will be left sitting around until another makeNexus*() calls when it will get thrown away. But in general if you do the calling in the sequence as shown, you can't forget to call complete() at the end, since otherwise the return type would not match and the compiler will fail.
bool flushWriters();
Flush all the writer facets. Returns true on the successfull completion, false if the thread was requested to die, and thus all the output was thrown away.
bool nextXtray(bool wait = true, const struct timespec &abstime = *(const struct timespec *)NULL);
Read and process the next Xtray. Automatically calls the flushWriters() after processing. The rest works in the same way as described in Perl, however this is a combination of Perl's nextXtray(), nextXtrayNoWait() and nextXtrayTimeLimit(). If wait is false, this method will never wait. If wait is true and the abstime reference is not NULL, it might wait but not past that absolute time. Otherwise it will wait until the data becomes available of the thread is requested to die.
Returns true if an Xtray has been processed, false if it wasn't (for any reason, a timeout expiring or thread being requested to die).
bool nextXtrayNoWait();
A convenience wrapper over nextXtray(false).
bool nextXtrayTimeout(int64_t sec, int32_t nsec);
Another convenience wrapper over nextXtray(): read and process an Xtray, with a timeout limit. The timeout value consists of the seconds and nanoseconds parts.
void mainLoop();
Run the main loop, calling nextXtray() repeatedly until the thread is requested to die.
bool isRqDrain();
Check if a drain is currently requested by any thread (and applies to this thread). In case if an exclusive drain is requested with the exclusion of this thread, this method will return false.
void requestDrainShared();
void requestDrainExclusive();
void waitDrain();
bool isDrained();
void drainShared();
void drainExclusive();
void undrain();
The drain control, same as in Perl. These methods are really wrappers over the corresponding App methods. And generally a better idea is to do the scoped drains with AutoDrain rather than to call these methods directly.
The C++ API provides no methods for the file descriptor tracking as such. Instead these methods are implemented in the class FileInterrupt. TrieadOwner has a public field
Autoref<FileInterrupt> fileInterrupt_;
to keep an instance of the interruptor. TrieadOwner itself has no use for it, nor does any other part of Triceps, it's just a location to keep this reference for the convenience of the application developer. The Perl API makes use of this location.
But how does the thread then get interrupted when it's requested to die? The answer is that the TrieadJoin object also has a reference to the FileInterrupt object, and even creates that object in its own constructor. So when a joiner method is defined for a thread, that supplies the App with the access to its interruptor as well. And to put the file descriptors into the FileInterrupt object, you can either keep a direct reference to it somewhere in your code, or copy that reference into the TrieadOwner object.
Here is for example how the Perl thread joiner is created for the TrieadOwner inside the Perl implementation:
string tn(tname);
Autoref<TrieadOwner> to = appv->makeTriead(tn, fragname);
PerlTrieadJoin *tj = new PerlTrieadJoin(appv->getName(), tname,
SvIOK(tid)? SvIV(tid): -1,
SvIOK(handle)? SvIV(handle): 0,
testfail);
to->fileInterrupt_ = tj->fileInterrupt();
appv->defineJoin(tn, tj);
This is somewhat cumbersome, but the native C++ programs can create their threads using the class BasicPthread that takes care of this and more.
TrieadOwner and TrackedFile reference, Perl, part 4
The file interruption part of the API deals with how the thread handles a request to die. It should stop its work and exit, and for the normal threads that read from nexuses, this is fairly straightforward. The difficulty is with the threads that read from the outside sources (sockets and such). They may be in the middle of a read call, with no way to tell when the next chunk of data will arrive. These long calls on the file descriptors need to be interrupted when the thread is requested to die.
The interruption is done by revoking the file descriptors (dupping a /dev/null into it) and sending the signal SIGUSR2 to the thread. Even if the dupping doesn't interrupt the file operation, SIGUSR2 does, and on restart it will find that the file descriptor now returns to /dev/null and return immediately. Triceps defines a SIGUSR2 handler that does nothing, but you can override it with a custom one.
For this to work, Triceps needs to know, which file descriptors are to be revoked, which is achieved by registering for tracking them in the TrieadOwner. To avoid revoking the unrelated descriptors, they must be unregistered before closing. The normal sequence is:
The lowest-level calls deal with the raw tracking:
$to->trackFd($fd);
Register the file descriptor (obtained with a fileno() or such) for tracking. The repeated calls for the same descriptor have no effect.
$to->forgetFd($fd);
Unregister the file descriptor. If the descriptor is not registered, the call is ignored.
The next level of the API deals with the file handles, extracting the file descriptors from them as needed.
$to->track(FILE);
Get a file descriptor from the file handle and track it.
$to->forget(FILE);
Get a file descriptor from the file handle and unregister it.
$to->close(FILE);
Unegister the file handle's descriptor then close the file handle. It's a convenience wrapper, to make the unregistering easier to remember.
The correct sequence might be hard to follow if the code involves some dying and evals catching these deaths. To handle that, the next level of the API provides the automatic tracking by scope. The scoping is done with the class Triceps::TrackedFile.Which is probably easier to describe right here.
A TrackedFile object keeps a reference of a file handle and also knows of its file descriptor being tracked by the TrieadOwner (so yes, it has a reference to the TrieadOwner too). Until the TrackedFile is destroyed, the file handle in it will never have its reference count go to 0, so it will never be automatically closed and destroyed. And when the TrackedFile is destroyed, first it tells the TrieadOwner to forget the file descriptor and only then unreferences the file handle, preserving the correct sequence.
And the scope of the TrackedFile is controlled by the scope of a variable that keeps a reference to it. If it's a local/my variable, TrackedFile will be destroyed on leaving the block, if it's a field in an object, it will be destroyed when the object is destroyed or the field is reset. Basically, the usual Perl scope business in Triceps.
If you want to close the file handle before leaving the scope of TrackedFile, don't call close() on the file handle. Instead, call close on the TrackedFile:
$trf->close();
This will again properly untrack the file descriptor, and then close the file handle, and remember that it has been closed, so no seconds attempt at that will be done when the TrackedFile gets destroyed.
There also are a couple of getter methods on the TrackedFile:
$fd = $trf->fd();
Get the tracked file descriptor. If the file handle has been already closed, will return -1.
$fh = $trf->get();
Get the tracked file handle. If the file handle had already been closed, will confess.
Now we get back to the TrieadOwner.
$trf = $to->makeTrackedFile(FILE);
Create a TrackedFile object from a file handle.
$trf = $to->makeTrackedFileFd(FILE, $fd);
The more low-level way to construct a TrackedFile, with specifying the file descriptor as a separate explicit argument. makeTrackedFile() is pretty much a wrapper that calls fileno() on the file handle and then calls makeTrackedFileFd().
And the following methods combine the loading of a file descriptor from the App and tracking it by the TrieadOwner. They are the most typically used interface for passing around the file handles through the App.
($trf, $file) = $to->trackDupFile($name, $mode);
Load the dupped file handle from the App, create an IO::Handle file object from it according to the $mode (in either r/w/a/r+/w+/a+ or </>/>>/+</+>/+>> format), and make a TrackedFile from it. Returns a pair of the TrackedFile object and the created file handle. The App still keeps the original file descriptor. The $mode must be consistent with the original mode of the file stored into the App. $name is the name used to store the file descriptor into the App.
($trf, $file) = $to->trackGetFile($name, $mode);
Similar to trackDupFile() only the file descriptor is moved from the App, and the App forgets about it.
($trf, $file) = $to->trackDupSocket($name, $mode);
($trf, $file) = $to->trackGetSocket($name, $mode);
Similar to the File versions, only create a file handle object of the class IO::Socket::INET.
($trf, $file) = $to->trackDupClass($name, $mode, $class);
($trf, $file) = $to->trackGetClass($name, $mode, $class);
Similar to the File versions, only creates the file handle of an arbitrary subclass of IO::Handle (with the name specified in $class). So for example if you ever want to load an IPv6 or Unix domain socket, these are the methods to use.
The interruption is done by revoking the file descriptors (dupping a /dev/null into it) and sending the signal SIGUSR2 to the thread. Even if the dupping doesn't interrupt the file operation, SIGUSR2 does, and on restart it will find that the file descriptor now returns to /dev/null and return immediately. Triceps defines a SIGUSR2 handler that does nothing, but you can override it with a custom one.
For this to work, Triceps needs to know, which file descriptors are to be revoked, which is achieved by registering for tracking them in the TrieadOwner. To avoid revoking the unrelated descriptors, they must be unregistered before closing. The normal sequence is:
- open a file descriptor
- register it for tracking
- do the file operations
- unregister the file descriptor
- close it
The lowest-level calls deal with the raw tracking:
$to->trackFd($fd);
Register the file descriptor (obtained with a fileno() or such) for tracking. The repeated calls for the same descriptor have no effect.
$to->forgetFd($fd);
Unregister the file descriptor. If the descriptor is not registered, the call is ignored.
The next level of the API deals with the file handles, extracting the file descriptors from them as needed.
$to->track(FILE);
Get a file descriptor from the file handle and track it.
$to->forget(FILE);
Get a file descriptor from the file handle and unregister it.
$to->close(FILE);
Unegister the file handle's descriptor then close the file handle. It's a convenience wrapper, to make the unregistering easier to remember.
The correct sequence might be hard to follow if the code involves some dying and evals catching these deaths. To handle that, the next level of the API provides the automatic tracking by scope. The scoping is done with the class Triceps::TrackedFile.Which is probably easier to describe right here.
A TrackedFile object keeps a reference of a file handle and also knows of its file descriptor being tracked by the TrieadOwner (so yes, it has a reference to the TrieadOwner too). Until the TrackedFile is destroyed, the file handle in it will never have its reference count go to 0, so it will never be automatically closed and destroyed. And when the TrackedFile is destroyed, first it tells the TrieadOwner to forget the file descriptor and only then unreferences the file handle, preserving the correct sequence.
And the scope of the TrackedFile is controlled by the scope of a variable that keeps a reference to it. If it's a local/my variable, TrackedFile will be destroyed on leaving the block, if it's a field in an object, it will be destroyed when the object is destroyed or the field is reset. Basically, the usual Perl scope business in Triceps.
If you want to close the file handle before leaving the scope of TrackedFile, don't call close() on the file handle. Instead, call close on the TrackedFile:
$trf->close();
This will again properly untrack the file descriptor, and then close the file handle, and remember that it has been closed, so no seconds attempt at that will be done when the TrackedFile gets destroyed.
There also are a couple of getter methods on the TrackedFile:
$fd = $trf->fd();
Get the tracked file descriptor. If the file handle has been already closed, will return -1.
$fh = $trf->get();
Get the tracked file handle. If the file handle had already been closed, will confess.
Now we get back to the TrieadOwner.
$trf = $to->makeTrackedFile(FILE);
Create a TrackedFile object from a file handle.
$trf = $to->makeTrackedFileFd(FILE, $fd);
The more low-level way to construct a TrackedFile, with specifying the file descriptor as a separate explicit argument. makeTrackedFile() is pretty much a wrapper that calls fileno() on the file handle and then calls makeTrackedFileFd().
And the following methods combine the loading of a file descriptor from the App and tracking it by the TrieadOwner. They are the most typically used interface for passing around the file handles through the App.
($trf, $file) = $to->trackDupFile($name, $mode);
Load the dupped file handle from the App, create an IO::Handle file object from it according to the $mode (in either r/w/a/r+/w+/a+ or </>/>>/+</+>/+>> format), and make a TrackedFile from it. Returns a pair of the TrackedFile object and the created file handle. The App still keeps the original file descriptor. The $mode must be consistent with the original mode of the file stored into the App. $name is the name used to store the file descriptor into the App.
($trf, $file) = $to->trackGetFile($name, $mode);
Similar to trackDupFile() only the file descriptor is moved from the App, and the App forgets about it.
($trf, $file) = $to->trackDupSocket($name, $mode);
($trf, $file) = $to->trackGetSocket($name, $mode);
Similar to the File versions, only create a file handle object of the class IO::Socket::INET.
($trf, $file) = $to->trackDupClass($name, $mode, $class);
($trf, $file) = $to->trackGetClass($name, $mode, $class);
Similar to the File versions, only creates the file handle of an arbitrary subclass of IO::Handle (with the name specified in $class). So for example if you ever want to load an IPv6 or Unix domain socket, these are the methods to use.
TrieadOwner reference, Perl, part 3
@exports = $to->exports();
The same as the method on the Triead, equivalent to $to->get()->exports(). The returned array contains the name-value pairs of the nexus names and objects.
@imports = $to->imports();
This method is different from the Triead method. It still returns an array of name-value pairs but the values are Facets (not Nexuses, as in Triead). It's a natural difference, since the facets are useful in the owner thread, and available only in it.
$result = $to->flushWriters();
Flush any data collected in the writer facets, sending them to the appropriate nexuses. The data in each facet becomes a tray that is sent to the nexus (if there was no data collected on a facet, nothing will be sent from it). Returns 1 if the flush was completed, 0 if the thread was requested to die and this the data was discarded. The data is never sent out of a facet by itself, it always must be flushed in one of the explicit ways (TrieadOwner::flushWriters(), Facet::flushWriter(), or enqueueing a rowop on the facet's labels _BEGIN_ and _END_). The flush may get stuck if this is an input-only thread and a drain is active, it will wait until the drain is released.
$to->requestMyselfDead();
Request this Triead itself to die. This is the way to disconnect from the nexuses while the thread is exiting on its own. For example, if this thread going to dump its data before exit to a large file that takes half an hour to write, normally the data queued for this thread might fill up the queues in the nexuses, and it's a bad practice to keep the other threads stuck due to the overflowing buffers. Requesting this thread to die disconnects it from the nexuses and prevents the data from collecting.The thread could also be disconnected by marking it dead, but it will keep the harvester stuck waiting to join it while the thread completes its long write, and that's not so good either. So this call provides the solution, avoiding both pitfalls.
$result = $to->nextXtray();
Process one incoming tray from a single reader nexus (any nexus where data is available). A tray essentially embodies a transaction, and "X" stands for "cross-thread". There actually is the Xtray type that represents the Tray in a special thread-safe format but it's used only inside the nexuses and not visible from outside.
If there is currently no data to process, this method will wait.
The return value is 1 normally, or 0 if the thread was requested to die. So the typical usage is:
while($to->nextXtray()) { ... }
The method mainLoop() encapsulates the most typical usage, and nextXtray() needs to be used directly only in the more unusual circumstances.
The data is read from the reverse nexuses first, at a higher priority. If any reverse nexus has data available, it will always be read before the direct nexuses. A reverse nexus typically completes a topological loop, so this priority creates the preference to cycle the data through the loop until it comes out, before accepting more data into the loop. Since all the nexuses have non-zero-length queues, obviously, there will be multiple data items traveling through the loop, in different phases, but this priority solution limits the amount of data kept in the loop's queues and allows the queue flow control to prevent too much data from entering the loop.
The raised priority of the reverse nexuses can also be used to deliver the urgent messages. Remember, there is nothing preventing you from marking any nexus as reverse (as long as it doesn't create a loop consisting of only the reverse nexuses).
The downside of having the reverse nexuses connected to a thread is that it causes an extra overhead from the check with a mutex synchronization on each nextXtray(). The regular-priority direct nexuses use double-buffering, with locking a mutex only when the second buffer runs dry, and refilling it by swapping its contents with the whole collected first buffer. But the high-priority reverse nexuses have to be checked every time, even if they have no incoming data.
Within the same priority the data is processed in the round-robin order. More exactly, each refill of the double-buffering grabs the data from the first buffer of each facet and moves it to the second buffer. Then the second buffer is processed in the round-robin fashion until it runs out and another refill becomes needed.
The nextXtray() processes all the rowops from the incoming tray by calling them on the facet's FnReturn. Two special rowops are generated automatically even if they haven't been queued up explicitly, on the facet's labels _BEGIN_ and _END_ (to avoid extra overhead, they are actually generated only if there is any processing chained for them).
The nextXtray() automatically flushes the writers after processing a tray.
If a fatal error is encountered during processing (such as some code in a label died), nextXtray() will catch the exception, discard the rest of the tray and confess itself (without flushing the writers).
$result = $to->nextXtrayNoWait();
Similar to nextXtray(), but returns immediately if there is no data to process. Returns 0 if there is either no input data or the thread was requested to die (the way to differentiate between these cases is to call $to->isRqDead()).
$result = $to->nextXtrayTimeout($timeout);
Similar to nextXtrayNoWait(), only if there is no data, waits for up to the length of timeout. The timeout value is floating-point seconds. Returns 0 if the timeout has expired or the thread was requested to die.
$result = $to->nextXtrayTimeLimit($deadline);
Similar to nextXtrayNoWait(), only if there is no data, waits until the absolute deadline. The deadline value is time since epoch floating-point seconds, such as returned by Triceps::now(). Returns 0 if the wait reached the deadline or the thread was requested to die.
$to->mainLoop();
Process the incoming trays until the thread is requested to die. The exact implementation of the main loop (in C++) is:
void TrieadOwner::mainLoop()
{
while (nextXtray())
{ }
}
It also used to call markDead() after the loop, and my earlier posts might say so, but now I've realized that it would not be advisable in the situations when the thread needs to do a lengthy saving of its state, blocking the harvester, so I've removed it.
The drain API of the TrieadOwner is very similar to the one in the App. The best way to do the drain is by the automatically-scoped AutoDrain class. If a drain doesn't need an automatic scoping, use the TrieadOwner API. And finally if you want to mess with drains from outside an app's Triead and thus don't have a TrieadOwner, only then use the App API.
$to->requestDrainShared();$to->requestDrainExclusive();
$to->waitDrain();
$to->drainShared();
$to->drainExclusive();
$to->undrain();
$result = $to->isDrained();
The methods are used in exactly the same way as the similar App methods, with only the difference of the names on the shared drains.
The exclusive drains always make the exclusion for this Triead. (Only one thread can be excluded from a drain). Normally the exclusive drains should be used only for the input-only threads. They could potentially be used to exclude the non-input-only thread too but I'm not sure, what's the point, and haven't worked out if it would work reliably (it might, or it might not).
$result = $to->isRqDrain();
Check whether a drain request is active. This can be used in the threads that generate data based on the real-time clock yet aren't input-only: if they find that a drain is active, they should refrain from generating the data and go back to the waiting. There is no way for them to find when the drain is released, so they should just continue to the next timeout as usual. Such code must use nextXtrayTimeout() or nextXtrayTimeLimit() for the timeouts, or the drain would never complete. The input-only threads don't have this limitation. And of course keep in mind that the better practice is to deal with the deal time either in the input-only threads or by driving it from outside the model altogether.
The same as the method on the Triead, equivalent to $to->get()->exports(). The returned array contains the name-value pairs of the nexus names and objects.
@imports = $to->imports();
This method is different from the Triead method. It still returns an array of name-value pairs but the values are Facets (not Nexuses, as in Triead). It's a natural difference, since the facets are useful in the owner thread, and available only in it.
$result = $to->flushWriters();
Flush any data collected in the writer facets, sending them to the appropriate nexuses. The data in each facet becomes a tray that is sent to the nexus (if there was no data collected on a facet, nothing will be sent from it). Returns 1 if the flush was completed, 0 if the thread was requested to die and this the data was discarded. The data is never sent out of a facet by itself, it always must be flushed in one of the explicit ways (TrieadOwner::flushWriters(), Facet::flushWriter(), or enqueueing a rowop on the facet's labels _BEGIN_ and _END_). The flush may get stuck if this is an input-only thread and a drain is active, it will wait until the drain is released.
$to->requestMyselfDead();
Request this Triead itself to die. This is the way to disconnect from the nexuses while the thread is exiting on its own. For example, if this thread going to dump its data before exit to a large file that takes half an hour to write, normally the data queued for this thread might fill up the queues in the nexuses, and it's a bad practice to keep the other threads stuck due to the overflowing buffers. Requesting this thread to die disconnects it from the nexuses and prevents the data from collecting.The thread could also be disconnected by marking it dead, but it will keep the harvester stuck waiting to join it while the thread completes its long write, and that's not so good either. So this call provides the solution, avoiding both pitfalls.
$result = $to->nextXtray();
Process one incoming tray from a single reader nexus (any nexus where data is available). A tray essentially embodies a transaction, and "X" stands for "cross-thread". There actually is the Xtray type that represents the Tray in a special thread-safe format but it's used only inside the nexuses and not visible from outside.
If there is currently no data to process, this method will wait.
The return value is 1 normally, or 0 if the thread was requested to die. So the typical usage is:
while($to->nextXtray()) { ... }
The method mainLoop() encapsulates the most typical usage, and nextXtray() needs to be used directly only in the more unusual circumstances.
The data is read from the reverse nexuses first, at a higher priority. If any reverse nexus has data available, it will always be read before the direct nexuses. A reverse nexus typically completes a topological loop, so this priority creates the preference to cycle the data through the loop until it comes out, before accepting more data into the loop. Since all the nexuses have non-zero-length queues, obviously, there will be multiple data items traveling through the loop, in different phases, but this priority solution limits the amount of data kept in the loop's queues and allows the queue flow control to prevent too much data from entering the loop.
The raised priority of the reverse nexuses can also be used to deliver the urgent messages. Remember, there is nothing preventing you from marking any nexus as reverse (as long as it doesn't create a loop consisting of only the reverse nexuses).
The downside of having the reverse nexuses connected to a thread is that it causes an extra overhead from the check with a mutex synchronization on each nextXtray(). The regular-priority direct nexuses use double-buffering, with locking a mutex only when the second buffer runs dry, and refilling it by swapping its contents with the whole collected first buffer. But the high-priority reverse nexuses have to be checked every time, even if they have no incoming data.
Within the same priority the data is processed in the round-robin order. More exactly, each refill of the double-buffering grabs the data from the first buffer of each facet and moves it to the second buffer. Then the second buffer is processed in the round-robin fashion until it runs out and another refill becomes needed.
The nextXtray() processes all the rowops from the incoming tray by calling them on the facet's FnReturn. Two special rowops are generated automatically even if they haven't been queued up explicitly, on the facet's labels _BEGIN_ and _END_ (to avoid extra overhead, they are actually generated only if there is any processing chained for them).
The nextXtray() automatically flushes the writers after processing a tray.
If a fatal error is encountered during processing (such as some code in a label died), nextXtray() will catch the exception, discard the rest of the tray and confess itself (without flushing the writers).
$result = $to->nextXtrayNoWait();
Similar to nextXtray(), but returns immediately if there is no data to process. Returns 0 if there is either no input data or the thread was requested to die (the way to differentiate between these cases is to call $to->isRqDead()).
$result = $to->nextXtrayTimeout($timeout);
Similar to nextXtrayNoWait(), only if there is no data, waits for up to the length of timeout. The timeout value is floating-point seconds. Returns 0 if the timeout has expired or the thread was requested to die.
$result = $to->nextXtrayTimeLimit($deadline);
Similar to nextXtrayNoWait(), only if there is no data, waits until the absolute deadline. The deadline value is time since epoch floating-point seconds, such as returned by Triceps::now(). Returns 0 if the wait reached the deadline or the thread was requested to die.
$to->mainLoop();
Process the incoming trays until the thread is requested to die. The exact implementation of the main loop (in C++) is:
void TrieadOwner::mainLoop()
{
while (nextXtray())
{ }
}
It also used to call markDead() after the loop, and my earlier posts might say so, but now I've realized that it would not be advisable in the situations when the thread needs to do a lengthy saving of its state, blocking the harvester, so I've removed it.
The drain API of the TrieadOwner is very similar to the one in the App. The best way to do the drain is by the automatically-scoped AutoDrain class. If a drain doesn't need an automatic scoping, use the TrieadOwner API. And finally if you want to mess with drains from outside an app's Triead and thus don't have a TrieadOwner, only then use the App API.
$to->requestDrainShared();$to->requestDrainExclusive();
$to->waitDrain();
$to->drainShared();
$to->drainExclusive();
$to->undrain();
$result = $to->isDrained();
The methods are used in exactly the same way as the similar App methods, with only the difference of the names on the shared drains.
The exclusive drains always make the exclusion for this Triead. (Only one thread can be excluded from a drain). Normally the exclusive drains should be used only for the input-only threads. They could potentially be used to exclude the non-input-only thread too but I'm not sure, what's the point, and haven't worked out if it would work reliably (it might, or it might not).
$result = $to->isRqDrain();
Check whether a drain request is active. This can be used in the threads that generate data based on the real-time clock yet aren't input-only: if they find that a drain is active, they should refrain from generating the data and go back to the waiting. There is no way for them to find when the drain is released, so they should just continue to the next timeout as usual. Such code must use nextXtrayTimeout() or nextXtrayTimeLimit() for the timeouts, or the drain would never complete. The input-only threads don't have this limitation. And of course keep in mind that the better practice is to deal with the deal time either in the input-only threads or by driving it from outside the model altogether.
Subscribe to:
Posts (Atom)