PDG_GraphContext.h

Go to the documentation of this file.

/*
 * PROPRIETARY INFORMATION.  This software is proprietary to
 * Side Effects Software Inc., and is not to be reproduced,
 * transmitted, or disclosed in any way without written permission.
 *
 * COMMENTS:
 */

#ifndef __PDG_GRAPH_CONTEXT_H__
#define __PDG_GRAPH_CONTEXT_H__

#include "PDG_API.h"

#include "PDG_ApplicationShim.h"
#include "PDG_CheckpointManager.h"
#include "PDG_CommandManager.h"
#include "PDG_CookState.h"
#include "PDG_EventEmitter.h"
#include "PDG_EventQueue.h"
#include "PDG_Graph.h"
#include "PDG_Types.h"

#include "PDG_WorkItemSerialization.h"

#include <PDGE/PDGE_Evaluator.h>
#include <PDGT/PDGT_ValueArgs.h>

#include <UT/UT_Lock.h>
#include <UT/UT_Map.h>
#include <UT/UT_NonCopyable.h>
#include <UT/UT_Set.h>
#include <UT/UT_StringArray.h>
#include <UT/UT_StringHolder.h>
#include <UT/UT_StringMap.h>
#include <UT/UT_TBBSpinLock.h>
#include <UT/UT_UniquePtr.h>
#include <UT/UT_ValArray.h>
#include <UT/UT_VectorTypes.h>

#include <SYS/SYS_AtomicInt.h>

class PDG_CommandChunk;
class PDG_Command;
struct PDG_CookOptions;
class PDG_InProcessScheduler;
struct PDG_MemoryInfo;
class PDG_Scheduler;
class PDG_SchedulerType;
class PDG_ServiceScheduler;
class PDG_WorkItem;

class UT_Thread;
class UT_WorkBuffer;

/*
 * Stores core components of a PDG graph, such as the node graph itself, the
 * schedulers and the event queue
 */
class PDG_API PDG_GraphContext : public PDG_EventEmitter, public PDGE_Evaluator
{
public:
    using SchedulerMap = UT_StringMap<PDG_SchedulerPtr>;

    /*
     * Observer that gets called each time a graph context is created or
     * destroyed
     */
    class PDG_API Observer
    {
    public:
                        Observer() {}
        virtual         ~Observer() {}
                        UT_NON_COPYABLE(Observer)

        virtual void    onPDGGraphContextRegistered(
                            PDG_GraphContext *context) = 0;
        virtual void    onPDGGraphContextDeregistered(
                            PDG_GraphContext *context) = 0;
    };

public:
    /// Returns the context with the specified name, or nullptr if no such
    /// context exists
    static PDG_GraphContext*        contextByName(const UT_StringHolder& name);

    /// Returns the context with the specified ID, or nullptr if no such
    /// context exists
    static PDG_GraphContext*        contextByID(int id);

    /// Returns the list of all registered context names
    static UT_StringArray           contextNames();

    /// Returns the list of all registered context IDs
    static UT_IntArray              contextIDs();

    /// Adds and removes a global observer, which is notified each time a
    /// graph context is created
    static void                     addGraphContextObserver(Observer *observer);
    static void                     removeGraphContextObserver(Observer *observer);

    /// Constructs a new context with the specified name and id. It is up to
    /// the caller to supply a unique id, which is typically set to an
    /// application-specific value such as the unique ID of a TOP_Network.
                                    PDG_GraphContext(
                                        const char* name="context",
                                        int id=-1);
                                    ~PDG_GraphContext() override;

    /// Returns the list of supported event types
    PDG_EventEmitter::Filter&       supportedEventTypes() const override;

    /// Returns total memory usage of the graph and schedulers contained in
    /// this context
    int64                           getMemoryUsage(
                                        bool inclusive) const override;

    /// Returns a fined grained breakdown of memory used by the nodes and 
    /// work items in the graph context
    void                            memoryInfo(
                                        PDG_MemoryInfo& memory_info,
                                        bool inclusive) const;

