チェックポイント

stream<...> Out1 = MyOper1(...) {
   ...
   config
      checkpoint : operatorDriven; // operator driven checkpointing
      restartable : true;
}
 
stream<...> Out2 = MyOper2(...) {
   ...
   config
      checkpoint : periodic(10.5); // periodic checkpointing; checkpointing interval is 10.5 seconds
      restartable : true;
}

#include <SPL/Runtime/Operator/StateHandler.h>
 
class MY_OPERATOR : public MY_BASE_OPERATOR, StateHandler // inherit from StateHandler
{
public:
    ...
 
    // callback functions from StateHandler.h
    virtual void checkpoint(Checkpoint & ckpt);
    virtual void reset(Checkpoint & ckpt);
    virtual void resetToInitialState();
private:
    // variables myInt_ and myVector_ constitute the state of this operator
    uint32_t myInt_;
    std::vector<IPort0Type> myVector_;    
    Mutex myMutex_; // lock for guarding concurrent access to operator state
};

// Constructor of operator
MY_OPERATOR::MY_OPERATOR()
    : myInt_(0)
{
    ...
    // register StateHandler callbacks
    getContext().registerStateHandler(*this);
}
 
// process() function for tuple processing
void MY_OPERATOR::process(Tuple const & tuple, uint32_t port)
{
    // acquire lock
    AutoMutex am(myMutex_);
              
    // manipulate operator state during tuple processing
    IPort0Type & ituple = static_cast<IPort0Type &>(tuple);
    myInt_ ++;
    myVector_.push_back(ituple);
 
    // submit tuple downstream
    OPort0Type otuple(tuple.getAttributeValue(0), myInt_);
    submit(otuple, 0);
}
 
// checkpoint() callback to checkpoint operator state data
void MY_OPERATOR::checkpoint(Checkpoint & ckpt)
{
    // acquire lock
    AutoMutex am(myMutex_);
 
    SPLAPPTRC(L_TRACE, "Checkpointing with sequence id: " << ckpt.getSequenceId(), "MY_OP");
              
    // serialize state data to Checkpoint
    ckpt << myInt_;
    ckpt << myVector_; 
}
 
// reset() callback to reset operator state data
void MY_OPERATOR::reset(Checkpoint & ckpt)
{
    // acquire lock
    AutoMutex am(myMutex_);
 
    SPLAPPTRC(L_TRACE, "Resetting with sequence id: " << ckpt.getSequenceId(), "MY_OP");
              
    // de-serialize state data from Checkpoint   
    ckpt >> myInt_;
    ckpt >> myVector_; 
}
 
// resetToIntialState() callback to reset operator to initial state
void MY_OPERATOR::resetToInitialState()
{
    // acquire lock
    AutoMutex am(myMutex_);
    myInt_ = 0;
    myVector_.clear();
}

// auto-generated functions for checkpointing and resetting logic-state variables
void MY_BASE_OPERATOR::checkpointStateVariables(Checkpoint & ckpt);
void MY_BASE_OPERATOR::resetStateVariables(Checkpoint & ckpt);

// the header file which defines CheckpointContext class
#include <SPL/Runtime/Operator/State/CheckpointContext.h>
.... 
// operator code can use CheckpointContext::createCheckpoint() function for operator driven checkpointing;
// to do so, first retrieve the CheckpointContext instance
CheckpointContext * ckptContext = static_cast<CheckpointContext *>(this->getContext().getOptionalContext(CHECKPOINT));
if (ckptContext != NULL) {
    // then call CheckpointContext::createCheckpoint() to perform checkpointing
    ckptContext->createCheckpoint();
}

<%
    # use CodeGen API to query whether the operator is used inside a consistent region
    my $isInConsistentRegion = $model->getContext()->getOptionalContext("ConsistentRegion");
    # if operator is outside consistent region, use CodeGen API to determine the kind of checkpointing (none, periodic, or operatorDriven)
    my $ckptKind = $model->getContext()->getCheckpointingKind();
    my @includes;
    if ($isInConsistentRegion) {
        push @includes, "#include <SPL/Runtime/Operator/State/StateHandler.h>";
    }
    if ($ckptKind ne "none") {
        push @includes, "#include <SPL/Runtime/Operator/State/CheckpointContext.h>";
    }
    SPL::CodeGen::headerPrologue($model, \@includes);   
%>

<% if ($ckptKind eq "operatorDirven") { %>
#include <boost/thread/recursive_mutex.hpp>
<% } %>
 
