PDG_Scheduler.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_SCHEDULER_H__
#define __PDG_SCHEDULER_H__

#include "PDG_API.h"

#include "PDG_BatchWorkItem.h"
#include "PDG_CookOptions.h"
#include "PDG_EventEmitter.h"
#include "PDG_File.h"
#include "PDG_FileUtils.h"
#include "PDG_NodeInterface.h"
#include "PDG_Types.h"
#include "PDG_WorkItemSort.h"

#include <PDGT/PDGT_ValueArgs.h>

#include <UT/UT_ConcurrentPriorityQueue.h>
#include <UT/UT_ConcurrentHashMap.h>
#include <UT/UT_ConcurrentQueue.h>
#include <UT/UT_ConcurrentSet.h>
#include <UT/UT_StringArray.h>
#include <UT/UT_StringHolder.h>
#include <UT/UT_TBBSpinLock.h>

#include <time.h>

class PDG_CookState;
class PDG_GraphContext;
class PDG_Node;
class PDG_Service;
class PDG_WorkItem;

namespace PDGN
{
class PDGN_PollingClientNNG;
}

/*
 * Base class for scheduler nodes, which process work items and run them
 * either in process or out of process
 */
class PDG_API PDG_Scheduler : public PDG_NodeInterface,
                              public PDG_EventEmitter,
                              public PDG_TypeInstance,
                              public PDGE_DependencyOwner
{
public:
    /*
     * Enumeration of possible return values from onScheduler callback
     */
    enum ScheduleResult
    {
        /// The work item was cooked, and should be marked as canceled
        eCookCanceled,

        /// The work item was cooked, and should be marked as failed
        eCookFailed,

        /// The work item was cooked, and should be marked as succeeded
        eCookSucceeded,

        /// The call to schedule the work item failed
        eFailed,

        /// The call to schedule the work item succeeded
        eSucceeded,

        /// The scheduler cannot handle the work item at this time
        eDeferred,

        /// The scheduler cannot handle any work items at this time, and
        /// all further scheduling requests will be deferred
        eFullDeferred
    };

    /*
     * Enumeration of possible return values from onTick callback
     */
    enum TickResult
    {
        /// The scheduler is ready to accept work items
        eSchedulerReady,

        /// The scheduler is busy, and cannot process additional work items
        eSchedulerBusy,

        /// The scheduler has encountered some sort of fatal error or request
        /// to cancel, and will not longer accept work items for the remainder
        /// of the cook.
        eSchedulerCancelCook
    };

    /*
     * Enumeration of possible return values from onAcceptWorkItem callback
     */
    enum AcceptResult
    {
        /// The scheduler is able to handle the specified work item
        eSchedulerAccept,

        /// The scheduler is not able to handle the specified work item
        eSchedulerReject,

        /// The scheduler cannot tell if can/can't schedule the specified
        /// work item, and the default logic for checking should be used
        /// instead
        eSchedulerDefault
    };

    /*
     * Enumeration of path mapping modes available on the scheduler
     */
    enum PathMappingMode
    {
        /// Use the default, global path mapping
        eDefaultPathMapping,

        /// Disable path mapping
        eNoPathMapping
    };

    /*
     * Enumeration of work item data transfer modes
     */
    enum WorkItemDataSource
    {
        /// Work item data is transfered to the farm using a shared
        /// drive, by writing it to .json file on disk.
        eSourceJSONFile,

        /// Work item data is transfered via RPC
        eSourceRPCMessage
    };

    /*
     * Enumeration of temp dir cleanup options
     */
    enum TempDirCleanup
    {
        /// The temp file dir is not cleaned up automatically
        eCleanupNone,

        /// The temp file dir is cleaned up when the scheduler is removed or
        /// the session shuts down
        eCleanupShutdown,

        /// The temp file dir is cleaned up after the scheduler finishes
        /// cooking
        eCleanupCook
    };

public:
                            PDG_Scheduler(const PDG_BaseType* type,
                                          const PDGT_ValueArgs& extra_args,
                                          const UT_StringHolder& name);
                            ~PDG_Scheduler() override;

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

    int64                   getMemoryUsage(bool inclusive) const override;

    /// Debug name method, required by PDGE_DependencyOwner
    UT_StringHolder         debugName() const override
                                { return name(); }

    /// Resets the dependency owner
    void                    resetOwner() override;

    /// Returns the is cooked dep for this scheduler
    inline PDGE_Dependency* isCookedDep()
                                { return &myIsCookedDep; }

    /// Returns the name of the shceudler
    const UT_StringHolder&  name() const
                                { return myName; }

    /// Called when the type object should reload itself because of a change
    /// to the underlying Python module
    bool                    reloadInstance(UT_WorkBuffer& errors) override;

    /// Returns true if the scheduler is able to queue the specified work item,
    /// else false.
    bool                    canSchedule(const PDG_WorkItem* work_item);

    /// Called when the scheduler should process a static dependency graph
    virtual void            onScheduleStatic(
                                const PDG_WorkItemMap& dependencies,
                                const PDG_WorkItemMap& dependents,
                                const PDG_WorkItemArray& ready_items) = 0;

    /// Called whenever the scheduler should schedule a work item. This is a 
    /// required callback.
    virtual ScheduleResult  onSchedule(PDG_WorkItem* work_item) = 0;

    /// Called when a specific work item should be canceled
    virtual void            onCancelWorkItems(
                                const PDG_WorkItemArray& work_item,
                                PDG_Node* node) = 0;

    /// Called when the scheduler is first started, should be used to initialize
    /// scheduler state, threads, etc. This is an optional callback.
    virtual bool            onStart()  = 0;

    /// Called when the scheduler is stopped, to clean up any resources held
    /// by the scheduler (sockets, threads, etc). Optional callback.
    virtual bool            onStop() = 0;

    /// Called when cooking begins
    virtual bool            onStartCook(
                                bool static_cook,
                                const PDG_NodeSet&) = 0;

    /// Called when cooking completes or is canceled
    virtual bool            onStopCook(bool cancel) = 0;

    /// Called once at the start of the cook, to allow the scheduler to 
    /// set any custom cook options
    virtual void            onConfigureCook(PDG_CookOptions* cook_ptions) = 0;

    /// Called periodically during cooking to update state
    virtual TickResult      onTick() = 0;

    /// Returns an enum indicating whether or not the scheduler accepts the
    /// specified work item
    virtual AcceptResult    onAcceptWorkItem(
                                const PDG_WorkItem* work_item) = 0;

    /// Custom file transfer logic, implemented per-scheduler
    virtual bool            onTransferFile(const UT_StringHolder& file_path);

    /// Transfers a file from the local machine to the machine that will be
    /// executing processor tasks, e.g. a remote machine on the farm. This
    /// callback is required since the main PDG code depends on its existence.
    /// At a later point we may be able to make this optional.
    bool                    transferFile(
                                const UT_StringHolder& file_path,
                                PDG_FileUtils::TransferType default_type,
                                UT_WorkBuffer& errors) const;

    /// Returns the path on the scheduler that a local file will be copied 
    /// to, using the standard file transfer mechanism
    UT_StringHolder         formatTransferPath(
                                const UT_StringHolder& local_path,
                                PDG_FileUtils::TransferType default_type,
                                bool local) const;

    /// Returns the local version of the given path.
    UT_StringHolder         localizePath(
                                const UT_StringHolder& deloc_path) const;

    /// Returns the delocalized (remote) version of the given local path.
    UT_StringHolder         delocalizePath(
                                const UT_StringHolder& local_path) const;

    /// Returns the job name for the specified work item
    UT_StringHolder         jobName(const PDG_WorkItem* work_item) const;

    /// Expands special tokens in the command string
    virtual UT_StringHolder expandCommandTokens(
                                const UT_StringHolder& command,
                                const PDG_WorkItem* work_item) = 0;

    /// cook the output node in the graph context of the given file as a single job
    virtual UT_StringHolder submitAsJob(
                                const UT_StringHolder& graph_file,
                                const UT_StringHolder& node_name) = 0;

    /// set the working directory local and network path, equivalent
    ///  to the __PDG_DIR__
    void                    setWorkingDir(const UT_StringHolder& local_path,
                                          const UT_StringHolder& remote_path);
    /// set the temp directory local and network path, equivalent
    ///  to the __PDG_TEMP__
    void                    setTempDir(const UT_StringHolder& local_path,
                                       const UT_StringHolder& remote_path);
    /// set the script directory local and network path, equivalent
    ///  to the __PDG_SCRIPTDIR__
    void                    setScriptDir(const UT_StringHolder& local_path,
                                         const UT_StringHolder& remote_path);

    /// sets whether or not the scheduler accepts in-process work items
    void                    setAcceptInProcess(bool in_process);

    /// returns the working directory (network) path, equivalent
    ///  to the __PDG_DIR__ command token when local is false.  
    /// When local is true, returns the absolute path to the shared root
    /// on the local file system
    UT_StringHolder         workingDir(bool local) const;

    /// returns the temp directory (network) path, equivalent
    ///  to the __PDG_TEMP__ command token when local is false.  
    /// When local is true, returns the absolute path to the shared temp
    /// on the local file system
    UT_StringHolder         tempDir(bool local) const;

    /// returns the temp script directory (network) path, equivalent
    ///  to the __PDG_SCRIPTDIR__ command token when local is false.  
    /// When local is true, returns the absolute path to the temp
    /// script on the local file system
    UT_StringHolder         scriptDir(bool local) const;

    /// Returns the path to the directory where log files generated during the
    /// cook should be written.
    /// When local is true, returns the aboslute path on the local file system.
    UT_StringHolder         logDir(bool local) const;

    /// Returns the path to the directory where work items are serialized
    /// before their commands are executed.
    /// When local is true, returns the aboslute path on the local file system.
    UT_StringHolder         dataDir(bool local) const;

    /// Returns the path to an application for the given name, for example
    /// "python", "hython"
    virtual UT_StringHolder applicationBin(const UT_StringHolder& name,
                                           const PDG_WorkItem* work_item);

    /// returns the server endpoint for work item results, in format <HOST>:<PORT>, 
    /// equivalent to the __PDG_RESULT_SERVER__ command token
    virtual UT_StringHolder workItemResultServerAddr() = 0;

    /// returns a URI for the workitem cook log (text file)
    virtual UT_StringHolder getLogURI(const PDG_WorkItem* work_item) = 0;
    /// returns a URI for the workitem status page if relevant
    virtual UT_StringHolder getStatusURI(const PDG_WorkItem* work_item) = 0;

    /// Terminate the give shared server
    virtual bool            endSharedServer(
                                const UT_StringHolder& sharedserver_name) = 0;
    /// Clears metadata associated with the given shared server name
    virtual void            clearSharedServerInfo(
                                const UT_StringHolder &sharedserver_name);
    /// Sets metadata associated with the given shared server name, returns full name
    virtual UT_StringHolder setSharedServerInfo(
                                const UT_StringHolder &sharedserver_name,
                                const PDGT_ValueArgs& info);
    /// Gets metadata associated with the given shared server name
    PDGT_ValueArgs          getSharedServerInfo(
                                const UT_StringHolder &sharedserver_name);
    /// Gets the list of shared servers
    UT_StringArray          getSharedServers();

    /// Delete the temporary directory and all its contents
    virtual void            cleanTempDirectory();

    /// Called to stop the running cook
    void                    stop(bool exiting);
    bool                    startCook(PDG_CookState& cook_state);
    void                    stopCook();
    void                    cancelCook(bool pause);
    bool                    tick(PDG_Scheduler::TickResult& tick_result);
    bool                    cookWorkItem(PDG_WorkItem* work_item);

    void                    submitStatic();
    virtual PDG_WorkItemState
                            queueWorkItem(PDG_WorkItem* work_item);
    void                    removeWorkItem(const PDG_WorkItem* work_item);

    void                    cancelWorkItem(PDG_WorkItem* work_item);
    void                    cancelNode(PDG_Node* node);

    void                    onWorkItemPriorityChanged(PDG_WorkItem* work_item);
    void                    process(bool skip_ready_items);

    bool                    isRunning(int count=0);
    bool                    isRunning(const PDG_WorkItem* work_item);
    bool                    isDefault();

    bool                    isWaitForFailures() const;
    bool                    isValidateOutputFiles() const;
    bool                    isCheckExpectedOutputFiles() const;
    bool                    isCompressWorkItemData() const;
    virtual PDGN::PDGN_PollingClientNNG*            
                            getPollingClient() { return nullptr; }

    WorkItemDataSource      workItemDataSource() const;

    int                     numRunningItems() const;
    int                     numFailedItems() const;
    int                     numQueuedItems() const;

    void                    setContext(PDG_GraphContext* context);
    PDG_GraphContext*       context() const;

    void                    dependencyGraph(
                                    PDG_WorkItemMap& dependencies,
                                    PDG_WorkItemMap& dependents,
                                    PDG_WorkItemArray& ready,
                                    bool expand);

    /// Starts a service using this scheduler
    virtual bool            startService(UT_WorkBuffer& errors,
                                         PDG_Service* service)
                            {
                                return false;
                            }
    
    /// Stops a service that was started with this scheduler
    virtual bool            stopService(UT_WorkBuffer& errors,
                                        PDG_Service* service)
                            {
                                return false;
                            }

    template <typename T, typename D=T>
    bool                    evaluateOverride(T& result,
                                PDG_NodeInterface* node,
                                const UT_StringHolder& prefix,
                                const UT_StringHolder& parm,
                                PDG_WorkItem* work_item,
                                const D& default_value,
                                UT_WorkBuffer& errors) const
    {
        UT_WorkBuffer parm_name;
        UT_WorkBuffer toggle_name;

        if (prefix.isstring() && prefix.length() > 0)
        {
            toggle_name.format("{}_override_toggle", templateName());
            parm_name.format("{}_{}", prefix, parm);
        }
        else
        {
            parm_name.append(parm);
        }

        // If the user supplied us with a batch item, evaluate against the first
        // sub item instead
        PDG_WorkItem* eval_work_item = work_item;
        if (work_item && work_item->isBatch())
        {
            auto batch_item =
                static_cast<PDG_BatchWorkItem*>(work_item);
            if (batch_item->batchSize() > 0)
                eval_work_item = batch_item->batchItems()[0];
        }

        result = default_value;
        PDG_Port* port = parameter(parm_name.buffer());
        if (port)
        {
            if (!port->evaluate(0, result, eval_work_item, errors))
                return false;
        }

        if (!node)
            return true;

        port = node->parameter(toggle_name.buffer());

        if (port)
        {
            exint toggle_result;
            port->evaluate(0, toggle_result, eval_work_item, errors);

            if (toggle_result < 1)
                return true;
        }

        port = node->parameter(parm_name.buffer());
        if (port)
        {
            if (!port->evaluate(0, result, eval_work_item, errors))
                return false;
        }

        return true;
    }


    void                    onWorkItemSetStringArray(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const PDG_AttributeString::Array& value);
    void                    onWorkItemSetFloatArray(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const PDG_AttributeFloat::Array& value);
    void                    onWorkItemSetIntArray(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const PDG_AttributeInteger::Array& value);
    void                    onWorkItemSetFileArray(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const PDG_AttributeFile::Array& value);

    void                    onWorkItemSetPyObject(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const UT_StringHolder& pyobject_repr);
    void                    onWorkItemSetString(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const UT_StringHolder& value,
                                int attrib_index);
    void                    onWorkItemSetFloat(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                fpreal value,
                                int attrib_index);
    void                    onWorkItemSetInt(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                exint value,
                                int attrib_index);
    void                    onWorkItemSetFile(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& attribute_name,
                                const PDG_AttributeFile::Data& value,
                                int attrib_index);
    
    void                    onWorkItemAddOutput(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& path,
                                const UT_StringHolder& tag,
                                PDG_File::Hash hash_code,
                                bool active_only);
    void                    onWorkItemAddOutputs(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringArray& paths,
                                const UT_StringHolder& tags,
                                const PDG_File::HashArray& hashes,
                                bool active_only);
    void                    onWorkItemAddOutputs(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringArray& paths,
                                const UT_StringArray& tags,
                                const PDG_File::HashArray& hashes,
                                bool active_only);
    void                    onWorkItemInvalidateCache(
                                PDG_WorkItemID work_item_id,
                                int index);

    void                    onWorkItemSucceeded(
                                PDG_WorkItemID work_item_id,
                                int index,
                                fpreal cook_duration);
    void                    onWorkItemFailed(
                                PDG_WorkItemID work_item_id,
                                int index);
    void                    onWorkItemCanceled(
                                PDG_WorkItemID work_item_id,
                                int index);
    void                    onWorkItemStartCook(
                                PDG_WorkItemID work_item_id,
                                int index,
                                bool clear_outputs);

    void                    onWorkItemSetCustomState(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& custom_state);
    void                    onWorkItemSetCookPercent(
                                PDG_WorkItemID work_item_id,
                                int index,
                                fpreal &cook_percent);
    void                    onWorkItemAppendLog(
                                PDG_WorkItemID work_item_id,
                                int index,
                                const UT_StringHolder& log_data,
                                PDG_WorkItemLogType log_type);

    PDG_WorkItemEvalState   isWorkItemReady(
                                PDG_WorkItemID work_item_id,
                                int index);

    /// Emits errors or warnings from the scheduler as PDG_Events
    void                    addError(
                                const UT_StringHolder& message) const override;
    void                    addWarning(
                                const UT_StringHolder& message) const override;

    virtual fpreal          tickPeriod() const;
    virtual int             maxItemsPerTick() const;
    virtual int             maxConcurrentTasks() const;

    PathMappingMode         mapMode() const;

protected:
    /// Find the workitem and report error if not found
    PDG_WorkItem*           workItemChecked(
                                PDG_WorkItemID work_item_id,
                                int index);

    PDGE_Dependency::State  evalResolve(
                                PDGE_Resolutions& resolutions,
                                const PDGE_Evaluator& evaluator,
                                PDGE_Dependency* dependency) override;

private:
    struct Result
    {
        PDG_WorkItemState   myState;
        PDG_WorkItemID      myId;
        fpreal              myDuration;
        int                 myIndex;
        bool                myUnresolve;
    };

    using ResultMap     = UT_Map<PDG_WorkItemID, Result>;
    using ResultQueue   = UT_ConcurrentQueue<Result>;

    using PriorityQueue =
        UT_ConcurrentPriorityQueue<PDG_WorkItem*,
            PDG_WorkItemSort::Priority<true>>;
    using WorkItemSet = UT_ConcurrentSet<PDG_WorkItemID>;
    using NodeSet = UT_ConcurrentSet<PDG_Node*>;
    using SharedServerInfoMap =
        UT_ConcurrentHashMap<UT_StringHolder, PDGT_ValueArgs>;

private:
    ScheduleResult          schedule(PDG_WorkItem* work_item,
                                UT_Array<Result>& pending);
    int                     markCompleted(const Result& result,
                                UT_Array<Result>& pending);
    PDG_WorkItem*           workItem(
                                PDG_WorkItemID work_item_id,
                                int index);
    void                    updateDefaultDirectories();

private:
    PDGE_Dependency         myIsCookedDep;

    PriorityQueue           myScheduleQueue;
    ResultQueue             myScheduleResults;
    ResultMap               myPauseResults;
    SharedServerInfoMap     mySharedServerInfo;
    PDG_WorkItemIDSet       myPausedItems;

    WorkItemSet             myRunningItems;
    WorkItemSet             myFailedItems;
    WorkItemSet             myLongRunningItems;

    WorkItemSet             myCanceledItems;
    NodeSet                 myCanceledNodes;

    UT_StringHolder         myName;
    UT_StringHolder         myLocalWorkingDir;
    UT_StringHolder         myLocalTempDir;
    UT_StringHolder         myLocalScriptDir;
    UT_StringHolder         myRemoteWorkingDir;
    UT_StringHolder         myRemoteTempDir;
    UT_StringHolder         myRemoteScriptDir;

    PDG_GraphContext*       myContext;

    time_t                  myLastTick;
    WorkItemDataSource      myDataSource;
    TempDirCleanup          myTempDirCleanup;

    UT_TBBSpinLock          myProcessLock;

    bool                    myCancelingFlag;
    bool                    myPausingFlag;
    bool                    myStaticCook;
    bool                    myInProcess;

    bool                    myIsWaitForFailures;
    bool                    myIsValidateOutputFiles;
    bool                    myIsCheckExpectedOutputFiles;
    bool                    myIsCompressWorkItemData;
};

#endif /* __PDG_SCHEDULER_H__ */