    /// Sets the application-specific graph object
    void                            setAppGraph(PDG_ApplicationShim::Graph* graph)
                                        { myAppGraph = graph; }

    /// The graph owned by this instance
    PDG_Graph*                      graph()
                                        { return &myGraph; }

    /// Returns the event queue associted with this context
    PDG_EventQueue*                 getEventQueue()
                                        { return myEventQueue.get(); }

    /// Returns the checkpoint manager associated with this instance
    PDG_CheckpointManager*          checkpointManager()
                                        { return &myCheckpointManager; }

    /// The unique name of the graph context
    const UT_StringHolder&          name() const
                                        { return myName; }

    /// Returns the id of the graph context
    int                             id() const
                                        { return myId; }

    /// Returns the default work item label
    bool                            defaultWorkItemLabel(
                                        UT_StringHolder& label) const;

    /// Returns the current cook state object, which stores various
    /// metadata relating to the active grap hook
    const PDG_CookState&            cookState() const
                                        { return myCookState; }

    /// Returns true if the context is actively cooking
    bool                            cooking() const
                                        { return myCooking; }

    /// Returns true if the context is processing a cancel request from the
    /// user
    bool                            canceling() const
                                        { return myCanceling; }


    /// Returns the current default scheduler, or nullptr if no schedulers
    /// exist. The first scheduled created is automatically set as the
    /// default.
    PDG_Scheduler*                  defaultScheduler() const
                                        { return myDefaultScheduler; }

    /// Returns the service scheduler, which is used to process work items
    /// from nodes with service mode enabled.
    PDG_ServiceScheduler*           serviceScheduler() const
                                        { return myServiceScheduler; }

    /// Returns the default in process scheduler, which is used to process
    /// in process items that need to run on the main thread
    PDG_InProcessScheduler*         inProcessScheduler() const
                                        { return myInProcessScheduler; }

    /// Adds a new scheduler instance to the graph context
    void                            addScheduler(
                                        const UT_StringHolder& name,
                                        PDG_SchedulerPtr scheduler);

    /// Adds a new scheduler constructed from the specified type instance,
    /// and with the desired named. This method ensures that the name is
    /// unique.
    PDG_Scheduler*                  addScheduler(
                                        const UT_StringHolder& preferred_name,
                                        PDG_SchedulerType* scheduler_type,
                                        UT_WorkBuffer& errors);

    /// Removes the scheduler with the specified name, and returns true if
    /// it was found/removed.
    bool                            removeScheduler(
                                        const UT_StringHolder& name);

    /// Sets the specified scheduler as the new default.
    void                            setDefaultScheduler(
                                        PDG_Scheduler* scheduler);

    /// Returns the scheduler with the specified name, or nullptr if no such
    /// scheduler exists.
    PDG_Scheduler*                  schedulerForName(
                                        const UT_StringHolder& name) const;

    /// Returns the map of scheduler name -> scheduler pointer that exist
    /// in the graph context
    const SchedulerMap&             schedulers() const;

    /// Returns shared server info by searching for it across all active
    /// schedulers
    PDGT_ValueArgs                  getSharedServerInfo(
                                        const UT_StringHolder &sharedserver_name);

    /// Cleans the temporary file directory for the specified scheduler
    void                            cleanTempDirectory(
                                        const UT_StringHolder& scheduler_name);

    /// Cancel the cook if it's running and block until it's finished canceling
    void                            cancelCook();

    /// Similar to cancel, except work items that are in-progress are not
    /// killed, and results can be collected later.
    void                            pauseCook();

    /// Cooks the graph with the specified options. Returns false if there was
    /// a probably starting the cook
    bool                            cook(
                                        const PDG_CookOptions& options,
                                        UT_WorkBuffer& errors);


    /// Returns the current open command chunk
    PDG_CommandChunk*               currentChunk() const;

    /// Opens a new command chunk with an option description string
    PDG_CommandChunk*               openChunk(UT_StringHolder description = "");

    /// Opens a new delayed command chunk
    PDG_CommandChunk*               openChunk(
                                        bool delayed,
                                        const UT_StringHolder& description);

