parallelExecutorDataManager.h

Go to the documentation of this file.

//
// Copyright 2025 Pixar
//
// Licensed under the terms set forth in the LICENSE.txt file available at
// https://openusd.org/license.
//
#ifndef PXR_EXEC_VDF_PARALLEL_EXECUTOR_DATA_MANAGER_H
#define PXR_EXEC_VDF_PARALLEL_EXECUTOR_DATA_MANAGER_H
 
 
#include "pxr/pxr.h"
 
#include "pxr/exec/vdf/connection.h"
#include "pxr/exec/vdf/executorBufferData.h"
#include "pxr/exec/vdf/executorInvalidationData.h"
#include "pxr/exec/vdf/output.h"
#include "pxr/exec/vdf/parallelExecutorDataManagerInterface.h"
#include "pxr/exec/vdf/poolChainIndex.h"
#include "pxr/exec/vdf/types.h"
 
PXR_NAMESPACE_OPEN_SCOPE
 
class VdfMask;
class VdfVector;
 
template<typename DerivedClass> struct Vdf_ParallelExecutorDataManagerTraits;
 
template < typename DerivedClass >
class VdfParallelExecutorDataManager :
    public Vdf_ParallelExecutorDataManagerInterface<
        DerivedClass,
        typename Vdf_ParallelExecutorDataManagerTraits<DerivedClass>::DataHandle
        >
{
public:
 
    typedef
        typename Vdf_ParallelExecutorDataManagerTraits<DerivedClass>::DataHandle
        DataHandle;
 
    typedef
        Vdf_ParallelExecutorDataManagerInterface<DerivedClass, DataHandle>
        Base;
 
    VdfVector *GetInputValue(
        const VdfConnection &connection,
        const VdfMask &mask) const;
 
    VdfVector *GetOutputValueForReading(
        const DataHandle dataHandle,
        const VdfMask &mask) const;
 
    VdfVector *GetOrCreateOutputValueForWriting(
        const VdfOutput &output,
        const DataHandle dataHandle) const;
 
    void SetOutputValue(
        const VdfOutput &output,
        const VdfVector &value,
        const VdfMask &mask);
 
    bool TakeOutputValue(
        const VdfOutput &output,
        VdfVector *value,
        const VdfMask &mask);
 
    void SetReferenceOutputValue(
        const VdfVector *sourceValue,
        const VdfId destOutputId) const;
 
    static VdfVector *CreateOutputCache(
        const VdfOutput &output,
        VdfExecutorBufferData *bufferData);
 
    static VdfVector *CreateOutputCache(
        const VdfOutput &output,
        VdfExecutorBufferData *bufferData,
        const VdfMask::Bits &bits);
 
    void DuplicateOutputData(
        const VdfOutput &sourceOutput,
        const VdfOutput &destOutput);
 
    bool IsOutputInvalid(
        const VdfId outputId,
        const VdfMask &invalidationMask) const;
 
    bool InvalidateOutput(
        const VdfOutput &output,
        const VdfMask &invalidationMask);
 
    void Touch(const VdfOutput &output) const;
 
    void UpdateInvalidationTimestamp(
        const VdfInvalidationTimestamp &timestamp) {
        _invalidationTimestamp = timestamp;
    }
 
    const VdfInvalidationTimestamp &GetInvalidationTimestamp() const {
        return _invalidationTimestamp;
    }
 
    bool HasInvalidationTimestampMismatch(
        const DataHandle &sourceHandle,
        const DataHandle &destHandle) const;
 
protected:
 
    VdfParallelExecutorDataManager() :
        _invalidationTimestamp(
            VdfExecutorInvalidationData::InitialInvalidationTimestamp + 1)
    { }
 
    ~VdfParallelExecutorDataManager() {};
 
private:
 
    // Returns an output value for reading.
    //
    VdfVector *_GetOutputValueForReading(
        VdfExecutorBufferData *bufferData,
        const VdfMask &mask) const;
 
    // The current invalidation timestamp, recording the timestamp that was
    // applied to the last (if any) round of outputs traversed during 
    // invalidation.  This record and the timestamps on individual ExecutorData
    // objects are the keys to activating mung buffer locking.
    //
    VdfInvalidationTimestamp _invalidationTimestamp;
};
 
 
template<typename DerivedClass>
VdfVector *
VdfParallelExecutorDataManager<DerivedClass>::GetInputValue(
    const VdfConnection &connection,
    const VdfMask &mask) const
{
    // For associated inputs, we need to grab the input value from the
    // associated output. This is because read/write buffers have been
    // prepared before the node callback is invoked.
    // Values for read outputs originate from the source output on the
    // input connection.
 
    const VdfInput &input = connection.GetTargetInput();
    const VdfOutput *ao = input.GetAssociatedOutput();
 
    if (!ao || input.GetNumConnections() != 1) {
        return GetOutputValueForReading(
            Base::_GetDataHandle(connection.GetSourceOutput().GetId()), mask);
    }
 
    // Note that read/write output values are always passed via the private
    // buffers. The only time read/writes are published is when they reach
    // an output that no longer passes the value, or when a kept buffer is
    // being published. Therefore, we always read output values for read/writes
    // out of the private buffer, if requested as an input value.
 
    const DataHandle dataHandle = Base::_GetDataHandle(ao->GetId());
    return Base::_IsValidDataHandle(dataHandle)
        ? _GetOutputValueForReading(
            Base::_GetPrivateBufferData(dataHandle), mask)
        : nullptr;
}
 
