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_AttributeFile.h"
#include "PDG_AttributePrimitive.h"
#include "PDG_BatchWorkItem.h"
#include "PDG_CookOptions.h"
#include "PDG_EventEmitter.h"
#include "PDG_EventTypes.h"
#include "PDG_File.h"
#include "PDG_FileUtils.h"
#include "PDG_NodeInterface.h"
#include "PDG_SchedulerTypes.h"
#include "PDG_WorkItemSort.h"

#include <PDGT/PDGT_ValueArgs.h>

#include <UT/UT_Array.h>
#include <UT/UT_ConcurrentPriorityQueue.h>
#include <UT/UT_ConcurrentHashMap.h>
#include <UT/UT_ConcurrentQueue.h>
#include <UT/UT_ConcurrentSet.h>
#include <UT/UT_Map.h>
#include <UT/UT_Options.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_SchedulerType;
class PDG_WorkItem;

class UT_JSONValue;
class UT_JSONWriter;

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 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
    const PDG_EventFilter&      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 is cooked dep for this scheduler
    inline PDGE_Dependency*     isSetupDep()
                                    { return &myIsSetupDep; }

    /// 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 the node interface/template that describes the default 
    /// parameters and connections for the scheduler
    const PDG_NodeInterface*    templateInterface() const override;

    /// Returns the type object that defines the scheduler
    const PDG_SchedulerType*    typeObject() const;

    /// 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 at the before cooking on the background thread to setup
    /// anything needed for a cook.
    virtual bool                onSetupCook() = 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 for a work item that uses this scheduler, using the
    /// transfer settings on the scheduler itself.
    bool                        transferFile(
                                    const PDG_WorkItem* work_item,
                                    const PDG_File& file,
                                    PDG_FileUtils::TransferType default_type,
                                    bool check_type,
                                    UT_WorkBuffer& errors) const;

    /// Transfers a file from the local machine to the machine that will be
    /// executing processor tasks, e.g. a remote machine on the farm.
    bool                        transferFile(
                                    const UT_StringHolder& file_path,
                                    PDG_FileUtils::TransferType default_type,
                                    bool check_type,
                                    UT_WorkBuffer& errors) const;

    /// 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 
    /// variant of the method preserves the directory structure relative to
    /// the specified root directory
    bool                        transferFile(
                                    const UT_StringHolder& root_path,
                                    const UT_StringHolder& file_path,
                                    const UT_StringHolder& sub_path,
                                    PDG_FileUtils::TransferType default_type,
                                    bool check_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,
                                    bool check_type) const;

    /// Returns the path on the scheduler that a local file will be copied
    /// to. Unlike the above method, this method preserves the directory
    /// structure relative to the specified root directory
    UT_StringHolder             formatTransferPath(
                                    const UT_StringHolder& root_path,
                                    const UT_StringHolder& local_path,
                                    const UT_StringHolder& sub_path,
                                    PDG_FileUtils::TransferType default_type,
                                    bool local,
                                    bool check_type) 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 scheduler
    void                        stop(bool exiting);

    /// Called when a cook begins with the cook state data for that cook
    bool                        startCook(PDG_CookState& cook_state);

    /// Called when a cook stops
    void                        stopCook();

    /// Cancels an active cook
    void                        cancelCook(bool pause);

    /// Ticks the scheduler
    bool                        tick(PDG_Scheduler::TickResult& tick_result);

    /// Cooks a specific work item using it's internal cook method
    bool                        cookWorkItem(PDG_WorkItem* work_item);


    /// Submits a static cook with this scheduler
    void                        submitStatic();


    /// Queues a work item and returns a state that represents the result of
    /// the operation
    virtual PDG_WorkItemState   queueWorkItem(PDG_WorkItem* work_item);

    /// Removes a work item from any internal tracking lists
    void                        removeWorkItem(const PDG_WorkItem* work_item);


    /// Cancels a cook for a specific work tiem
    bool                        cancelWorkItem(PDG_WorkItem* work_item);

    /// Cancels a cook for a specific node
    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,
                                    bool cancel)
                                    { return false; }

    /// Starts a job for the specified service client
    virtual bool                startServiceClient(
                                    UT_WorkBuffer& errors,
                                    PDG_Service* service,
                                    PDG_ServiceClient* client)
                                    { return false; }

    /// Stops the job associated with a specific client process
    virtual bool                stopServiceClient(
                                    UT_WorkBuffer& errors,
                                    PDG_Service* service,
                                    PDG_ServiceClient* client)
                                    { 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                        onWorkItemSetDictArray(
                                    PDG_WorkItemID work_item_id,
                                    int index,
                                    const UT_StringHolder& attribute_name,
                                    const UT_Array<UT_OptionsHolder>& 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                        onWorkItemSetDict(
                                    PDG_WorkItemID work_item_id,
                                    int index,
                                    const UT_StringHolder& attribute_name,
                                    const UT_OptionsHolder& dict_repr,
                                    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;

    /// Writes the current parameter value configuration of this scheduler
    /// to JSON
    bool                        asJSON(
                                    UT_JSONWriter& writer,
                                    bool skip_defaults) const;

    /// Reads and constructs a scheduler from JSON and returns the instance
    static PDG_Scheduler*       fromJSON(
                                    const UT_JSONValue* value,
                                    PDG_GraphContext* context,
                                    UT_WorkBuffer& errors);

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

    PDG_PathMappingMode         mapMode() const;

    PDG_FileTransferType        transferType() const;
    UT_StringHolder             transferRoot() 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>;

    /// The key for the name of the scheduler when writing it to JSON
    static const UT_StringHolder theNameKey;

    /// The key for the type of the scheduler when writing it to JSON
    static const UT_StringHolder theTypeKey;

    /// The key for the parameters of the scheduler when writing them to JSON
    static const UT_StringHolder theParametersKey;

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;
    PDGE_Dependency             myIsSetupDep;

    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;
    const PDG_SchedulerType*    myTypeObject;

    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__ */