    /// Record, undo and redo commands
    bool                            commitChunk(UT_WorkBuffer& errors);
    bool                            rollbackChunk(UT_WorkBuffer& errors);
    bool                            undoIt(UT_WorkBuffer& errors);
    bool                            redoIt(UT_WorkBuffer& errors);
    bool                            recordCommand(
                                        UT_UniquePtr<PDG_Command> cmd,
                                        UT_WorkBuffer& errors);

    /// Returns the number of entries in the command manager
    int                             stateCount() const;

    /// Returns the number of entries in the current command chunk
    int                             chunkDepth() const;

    /// Dumps the current command manager contents to the input buffer
    void                            commandDescriptions(
                                        UT_WorkBuffer& buffer) const;

    /// Serializes the node graph to a buffer
    void                            serializeGraph(UT_WorkBuffer& buffer);
    void                            serializeGraph(
                                        const UT_StringHolder& file_name,
                                        UT_StringHolder extras = "");

    /// Serializes the work item graph to a buffer
    void                            serializeWorkItems(UT_WorkBuffer& buffer);
    void                            serializeWorkItems(
                                        const UT_StringHolder& file_name,
                                        UT_StringHolder extras = "");

    /// Serializes both the work item and node graph
    void                            serialize(UT_WorkBuffer& buffer);
    void                            serialize(
                                        const UT_StringHolder& file_name,
                                        UT_StringHolder extras = "");

    /// Serializes the specified work item to JSON
    UT_StringHolder                 serializeWorkItemToJSON(
                                        PDG_WorkItem* work_item,
                                        bool pretty_print) const;

    /// Loads a work item from JSON
    PDG_WorkItem*                   deserializeWorkItemFromJSON(
                                        const UT_StringHolder& buffer,
                                        UT_WorkBuffer& errors);

    /// Loads a work item checkpoint file
    bool                            deserializeCheckpoint(
                                        const UT_StringHolder& file_path,
                                        UT_WorkBuffer& errors);

    /// Begins deserializing a work item graph, part of the Python
    /// serialization API
    void                            beginDeserialization();

    /// Deserializes a work item from a serialization struct
    void                            addWorkItem(
                                        const PDG_WorkItemSerialization& serialization);

    /// Adds a dependency between two work items, given their names
    void                            addWorkItemDependency(
                                        const UT_StringHolder& item_name,
                                        const UT_StringHolder& dep_name,
                                        bool required,
                                        bool user_defined);

    /// Adds a dependency between two work items, given their IDs
    void                            addWorkItemDependencyId(
                                        PDG_WorkItemID item_id,
                                        PDG_WorkItemID dep_id,
                                        bool required,
                                        bool user_defined);

    /// Adds a propagate dependency between two work iems, given their names
    void                            addWorkItemPropagateDep(
                                        const UT_StringHolder& item_name,
                                        const UT_StringHolder& dep_name) {}

    /// Adds a propagate dependency between two work iems, given their IDs
    void                            addWorkItemPropagateDepId(
                                        PDG_WorkItemID item_id,
                                        PDG_WorkItemID dep_id) {}

    /// Adds work item output files onto the specified work item, by name
    void                            addWorkItemResults(
                                        const UT_StringHolder& item_name,
                                        const UT_StringArray& results,
                                        const UT_StringArray& tags,
                                        const UT_Array<int64>& hash_codes,
                                        const UT_Array<bool>& is_expected,
                                        const UT_Array<bool>& is_owned);

    /// Adds work item output files onto the specified work item, by id
    void                            addWorkItemResultsId(
                                        PDG_WorkItemID item_id,
                                        const UT_StringArray& results,
                                        const UT_StringArray& tags,
                                        const UT_Array<int64>& hash_codes,
                                        const UT_Array<bool>& is_expected,
                                        const UT_Array<bool>& is_owned);

    /// Commits a partially deserialized work item to the PDG graph
    void                            commitWorkItem(
                                        const UT_StringHolder& item_name);
    void                            commitWorkItemId(PDG_WorkItemID item_id);