template<typename DerivedClass>
VdfVector *
VdfParallelExecutorDataManager<DerivedClass>::GetOutputValueForReading(
    const DataHandle dataHandle,
    const VdfMask &mask) const
{
    // Output values are always read from public buffers. When nodes are
    // evaluated, they mutate the private buffers, which are then published
    // as soon as the node completes. Note, however, that read/write buffers
    // will never be published until they reach on output that no longer passes
    // the value.
 
    if (!Base::_IsValidDataHandle(dataHandle)) {
        return nullptr;
    }
 
    // Attempt to read from the public buffer, first.
    VdfExecutorBufferData *publicData = Base::_GetPublicBufferData(dataHandle);
    if (VdfVector *value = _GetOutputValueForReading(publicData, mask)) {
        return value;
    }
 
    // Then, fall back to reading from the transferred data, if available.
    VdfExecutorBufferData *transferData =
        Base::_GetTransferredBufferData(dataHandle);
    return transferData 
        ? _GetOutputValueForReading(transferData, mask)
        : nullptr;
}
 
template<typename DerivedClass>
VdfVector *
VdfParallelExecutorDataManager<DerivedClass>::GetOrCreateOutputValueForWriting(
    const VdfOutput &output,
    const DataHandle handle) const
{
    // If the handle is not valid, we can't return a value for writing.
    if (!Base::_IsValidDataHandle(handle)) {
        return nullptr;
    }
 
    // Note that output values are always written to private buffers. These
    // private buffers will then be published once the node has completed
    // evaluation.
 
    VdfExecutorBufferData *bufferData = Base::_GetPrivateBufferData(handle);
    if (VdfVector *value = bufferData->GetExecutorCache()) {
        return value;
    }
 
    // Create a new output value, if there isn't one already available.
    return CreateOutputCache(output, bufferData);
}
 
