19 \$cache functions using DGAsyncNodeOnInput

Parameterless $cache functions are treated specially because there is no need for a cache map. Instead they are more like an individual async dependent variable.

Consider the following example:


@import "Ceda/cxObject/DGAsyncNode.h"

$struct X isa ceda::IObject
{
    $cache <<async>> int y() const
        with int x = 7;
    {
        return x;
    }
};

Xcpp generates the following C++ code:

struct X : public ceda::IObjectBaseMixin<X,ceda::EmptyBase>
{
    typedef ceda::IObjectBaseMixin<X,ceda::EmptyBase> BaseClass;

    void _calc_y_input(int& x) const
    {
        x = 7;
    }

    int _calc_y(int x) const
    {
        return x;
    }

    struct _depnode_y : public ceda::DGAsyncNodeOnInput<_depnode_y,X,int,int>
    {
        typedef ceda::DGAsyncNodeOnInput<_depnode_y,X,int,int> base;
        using typename base::output_type;
        using typename base::input_type;
        static void CalcOutput(const X* _self, output_type& _output, const input_type& _input)
        {
            _output = _self->_calc_y(_input);
        }
        static void CalcInput(const X* _self, input_type& _input)
        {
            _self->_calc_y_input(_input);
        }
        X const* _GetSelf() const { return CONST_ATTRIB_CAST(X,_dn_y); }
        X const* _GetFinalSelf() const { return _GetSelf(); }
        virtual ceda::xstring Name() const { return "X::y()"; }
    } _dn_y;

    int const& y() const
    {
        return _dn_y.read();
    }

    void EvictDgsNodes() const
    {
        BaseClass::EvictDgsNodes();
        _dn_y.TryEvict(false);
    }
};

There is a nested struct named _depnode_y which represents an asynchronously calculated variable.

_depnode_y is a subclass of ceda::DGAsyncNodeOnInput which is defined in DGAsyncNode.h

Policy

We implement the following policy (there are others that may be appropriate in situations but let's do this one):

We never abort an async calculation.
We always apply the new output when the async calculation is finished.
We don't allow two async calculations to occur in parallel.
When finishing the async calculation we check whether we are dirty and if so we recalculate again.

This is a good default because it has the following nice features:

there is no starvation (which can happen if tasks are aborted before they are finished, and they never get a chance to complete)
no aborting tasks
no risk of creating thousands of tasks
only one input needs to be created and visited

DGAsyncNodeOnInput

template <typename FinalClass, typename Self, typename Input, typename Output>
class DGAsyncNodeOnInput : public DGDepNode
{
public:
    typedef Input input_type;
    typedef Output output_type;

    DGAsyncNodeOnInput() :
        DGDepNode(DF_ASYNC)
    {
    }

    virtual ssize_t ByteSize() const
    {
        // Note that we don't add CacheValueAdditionalSize(input_) because we clear the input
        // at the time the output is updated
        return CacheValueAdditionalSize(_output);
    }

    virtual void VisitContainingObject(IObjectVisitor& _v) const
    {
        _v << input_ << _output << static_cast<const FinalClass*>(this)->_GetFinalSelf();
    }

    virtual void OnEvict() const
    {
        OnEvictCacheValue(_output);
        ClearCacheValue(_output);
    }

    virtual bool IsEvictable() const
    {
        return !Enabled(DF_ASYNC_RUNNING) && CacheValueIsEvictable(_output);
    }

    virtual bool RecalcCache() const
    {
        FinalClass::CalcInput(static_cast<const FinalClass*>(this)->_GetSelf(), input_);
        return false;   // Pretend input not changed to avoid propagating dirtiness into the out-nodes
    }

    virtual void OnInvalidate() const
    {
        if (Enabled(DF_ASYNC_RUNNING))
        {
            // Already running an async task, don't post another
        }
        else
        {
            SetFlag(DF_ASYNC_RUNNING);
            PostAsyncTask(GetThreadPtr<CSpace>(), [this]()
            {
                PrepareThreadLocalStorage(static_cast<const FinalClass*>(this)->_GetFinalSelf());

                // Calculate the inputs
                {
                    CSpaceTxn txn;
                    cxAssert(Enabled(DF_ASYNC_RUNNING));
                    ReadBarrierDep();               // This cleans this node
                }

                while(1)
                {
                    // Calculate output from input without a CSpace lock, hold output in a local variable
                    output_type localOutput;
                    FinalClass::CalcOutput(static_cast<const FinalClass*>(this)->_GetSelf(), localOutput, input_);

                    // Apply the output and invalidate out-nodes
                    {
                        CSpaceTxn txn;
                        cxAssert(Enabled(DF_ASYNC_RUNNING));

                        UpdateOutput(_output, localOutput);
                        input_ = input_type();          // Clear the input - might allow some memory reclamation

                        MarkOutNodesAsSoftAndHardDirty();

                        if (!Enabled(DF_SOFT_DIRTY))
                        {
                            // Finished
                            ClearFlag(DF_ASYNC_RUNNING);
                            if (TypeHasAdditionalSize<output_type>()) UpdateByteSize();
                            return;
                        }
                        else
                        {
                            // This DGS node is dirty indicating that the inputs have changed again, so we need to recalculate again

                            // Recalculate the input. This cleans this node
                            ReadBarrierDep();

                            // continue in the while loop
                        }
                    }
                }
            });
        }
    }

    const output_type& read() const
    {
        if (!Enabled(DF_CALLED_READ))
        {
            SetFlag(DF_CALLED_READ);
            OnInvalidate();
        }
        ReadBarrierIndep();      // Attach out-nodes
        return _output;
    }

protected:
    ///////////////////////////// Protected by a CSpace lock
    mutable input_type input_;
    mutable output_type _output;
};

Comments

An async dep node is kind of like an dep+indep pair, where the dep node has no out-nodes. Of course the indep node has in-nodes. This suggests statewise it makes sense to use one object for both. Indeed it is like a normal dep node except

from the point of view of callers it is more like an indep node. For that purpose we have a ReadBarrierIndep() function which is like an indep node read barrier and which is used to attach the out-nodes.
When updating the output it is like an indep node, in that we call MarkOutNodesAsSoftAndHardDirty() to invalidate the out-nodes.
from the point of view of the async task which wants to calculate the inputs in a CSpaceLock, it behaves more like a dep node, in that it allows for attaching in-nodes.
Unlike a normal dep node, when we are invalidated because the inputs change we don't propagate dirtiness into the out-nodes.

This is what needs to happen:

The containing class has a task executer member for posting tasks for this object. The task executer is closed when the containing object destructs. This usually happens without a CSpace lock to avoid dead-locks.
The cache map is initially empty. There is no concept of "kickstarting" by posting any tasks
If foo(a,b) is called for the first time then
- a subclass of DGDepNode is inserted into the map. This holds the initial output value which is None.
- Invoke ReadBarrierIndep() in the manner of an indep node, in order to attach the out-nodes.
- Call OnInvalidate() to post a task, just as though the inputs had changed
- return _output
When the task runs, ReadBarrierDep() calls RecalcCache() which calculates the inputs, and the in-nodes are attached as required.
If the node is invalidated then unlike a normal dep node we don't propagate dirtiness into the out-nodes.

Previous18 \$cache functions using DGKeyedAsyncIndepNode Next20 \$cache functions using DGKeyedAsyncNodeOnInput