Primitive operators that start consistent regions

Consistent region permits

In general, start operators of a consistent region are source operators. In C++, source operators must implement the process(uint32_t) method, which is the function that is called by the InfoSphere® Streams instance to run the main logic of a thread. When the operator is outside of a consistent region, the operator can submit tuples at any time. When the operator is in a consistent region, the operator must pause tuple submission every time a drain or a reset occurs.

Code your operator so drains and resets happen at natural points of an operator execution, such as after the operators change their internal state and when they submit a tuple, or after a more elaborate operation, such as when the operators reach an end of file while they read a file. After you define these natural points, acquire permits to guard this code block with a tuple submission permit that is specific to a consistent region.

A consistent region permit ensures that if an operator thread is running the code block, drains, and resets do not occur. Multiple threads can hold permits at the same time. However, drains and resets occur after they receive the corresponding notifications from the consistent region controller, and after all operator threads release their permits. After that, the InfoSphere Streams instance calls the StateHandler methods (drain, checkpoint, and reset), and starts draining or resetting the downstream operators of the region.

If any thread attempts to acquire a permit when a checkpoint or reset is in progress, the thread blocks. A thread also blocks on a permit when the operator restarts and has not yet reset. The thread is granted a permit, that is, unblocks, after a consistent state is established or reset successfully and the consistent region controller notifies that tuple submission can be resumed.

A Java™ primitive operator can acquire and release permits by calling acquirePermit() and releasePermit() from a ConsistentRegionContext. Releasing a permit is an indication that a consistent state can be correctly established or reset. If an unrecoverable error happens when a permit is being held, the operator must explicitly indicate that a consistent region must reset via the ConsistentRegionContext.

A thread can block when it tries to acquire a lock. As a result, be mindful of when the code acquires a lock to avoid deadlock situations. If a thread must acquire the lock, then acquire it inside the code block, which is running with a permit. The reason to acquire a lock inside the code block with a permit is that the same lock might be acquired by a different thread, which then cannot acquire the lock until the drain or reset is complete. An example of a different thread that is unable to acquire the lock if it is not acquired inside a code block with a permit is the SPL runtime thread, which cannot not acquire the lock when the thread calls checkpoint or reset.

Sample start operator

This example concerns only source operators that start a consistent region. The same principles that are used for acquiring permits can be applied to primitive operators that do not start a region but have background threads that do tuple submission.

Header file

<%
  my $crContext = 
    $model->getContext()->getOptionalContext("ConsistentRegion");
  my @includes;
  if ($crContext) {
    push @includes,        
      "#include <SPL/Runtime/Operator/State/StateHandler.h>";
      "#include <SPL/Runtime/Operator/State/ConsistentRegionContext.h>";
   }
   my $isTriggerOperator =
     $crContext->isTriggerOperator();
   if ($isTriggerOperator) {
     SPL::CodeGen::errorln("Operator does not support trigger=operatorDriven in a 
consistent region.", $model->getContext()->getSourceLocation());
   }
   SPL::CodeGen::headerPrologue($model,\@includes); 
%>

C++ file

Similar to the header file, the C++ file uses the code generation API to conditionally compile code that is used only when the operator is in a consistent region.

By binding the tuple submission code to the state update code, the operator ensures that both operations occur as a single unit during a drain or a reset. This code means that when the operator checkpoints or resets its state, the state of the operator reflects the fact that both operations occurred. If these operations were not coupled, the operator code would need to be able to deal with different reset situations, for example, a reset in which the operator submitted the tuple but did not update the internal state. This issue arises because a drain or a reset can occur anytime the operator code is not running with an acquired permit.

void MY_OPERATOR::process(uint32_t) 
{
  ProcessingElement& pe = getPE();
  while(!pe.getShutdownRequested()) {
    <%if ($isInConsistentRegion) {%>
      {  
        ConsistentRegionPermit crp(_crContext);
    <%}%>
    {
      AutoMutex am (_mutex);
      OPort0Type tuple(_counter);
      submit (tuple, 0);
      _counter++;
    }
    <%if ($isInConsistentRegion) {%>
      }
    <%}%>
    pe.blockUntilShutdownRequest(0.2);
  }
}