template < typename DerivedClass >
void
VdfParallelExecutorDataManager<DerivedClass>::SetOutputValue(
    const VdfOutput &output,
    const VdfVector &value,
    const VdfMask &mask)
{
    // Make sure the data manager is appropriately sized.
    Base::_Resize(output.GetNode().GetNetwork());
 
    // Mark the output as having been touched by evaluation, in order
    // for it to be considered by invalidation.
    const VdfId outputId = output.GetId();
    Base::_Touch(outputId);
 
    // Retrieve the data at the output
    const DataHandle handle = Base::_GetOrCreateDataHandle(outputId);
    VdfExecutorBufferData *privateBuffer = Base::_GetPrivateBufferData(handle);
    VdfExecutorBufferData *publicBuffer = Base::_GetPublicBufferData(handle);
 
    // Merge with existing data or replace?
    const VdfMask &publicMask = publicBuffer->GetExecutorCacheMask();
    const VdfVector *publicValue = publicBuffer->GetExecutorCache();
    const bool mergeData =
        publicValue &&
        !publicMask.IsEmpty() &&
        publicMask != mask;
 
    // Set the new output value, merging in the previously available public
    // data, if any.
    if (mergeData) {
        const VdfMask privateMask = publicMask | mask;
        VdfVector *outputValue = privateBuffer->CreateExecutorCache(
            output.GetSpec(), privateMask.GetBits());
        outputValue->Merge(*publicValue, publicMask - mask);
        outputValue->Merge(value, mask);
        privateBuffer->SetExecutorCacheMask(privateMask);
    } else {
        VdfVector *outputValue =
            privateBuffer->CreateExecutorCache(output.GetSpec());
        outputValue->Copy(value, mask);
        privateBuffer->SetExecutorCacheMask(mask);
    }
 
    // We just set the private buffer data, now let's publish it. Note, that
    // we can no longer modify the private or public buffer data after this
    // step, since it could lead to race conditions.
    Base::_PublishPrivateBufferData(handle);
}
 
template < typename DerivedClass >
bool
VdfParallelExecutorDataManager<DerivedClass>::TakeOutputValue(
    const VdfOutput &output,
    VdfVector *value,
    const VdfMask &mask)
{
    // Make sure the data manager is appropriately sized.
    Base::_Resize(output.GetNode().GetNetwork());
 
    // Mark the output as having been touched by evaluation, in order
    // for it to be considered by invalidation.
    const VdfId outputId = output.GetId();
    Base::_Touch(outputId);
 
    // Retrieve the data at the output
    const DataHandle handle = Base::_GetOrCreateDataHandle(outputId);
 
    // Attempt to transfer the value.
    return Base::_TransferBufferData(handle, value, mask);
}
 
template < typename DerivedClass >
void
VdfParallelExecutorDataManager<DerivedClass>::SetReferenceOutputValue(
    const VdfVector *sourceValue,
    const VdfId destOutputId) const
{
    const DataHandle handle = Base::_GetDataHandle(destOutputId);
    TF_DEV_AXIOM(Base::_IsValidDataHandle(handle));
 
    VdfExecutorBufferData *bufferData = Base::_GetPrivateBufferData(handle);  
 
    // XXX:cleanup This const_cast is real trouble.
    bufferData->YieldOwnership(const_cast<VdfVector *>(sourceValue));
}
 
template < typename DerivedClass >
VdfVector *
VdfParallelExecutorDataManager<DerivedClass>::CreateOutputCache(
    const VdfOutput &output,
    VdfExecutorBufferData *bufferData)
{
    TfAutoMallocTag2 tag(
        "Vdf", "VdfParallelExecutorDataManager::CreateOutputCache");
 
    // If the executor is providing its own cache-reuse mechanism, then
    // we assert that the cache is NULL before we get here.  Otherwise,
    // we will try to reuse whatever cache is there already.
    TF_DEV_AXIOM(!bufferData->GetExecutorCache());
    
    return bufferData->CreateExecutorCache(output.GetSpec());
}
 
template < typename DerivedClass >
VdfVector *
VdfParallelExecutorDataManager<DerivedClass>::CreateOutputCache(
    const VdfOutput &output,
    VdfExecutorBufferData *bufferData,
    const VdfMask::Bits &bits)
{
    TfAutoMallocTag2 tag(
        "Vdf", "VdfParallelExecutorDataManager::CreateOutputCache");
 
    // If the executor is providing its own cache-reuse mechanism, then
    // we assert that the cache is NULL before we get here.  Otherwise,
    // we will try to reuse whatever cache is there already.
    TF_DEV_AXIOM(!bufferData->GetExecutorCache());
 
    return bufferData->CreateExecutorCache(output.GetSpec(), bits);
}
 