    /// Runs a function with the context lock held, for a specific graph 
    /// context. If a context with the name does not exist, the function is
    /// not called.
    template <typename Func>
    static void                     safeGraphContextAccess(
                                        const UT_StringHolder& name,
                                        const Func& func)
                                        {
                                            UT_Lock::Scope auto_lock(
                                                theContextLock);
                                            auto it = theContextMap.find(name);
                                            if (it != theContextMap.end())
                                                func(it->second);
                                        }

    /// Runs a function over all graph contexts with the context lock held
    template <typename Func>
    static void                     safeGraphContextAccess(const Func& func)
                                        {
                                            UT_Lock::Scope auto_lock(
                                                theContextLock);
                                            for (auto&& entry : theContextMap)
                                                func(entry.second);
                                        }
                        
    /// Runs a function with the scheduler lock held
    template<typename Func>
    void                            safeSchedulerAccess(
                                        const UT_StringHolder& name,
                                        const Func& func)
                                        {
                                            UT_Lock::Scope schedulersLock(
                                                mySchedulersLock);

                                            auto it = mySchedulers.find(name);
                                            if (it != mySchedulers.end())
                                                func(it->second.get());
                                        }

protected:
    bool                            preEvaluation() override;
    bool                            tickEvaluation() override;
    void                            postEvaluation(
                                        PDGE_Evaluator::State state,
                                        bool canceled) override;

private:
    static void                     registerContext(PDG_GraphContext* context);
    static void                     deregisterContext(PDG_GraphContext* context);
    static void                     onExit(void* data);
    static void                     onPyExit();
    static void*                    doTickThread(void* data);

    void                            commitWorkItemInternal(
                                        PDG_WorkItem* work_item);
    void                            addWorkItemDependencyInternal(
                                        PDG_WorkItem* work_item,
                                        PDG_WorkItem* dependency,
                                        bool required,
                                        bool user_defined);
    void                            addWorkItemResultsInternal(
                                        PDG_WorkItem* work_item,
                                        const UT_StringArray& results,
                                        const UT_StringArray& tags,
                                        const UT_Array<int64>& hash_codes,
                                        const UT_Array<bool>& is_expected,
                                        const UT_Array<bool>& is_owned);

    void                            serializeWorkItem(
                                        UT_WorkBuffer& buffer,
                                        PDG_WorkItemIDMap& serialization_map,
                                        const PDG_WorkItem* item) const;
    void                            serializeWorkItemCommit(
                                        UT_WorkBuffer& buffer,
                                        PDG_WorkItemIDMap& serialization_map,
                                        const PDG_WorkItem* item) const;

    PDG_GraphContext*               context()
                                        { return this; }

private:
    using ContextMap    = UT_StringMap<PDG_GraphContext*>;
    using ObserverSet   = UT_Set<Observer*>;

    static ContextMap               theContextMap;
    static UT_Lock                  theContextLock;
    static ObserverSet              theObservers;
    static UT_TBBSpinLock           theObserverLock;
    static bool                     theRegisteredAtExit;

    UT_StringHolder                 myName;
    int                             myId;

    PDG_Graph                       myGraph;
    PDG_CommandManager              myUserCmdManager;
    PDG_WorkItemSerialization::Map  mySerializationMap;

    PDG_CheckpointManager           myCheckpointManager;

    mutable UT_Lock                 mySchedulersLock;
    SchedulerMap                    mySchedulers;

    PDG_Scheduler*                  myDefaultScheduler;
    PDG_ServiceScheduler*           myServiceScheduler;
    PDG_InProcessScheduler*         myInProcessScheduler;

    UT_Thread*                      myTickThread;
    PDG_ApplicationShim::Graph*     myAppGraph;

    PDG_CookState                   myCookState;

    bool                            myCooking;
    bool                            myCanceling;
    SYS_AtomicInt32                 myTickActive;

    UT_UniquePtr<PDG_EventQueue>    myEventQueue;
};

#endif /* __PDG_GRAPH_CONTEXT_H__ */