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.
Showing posts with label drain. Show all posts
Showing posts with label drain. Show all posts
Saturday, July 13, 2013
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.
Friday, June 28, 2013
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.
Monday, April 29, 2013
Multithreaded socket server, part4, socket reader thread
As I've said before, each socket gets served by two threads: one sits reading from the socket and forwards the data into the model and another one sits getting data from the model and forwards it into the socket. Since the same thread can't wait for both a socket descriptor and a thread synchronization primitive, so they have to be separate.
The first thread started is the socket reader. Let's go through it bit by bit.
sub chatSockReadT
{
my $opts = {};
&Triceps::Opt::parse("chatSockReadT", $opts, {@Triceps::Triead::opts,
socketName => [ undef, \&Triceps::Opt::ck_mandatory ],
}, @_);
undef @_; # avoids a leak in threads module
my $owner = $opts->{owner};
my $app = $owner->app();
my $unit = $owner->unit();
my $tname = $opts->{thread};
# only dup the socket, the writer thread will consume it
my ($tsock, $sock) = $owner->trackDupSocket($opts->{socketName}, "<");
The beginning is quite usual. Then it loads the socked from the App and gets it tracked with the TrieadOwner. The difference between trackDupSocket() here and the trackGetSocket() used before is that trackDupSocket() leaves the socked copy in the App, to be found by the writer-side thread.
The socket is reopened in this thread as read-only. The writing to the socket from all the threads has to be synchronized to avoid mixing the half-messages. And the easiest way to synchronize is to always write from one thread, and if the other thread wants to write something, it has to pass the data to the writer thread through the control nexus.
# user messages will be sent here
my $faChat = $owner->importNexus(
from => "global/chat",
import => "writer",
);
# control messages to the reader side will be sent here
my $faCtl = $owner->makeNexus(
name => "ctl",
labels => [
ctl => $faChat->impRowType("ctl"),
],
reverse => 1, # gives this nexus a high priority
import => "writer",
);
Imports the chat nexus and creates the private control nexus for communication with the writer side. The name of the chat nexus is hardcoded hare, since it's pretty much a solid part of the application. If this were a module, the name of the chat nexus could be passed through the options.
The control nexus is marked as reverse even though it really isn't. But the reverse option has a side effect of making this nexus high-priority. Even if the writer thread has a long queue of messages from the chat nexus, the messages from the control nexus will be read first. Which again isn't strictly necessary here, but I wanted to show how it's done.
The type of the control label is imported from the chat nexus, so it doesn't have to be defined from scratch.
$owner->markConstructed();
Triceps::Triead::start(
app => $opts->{app},
thread => "$tname.rd",
fragment => $opts->{fragment},
main => \&chatSockWriteT,
socketName => $opts->{socketName},
ctlFrom => "$tname/ctl",
);
$owner->readyReady();
Then the construction is done and the writer thread gets started. And then the thread is ready and waits for the writer thread to be ready too. The readyReady() works in the fragments just as it does at the start of the app. Whenever a new thread is started, the App becomes not ready, and stays this way until all the threads report that they are ready. The rest of the App keeps working like nothing happened, at least sort of. Whenever a nexus is imported, the messages from this nexus start collecting for this thread, and if there are many of them, the nexus will become backed up and the threads writing to them will block. The new threads have to call readyReady() as usual to synchronize between themselves, and then everything gets on its way.
Of course, if two connections are received in a quick succession, that would start two sets of threads, and readyReady() will continue only after all of them are ready.
my $lbChat = $faChat->getLabel("msg");
my $lbCtl = $faCtl->getLabel("ctl");
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => "print", arg => "!ready," . $opts->{fragment});
$owner->flushWriters();
A couple of labels get remembered for the future use, and the connection ready message gets sent to the writer thread through the control nexus. By convention of this application, the messages go in the CVS format, with the control messages starting with "!". If this is the first client, this would send
!ready,cliconn1
to the client. It's important to call flushWriters() every time to get the message(s) delivered.
while(<$sock>) {
s/[\r\n]+$//;
my @data = split(/,/);
if ($data[0] eq "exit") {
last; # a special case, handle in this thread
} elsif ($data[0] eq "kill") {
eval {$app->shutdownFragment($data[1]);};
if ($@) {
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => "print", arg => "!error,$@");
$owner->flushWriters();
}
} elsif ($data[0] eq "shutdown") {
$unit->makeHashCall($lbChat, "OP_INSERT", topic => "*", msg => "server shutting down");
$owner->flushWriters();
Triceps::AutoDrain::makeShared($owner);
eval {$app->shutdown();};
} elsif ($data[0] eq "publish") {
$unit->makeHashCall($lbChat, "OP_INSERT", topic => $data[1], msg => $data[2]);
$owner->flushWriters();
} else {
# this is not something you want to do in a real chat application
# but it's cute for a demonstration
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => $data[0], arg => $data[1]);
$owner->flushWriters();
}
}
The main loop keeps reading lines from the socket and interpreting them. The lines are in CSV format, and the first field is the command and the rest are the arguments (if any). The commands are:
publish - send a message with a topic to the chat nexus
exit - close the connection
kill - close another connection, by name
shutdown - shut down the server
subscribe - subscribe the client to a topic
unsibscribe - unsubscribe the client from a topic
The exit just exits the loop, since it works the same if the socket just gets closed from the other side.
The kill shuts down by name the fragment where the threads of the other connection belongs. This is a simple application, so it doesn't check any permissions, whether this fragment should allowed to be shut down. If there is no such fragment, the shutdown call will silently do nothing, so the error check and reporting is really redundant (if something goes grossly wrong in the thread interruption code, an error might still occur, but theoretically this should never happen).
The shutdown sends the notification to the common topic "*" (to which all the clients are subscribed by default), then drains the model and shuts it down. The drain makes sure that all the messages in the model get processed (and even written to the socket) without allowing any new messages to be injected. "Shared" means that there is no special exceptions for some threads.
makeShared() actually creates a drain object that keeps the drain active during its lifetime. Here this object is not assigned anywhere, so it gets immediately destroyed and lifts the drain. So potentially more messages can get squeezed in between this point and shutdown. Which doesn't matter a whole lot here.
If it were really important that nothing get sent after the shutdown notification, it could be done like this (this is an untested fragment, so it might contain typos):
} elsif ($data[0] eq "shutdown") {
my $drain = Triceps::AutoDrain::makeExclusive($owner);
$unit->makeHashCall($lbChat, "OP_INSERT", topic => "*", msg => "server shutting down");
$owner->flushWriters();
$drain->wait();
eval {$app->shutdown();};
}
This starts the drain, but this time the exclusive mode means that this thread is allowed to send more data. When the drain is created, it waits for success, so when the new message is inserted, it will be after all the other messages. $drain->wait() does another wait and makes sure that this last message propagates all the way. And then the app gets shut down, while the drain is still in effect, so no more messages can be sent for sure.
The publish sends the data to the chat nexus (note the flushWriters(), as usual!).
And the rest of commands (that would be subscribe and unsubscribe but you can do any other commands like "print") get simply forwarded to the reader thread for execution. Sending through the commands like this without testing is not a good practice for a real application but it's cute for a demo.
{
# let the data drain through
my $drain = Triceps::AutoDrain::makeExclusive($owner);
# send the notification - can do it because the drain is excluding itself
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => "print", arg => "!exiting");
$owner->flushWriters();
$drain->wait(); # wait for the notification to drain
$app->shutdownFragment($opts->{fragment});
}
$tsock->close(); # not strictly necessary
}
The last part is when the connection get closed, either by the "exit" command or when the socket gets closed. Remember, the socket can get closed asymmetrically, in one direction, so even when the reading is closed, the writing may still work and needs to return the responses to any commands received from the socket. And of course the same is true for the "exit" command.
So here the full exclusive drain sequence is used, ending with the shutdown of this thread's own fragment, which will close the socket. Even though only one fragment needs to be shut down, the drain drains the whole model. Because of the potentially complex interdependencies, there is no way to reliably drain only a part, and all the drains are App-wide.
The last part, with $tsock->close(), is not technically necessary since the shutdown of the fragment will get the socket descriptor revoked anyway. But other than that, it's a good practice that unregisters the socket from the TrieadOwner and then closes it.
The first thread started is the socket reader. Let's go through it bit by bit.
sub chatSockReadT
{
my $opts = {};
&Triceps::Opt::parse("chatSockReadT", $opts, {@Triceps::Triead::opts,
socketName => [ undef, \&Triceps::Opt::ck_mandatory ],
}, @_);
undef @_; # avoids a leak in threads module
my $owner = $opts->{owner};
my $app = $owner->app();
my $unit = $owner->unit();
my $tname = $opts->{thread};
# only dup the socket, the writer thread will consume it
my ($tsock, $sock) = $owner->trackDupSocket($opts->{socketName}, "<");
The beginning is quite usual. Then it loads the socked from the App and gets it tracked with the TrieadOwner. The difference between trackDupSocket() here and the trackGetSocket() used before is that trackDupSocket() leaves the socked copy in the App, to be found by the writer-side thread.
The socket is reopened in this thread as read-only. The writing to the socket from all the threads has to be synchronized to avoid mixing the half-messages. And the easiest way to synchronize is to always write from one thread, and if the other thread wants to write something, it has to pass the data to the writer thread through the control nexus.
# user messages will be sent here
my $faChat = $owner->importNexus(
from => "global/chat",
import => "writer",
);
# control messages to the reader side will be sent here
my $faCtl = $owner->makeNexus(
name => "ctl",
labels => [
ctl => $faChat->impRowType("ctl"),
],
reverse => 1, # gives this nexus a high priority
import => "writer",
);
Imports the chat nexus and creates the private control nexus for communication with the writer side. The name of the chat nexus is hardcoded hare, since it's pretty much a solid part of the application. If this were a module, the name of the chat nexus could be passed through the options.
The control nexus is marked as reverse even though it really isn't. But the reverse option has a side effect of making this nexus high-priority. Even if the writer thread has a long queue of messages from the chat nexus, the messages from the control nexus will be read first. Which again isn't strictly necessary here, but I wanted to show how it's done.
The type of the control label is imported from the chat nexus, so it doesn't have to be defined from scratch.
$owner->markConstructed();
Triceps::Triead::start(
app => $opts->{app},
thread => "$tname.rd",
fragment => $opts->{fragment},
main => \&chatSockWriteT,
socketName => $opts->{socketName},
ctlFrom => "$tname/ctl",
);
$owner->readyReady();
Then the construction is done and the writer thread gets started. And then the thread is ready and waits for the writer thread to be ready too. The readyReady() works in the fragments just as it does at the start of the app. Whenever a new thread is started, the App becomes not ready, and stays this way until all the threads report that they are ready. The rest of the App keeps working like nothing happened, at least sort of. Whenever a nexus is imported, the messages from this nexus start collecting for this thread, and if there are many of them, the nexus will become backed up and the threads writing to them will block. The new threads have to call readyReady() as usual to synchronize between themselves, and then everything gets on its way.
Of course, if two connections are received in a quick succession, that would start two sets of threads, and readyReady() will continue only after all of them are ready.
my $lbChat = $faChat->getLabel("msg");
my $lbCtl = $faCtl->getLabel("ctl");
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => "print", arg => "!ready," . $opts->{fragment});
$owner->flushWriters();
A couple of labels get remembered for the future use, and the connection ready message gets sent to the writer thread through the control nexus. By convention of this application, the messages go in the CVS format, with the control messages starting with "!". If this is the first client, this would send
!ready,cliconn1
to the client. It's important to call flushWriters() every time to get the message(s) delivered.
while(<$sock>) {
s/[\r\n]+$//;
my @data = split(/,/);
if ($data[0] eq "exit") {
last; # a special case, handle in this thread
} elsif ($data[0] eq "kill") {
eval {$app->shutdownFragment($data[1]);};
if ($@) {
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => "print", arg => "!error,$@");
$owner->flushWriters();
}
} elsif ($data[0] eq "shutdown") {
$unit->makeHashCall($lbChat, "OP_INSERT", topic => "*", msg => "server shutting down");
$owner->flushWriters();
Triceps::AutoDrain::makeShared($owner);
eval {$app->shutdown();};
} elsif ($data[0] eq "publish") {
$unit->makeHashCall($lbChat, "OP_INSERT", topic => $data[1], msg => $data[2]);
$owner->flushWriters();
} else {
# this is not something you want to do in a real chat application
# but it's cute for a demonstration
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => $data[0], arg => $data[1]);
$owner->flushWriters();
}
}
The main loop keeps reading lines from the socket and interpreting them. The lines are in CSV format, and the first field is the command and the rest are the arguments (if any). The commands are:
publish - send a message with a topic to the chat nexus
exit - close the connection
kill - close another connection, by name
shutdown - shut down the server
subscribe - subscribe the client to a topic
unsibscribe - unsubscribe the client from a topic
The exit just exits the loop, since it works the same if the socket just gets closed from the other side.
The kill shuts down by name the fragment where the threads of the other connection belongs. This is a simple application, so it doesn't check any permissions, whether this fragment should allowed to be shut down. If there is no such fragment, the shutdown call will silently do nothing, so the error check and reporting is really redundant (if something goes grossly wrong in the thread interruption code, an error might still occur, but theoretically this should never happen).
The shutdown sends the notification to the common topic "*" (to which all the clients are subscribed by default), then drains the model and shuts it down. The drain makes sure that all the messages in the model get processed (and even written to the socket) without allowing any new messages to be injected. "Shared" means that there is no special exceptions for some threads.
makeShared() actually creates a drain object that keeps the drain active during its lifetime. Here this object is not assigned anywhere, so it gets immediately destroyed and lifts the drain. So potentially more messages can get squeezed in between this point and shutdown. Which doesn't matter a whole lot here.
If it were really important that nothing get sent after the shutdown notification, it could be done like this (this is an untested fragment, so it might contain typos):
} elsif ($data[0] eq "shutdown") {
my $drain = Triceps::AutoDrain::makeExclusive($owner);
$unit->makeHashCall($lbChat, "OP_INSERT", topic => "*", msg => "server shutting down");
$owner->flushWriters();
$drain->wait();
eval {$app->shutdown();};
}
This starts the drain, but this time the exclusive mode means that this thread is allowed to send more data. When the drain is created, it waits for success, so when the new message is inserted, it will be after all the other messages. $drain->wait() does another wait and makes sure that this last message propagates all the way. And then the app gets shut down, while the drain is still in effect, so no more messages can be sent for sure.
The publish sends the data to the chat nexus (note the flushWriters(), as usual!).
And the rest of commands (that would be subscribe and unsubscribe but you can do any other commands like "print") get simply forwarded to the reader thread for execution. Sending through the commands like this without testing is not a good practice for a real application but it's cute for a demo.
{
# let the data drain through
my $drain = Triceps::AutoDrain::makeExclusive($owner);
# send the notification - can do it because the drain is excluding itself
$unit->makeHashCall($lbCtl, "OP_INSERT", cmd => "print", arg => "!exiting");
$owner->flushWriters();
$drain->wait(); # wait for the notification to drain
$app->shutdownFragment($opts->{fragment});
}
$tsock->close(); # not strictly necessary
}
The last part is when the connection get closed, either by the "exit" command or when the socket gets closed. Remember, the socket can get closed asymmetrically, in one direction, so even when the reading is closed, the writing may still work and needs to return the responses to any commands received from the socket. And of course the same is true for the "exit" command.
So here the full exclusive drain sequence is used, ending with the shutdown of this thread's own fragment, which will close the socket. Even though only one fragment needs to be shut down, the drain drains the whole model. Because of the potentially complex interdependencies, there is no way to reliably drain only a part, and all the drains are App-wide.
The last part, with $tsock->close(), is not technically necessary since the shutdown of the fragment will get the socket descriptor revoked anyway. But other than that, it's a good practice that unregisters the socket from the TrieadOwner and then closes it.
Thursday, April 11, 2013
Multithreaded pipeline, part 2
The reader thread drives the pipeline:
sub ReaderMain # (@opts)
{
my $opts = {};
Triceps::Opt::parse("traffic main", $opts, {@Triceps::Triead::opts}, @_);
my $owner = $opts->{owner};
my $unit = $owner->unit();
my $rtPacket = Triceps::RowType->new(
time => "int64", # packet's timestamp, microseconds
local_ip => "string", # string to make easier to read
remote_ip => "string", # string to make easier to read
local_port => "int32",
remote_port => "int32",
bytes => "int32", # size of the packet
) or confess "$!";
my $rtPrint = Triceps::RowType->new(
text => "string", # the text to print (including \n)
) or confess "$!";
my $rtDumprq = Triceps::RowType->new(
what => "string", # identifies, what to dump
) or confess "$!";
my $faOut = $owner->makeNexus(
name => "data",
labels => [
packet => $rtPacket,
print => $rtPrint,
dumprq => $rtDumprq,
],
import => "writer",
);
my $lbPacket = $faOut->getLabel("packet");
my $lbPrint = $faOut->getLabel("print");
my $lbDumprq = $faOut->getLabel("dumprq");
$owner->readyReady();
while(<STDIN>) {
chomp;
# print the input line, as a debugging exercise
$unit->makeArrayCall($lbPrint, "OP_INSERT", "! $_\n");
my @data = split(/,/); # starts with a command, then string opcode
my $type = shift @data;
if ($type eq "new") {
$unit->makeArrayCall($lbPacket, @data);
} elsif ($type eq "dump") {
$unit->makeArrayCall($lbDumprq, "OP_INSERT", $data[0]);
} else {
$unit->makeArrayCall($lbPrint, "OP_INSERT", "Unknown command '$type'\n");
}
$owner->flushWriters();
}
{
# drain the pipeline before shutting down
my $ad = Triceps::AutoDrain::makeShared($owner);
$owner->app()->shutdown();
}
}
It starts by creating the nexus with the initial set of the labels: for the data about the network packets, for the lines to be printed at the end of the pipeline and for the dump requests to the tables in the other threads. It gets exported for the other threads to import, and also imported right back into this thread, for writing. And then the setup is done, readyReady() is called, and the processing starts.
It reads the CSV lines, splits them, makes a decision if it's a data line or dump request, and one way or the other sends it into the nexus. The data sent to a facet doesn't get immediately forwarded to the nexus. It's collected internally in a tray, and then flushWriters() sends it on. The mainLoop() shown in the last post calls flushWriters() automatically after every tray it processes from the input. But when reading from a file you've got to do it yourself. Of course, it's more efficient to send through multiple rows at once, so a smarter implementation would check if multiple lines are available from the file and send them in larger bundles.
The last part is the shutdown. After the end of file is reached, it's time to shut down the application. You can't just shut down it right away because there still might be data in the pipeline, and if you shut it down, that data will be lost. The right way is to drain the pipeline first, and then do the shutdown when the app is drained. AutoDrain::makeShared() creates a scoped drain: the drain request for all the threads is started when this object is created, and the object construction completes when the drain succeeds. When the object is destroyed, that lifts the drain. So in this case the drain succeeds and then the app gets shut down.
The shutdown causes the mainLoop() calls in all the other threads to return, and the threads to exit. Then startHere() in the first thread has the special logic in it that joins all the started threads after its own main function returns and before it completes. After that the script continues on its way and is free to exit.
sub ReaderMain # (@opts)
{
my $opts = {};
Triceps::Opt::parse("traffic main", $opts, {@Triceps::Triead::opts}, @_);
my $owner = $opts->{owner};
my $unit = $owner->unit();
my $rtPacket = Triceps::RowType->new(
time => "int64", # packet's timestamp, microseconds
local_ip => "string", # string to make easier to read
remote_ip => "string", # string to make easier to read
local_port => "int32",
remote_port => "int32",
bytes => "int32", # size of the packet
) or confess "$!";
my $rtPrint = Triceps::RowType->new(
text => "string", # the text to print (including \n)
) or confess "$!";
my $rtDumprq = Triceps::RowType->new(
what => "string", # identifies, what to dump
) or confess "$!";
my $faOut = $owner->makeNexus(
name => "data",
labels => [
packet => $rtPacket,
print => $rtPrint,
dumprq => $rtDumprq,
],
import => "writer",
);
my $lbPacket = $faOut->getLabel("packet");
my $lbPrint = $faOut->getLabel("print");
my $lbDumprq = $faOut->getLabel("dumprq");
$owner->readyReady();
while(<STDIN>) {
chomp;
# print the input line, as a debugging exercise
$unit->makeArrayCall($lbPrint, "OP_INSERT", "! $_\n");
my @data = split(/,/); # starts with a command, then string opcode
my $type = shift @data;
if ($type eq "new") {
$unit->makeArrayCall($lbPacket, @data);
} elsif ($type eq "dump") {
$unit->makeArrayCall($lbDumprq, "OP_INSERT", $data[0]);
} else {
$unit->makeArrayCall($lbPrint, "OP_INSERT", "Unknown command '$type'\n");
}
$owner->flushWriters();
}
{
# drain the pipeline before shutting down
my $ad = Triceps::AutoDrain::makeShared($owner);
$owner->app()->shutdown();
}
}
It starts by creating the nexus with the initial set of the labels: for the data about the network packets, for the lines to be printed at the end of the pipeline and for the dump requests to the tables in the other threads. It gets exported for the other threads to import, and also imported right back into this thread, for writing. And then the setup is done, readyReady() is called, and the processing starts.
It reads the CSV lines, splits them, makes a decision if it's a data line or dump request, and one way or the other sends it into the nexus. The data sent to a facet doesn't get immediately forwarded to the nexus. It's collected internally in a tray, and then flushWriters() sends it on. The mainLoop() shown in the last post calls flushWriters() automatically after every tray it processes from the input. But when reading from a file you've got to do it yourself. Of course, it's more efficient to send through multiple rows at once, so a smarter implementation would check if multiple lines are available from the file and send them in larger bundles.
The last part is the shutdown. After the end of file is reached, it's time to shut down the application. You can't just shut down it right away because there still might be data in the pipeline, and if you shut it down, that data will be lost. The right way is to drain the pipeline first, and then do the shutdown when the app is drained. AutoDrain::makeShared() creates a scoped drain: the drain request for all the threads is started when this object is created, and the object construction completes when the drain succeeds. When the object is destroyed, that lifts the drain. So in this case the drain succeeds and then the app gets shut down.
The shutdown causes the mainLoop() calls in all the other threads to return, and the threads to exit. Then startHere() in the first thread has the special logic in it that joins all the started threads after its own main function returns and before it completes. After that the script continues on its way and is free to exit.
Subscribe to:
Posts (Atom)