template < typename DerivedClass >
void
VdfParallelExecutorDataManager<DerivedClass>::DuplicateOutputData(
    const VdfOutput &sourceOutput,
    const VdfOutput &destOutput)
{
    // Make sure the data manager is appropriately sized for us to copy
    // the source value to the destination output.
    Base::_Resize(destOutput.GetNode().GetNetwork());
 
    // Untouch the destination data, unless the source data has been touched.
    // This needs to happen even if we don't get a sourceHandle below.
    const VdfId destOutputId = destOutput.GetId();
    const VdfId sourceOutputId = sourceOutput.GetId();
    Base::_Untouch(destOutputId);
    if (Base::_Untouch(sourceOutputId)) {
        Base::_Touch(sourceOutputId);
        Base::_Touch(destOutputId);
    }
 
    // If the source output data exists, clone it to the destination
    // output data.
    const DataHandle sourceHandle = Base::_GetDataHandle(sourceOutputId);
    if (!Base::_IsValidDataHandle(sourceHandle)) {
        return;
    }
 
    // Get the destination data handle.
    const DataHandle destHandle = Base::_GetOrCreateDataHandle(destOutputId);
 
    // Clone the buffer data.
    Base::_GetPublicBufferData(sourceHandle)->Clone(
        Base::_GetPublicBufferData(destHandle));
 
    // Clone the invalidation data.
    Base::_GetInvalidationData(sourceHandle)->Clone(
        Base::_GetInvalidationData(destHandle));
 
    // Copy the invalidation timestamp.
    Base::_SetInvalidationTimestamp(
        destHandle, Base::_GetInvalidationTimestamp(sourceHandle));
}
 
template < typename DerivedClass >
bool
VdfParallelExecutorDataManager<DerivedClass>::IsOutputInvalid(
    const VdfId outputId,
    const VdfMask &invalidationMask) const
{
    // If the output has been touched by evaluation, it is valid.
    const bool wasTouched = Base::_IsTouched(outputId);
    if (wasTouched) {
        return false;
    }
 
    // If there is no data handle for the output, the output has never been
    // evaluated and therefore is still invalid.
    const DataHandle handle = Base::_GetDataHandle(outputId);
    if (!Base::_IsValidDataHandle(handle)) {
        return true;
    }
 
    // Let's check if the invalidation data is marked invalid for the given
    // mask.
    return Base::_GetInvalidationData(handle)->IsInvalid(
        invalidationMask, wasTouched);
}
 