class MY_OPERATOR : public MY_BASE_OPERATOR
<% if ($isInConsistentRegion || $ckptKind ne "none") {%>
, public StateHandler
<% } %> {
public:
    // constructor
    MY_OPERATOR();
              
    // destructor
    virtual ~MY_OPERATOR();
   
    // tuple processing for mutating ports
    void process(Tuple & tuple, uint32_t port);
    ...
 
<% if ($isInConsistentRegion || $ckptKind ne "none") { %>
    // StateHandler callbacks
    virtual void checkpoint(Checkpoint & ckpt);
    virtual void reset(Checkpoint & ckpt);
    virtual void resetToIntialState();
<% } %>
 
private:
<% if ($ckptKind eq "operatorDirven") { %>
    // for operator-driven checkpointing, use a re-entrant lock to guard operator state
    // so that the thread calling createCheckpoint() does not deadlock itself
    boost::recursive_mutex myLock_; 
<% } else { %>
    // for all other cases, use SPL::Mutex
    Mutex myLock_;
<% } %>
    // variables myVar1_ and myVar2_ constitute this operator's state
    int myVar1_;
    float myVar2_;
};

<%
   # use CodeGen API to query whether the operator is used inside a consistent region
   my $isInConsistentRegion = $model->getContext()->getOptionalContext("ConsistentRegion");
   # if operator is outside consistent region, use CodeGen API to determine the kind of checkpointing (none, periodic, or operatorDriven)
   my $ckptKind = $model->getContext()->getCheckpointingKind();
%>
...
 
<%if ($isInConsistentRegion || $ckptKind ne "none") {%>
 
// operator-provided StateHandler::checkpoint() callback
void MY_OPERATOR::checkpoint(Checkpoint & ckpt)
{
    <% if ($ckptKind eq " operatorDirven ") {
        // this operator is used outside consistent region and is configured with operator driven checkpointing
        // in this case, myLock_ is an re-entrant lock
        boost::recursive_mutex::scoped_lock am(myLock_);
    <% } else { %>
        // this operator is in consistent region, or outside consistent region but configured with periodic checkpointing
        // in this case, myLock_ is a regular SPL::Mutex
        AutoMutex am(myLock_);
    <% } %>
    // checkpoint operator state
    ckpt << myVar1_ << myVar2_;
}
 
// operator-provided StateHandler::reset() callback
void MY_OPERATOR::reset(Checkpoint & ckpt)
{
    <% if ($ckptKind eq " operatorDirven ") {
        // this operator is used outside consistent region and is configured with operator driven checkpointing
        // in this case, myLock_ is an re-entrant lock
        boost::recursive_mutex::scoped_lock am(myLock_);
    <% } else { %>
        // this operator is in consistent region, or outside consistent region but configured with periodic checkpointing
        // in this case, myLock_ is a regular SPL::Mutex
        AutoMutex am(myLock_);
    <% } %>
    // reset operator state
    ckpt >> myVar1_ >> myVar2_;
}
 
// operator-provided StateHandler::resetToInitialState() callback
void MY_OPERATOR::resetToInitialState()
{
    <% if ($ckptKind eq " operatorDirven ") {
        // this operator is used outside consistent region and is configured with operator driven checkpointing
        // in this case, myLock_ is an re-entrant lock
        boost::recursive_mutex::scoped_lock am(myLock_);
    <% } else { %>
        // this operator is in consistent region, or outside consistent region but configured with periodic checkpointing
        // in this case, myLock_ is a regular SPL::Mutex
        AutoMutex am(myLock_);
    <% } %>
    // reset operator state to initial values
    myVar1_ = 0;
    myVar2_ = 0.0;
}
<%}%>
 
// tuple processing function
void MY_OPERATOR::process(Tuple & tuple, uint32_t port)
{
    <% if ($ckptKind eq " operatorDirven ") {
        // this operator is used outside consistent region and is configured with operator driven checkpointing
        // in this case, myLock_ is an re-entrant lock
        boost::recursive_mutex::scoped_lock am(myLock_);
    <% } else { %>
        // this operator is in consistent region, or outside consistent region but configured with periodic checkpointing
        // in this case, myLock_ is a regular SPL::Mutex
        AutoMutex am(myLock_);
    <% } %>
    // process tuple and manipulate operator state
    IPort0Type & ituple = static_cast<IPort0Type &>(tuple);
    myVar1_ += ituple.get_myInt();
    myVar2_ += ituple.get_myFloat();
 
    // perform operator driven checkpointing synchronously
    CheckpointContext * ckptContext = static_cast<CheckpointContext *>(this->getContext().getOptionalContext(CHECKPOINT));
    if (ckptContext != NULL && ckptContext->getKind() == CheckpointContext::operatorDriven) {
        ckptContext->createCheckpoint();
    }
}