template < typename DerivedClass >
bool 
VdfParallelExecutorDataManager<DerivedClass>::InvalidateOutput(
    const VdfOutput &output,
    const VdfMask &invalidationMask)
{
    // Untouch the output. If it has been touched, we need to creata a data
    // handle for it, even though it may have never been evaluated with this
    // data manager.
    const VdfId outputId = output.GetId();
    const bool wasTouched = Base::_Untouch(outputId);
 
    // Retrieve the data handle for the output.
    const DataHandle handle = wasTouched
        ? Base::_GetOrCreateDataHandle(outputId)
        : Base::_GetDataHandle(outputId);
 
    // Thou shalt not invalidate what has not been evaluated (unless it has
    // been touched, of course)!
    if (!Base::_IsValidDataHandle(handle)) {
        return false;
    }
    
    // Invalidate the output via the VdfExecutorInvalidationData. Make sure
    // to also untouch the output, if it has previously been touched by
    // evaluation.
    const bool didInvalidate = Base::_GetInvalidationData(handle)->Invalidate(
        invalidationMask, wasTouched);
 
    // If the output had previously been touched, and it has now been
    // invalidated. Make sure to invalidate the VdfExecutorBufferData.
    if (didInvalidate) {
 
        // Update the invalidation timestamp, by applying the timestamp from
        // the data manager to the timestamp stored at the output.
        Base::_SetInvalidationTimestamp(handle, GetInvalidationTimestamp());
 
        // Retrieve the buffer data stored at the output.
        VdfExecutorBufferData *bufferData = Base::_GetPublicBufferData(handle);
 
        // We only need to invalidate the executor cache if the buffer has one.
        if (!bufferData->GetExecutorCacheMask().IsEmpty()) {
 
            // If this is an output in the pool, let's apply sparse
            // invalidation.
            if (Vdf_IsPoolOutput(output)) {
 
                // Sparsely invalidate the executor cache mask, using the bits
                // in the invalidation mask. During a steady-state mung, the
                // cache and invalidation masks will likely always be the same
                // across iterations, so we memoize this operation.
                //
                // XXX: Memoize!
                //
                VdfMask newCacheMask =
                    bufferData->GetExecutorCacheMask() - invalidationMask;
 
                // If the new cache mask is now all-zeros, remove the cache
                // entirely, otherwise simply set the new cache mask.
                if (newCacheMask.IsAllZeros()) {
                    bufferData->ResetExecutorCache();
                } else {
                    bufferData->SetExecutorCacheMask(newCacheMask);
                }
            }
 
            // Otherwise, we simply remove the cache entirely.
            else {
                bufferData->ResetExecutorCache();
            }
        }
 
        // Also clear any transferred data.
        Base::_ResetTransferredBufferData(handle);
 
        // We did some invalidation.
        return true;
    }
    
    // Nothing to invalidate.
    return false;
}
 
template < typename DerivedClass >
void 
VdfParallelExecutorDataManager<DerivedClass>::Touch(
    const VdfOutput &output) const
{
    Base::_Touch(output.GetId());
}
 
template < typename DerivedClass >
bool
VdfParallelExecutorDataManager<DerivedClass>::HasInvalidationTimestampMismatch(
    const DataHandle &sourceHandle,
    const DataHandle &destHandle) const
{
    // For this method to return true, indicating that the source output should
    // be locked for mung buffer locking, the invalidation timestamp stored in
    // the destination data object must match the current invalidation
    // timestamp. Furthermore, the source data object must have an invalidation
    // timestamp different from our current invalidation timestamp.
    // Essentially, this means that in the latest round of invalidation, the
    // destination output has been invalidated, whereas the source output
    // remained untouched.
    // If this method returns true, the cache at the source output can be
    // mung buffer locked, because it won't receive invalidation, whereas any
    // output below (and including) the destination output will!
 
    return
        Base::_IsValidDataHandle(sourceHandle) &&
        Base::_IsValidDataHandle(destHandle) &&
        Base::_GetInvalidationTimestamp(sourceHandle) !=
            _invalidationTimestamp &&
        Base::_GetInvalidationTimestamp(destHandle) == 
            _invalidationTimestamp;
}
 
template < typename DerivedClass >
VdfVector *
VdfParallelExecutorDataManager<DerivedClass>::_GetOutputValueForReading(
    VdfExecutorBufferData *bufferData,
    const VdfMask &mask) const
{
    // We have the output value if
    //  o The output data exists
    //  o The cache is not empty
    //  o One of the following is true:
    //     o The request mask has no bits set because we ask for an
    //       attribute with shape of length zero (e.g., points attribute
    //       with zero points in it.
    //    or
    //     o The output is not dirty and
    //     o The computed mask covers what is requested.
 
    VdfVector *value = bufferData->GetExecutorCache();
    const bool hasValue =
        value &&
        (mask.IsAllZeros() ||
            (bufferData->GetExecutorCacheMask().IsAnySet() &&
             bufferData->GetExecutorCacheMask().Contains(mask)));
 
    return hasValue ? value : nullptr;
}
 
PXR_NAMESPACE_CLOSE_SCOPE
 
#endif