1. Create a pdg/types directory in your HOUDINI_PATH, for example ~/houdini18.5/pdg/types.

2. Create a python package for your scheduler, by creating a directory named after the new scheduler, and adding an init.py file for example ~/houdini18.5/pdg/types/customscheduler/__init__.py.

3. Create a python module for your scheduler within the new package directory. For example ~/houdini18.5/pdg/types/customscheduler/customscheduler.py

4. Inside customscheduler.py implement your scheduler type. It may be initially helpful to subclass our internal scheduler so you have something working right away. This is a simple example of a Local Scheduler which does serial spawning of jobs.

import os
import shlex
import subprocess
import sys
import tempfile
import six.moves.urllib.parse as parse

# Add the schedulers types folder in $HFS to the PYTHONPATH so that we can
# import LocalScheduler class (it won't be loaded yet)
import pdg
sys.path.append(os.path.expandvars("$HFS/houdini/pdg/types/schedulers"))
from local import LocalScheduler
sys.path.pop()
from pdg.utils import expand_vars

class CustomScheduler(LocalScheduler):
    def onSchedule(self, work_item):
        print("onSchedule {}".format(work_item.name))
        item_command = work_item.command
        item_id = work_item.id
        item_command = self.expandCommandTokens(item_command, work_item)
        job_env = self._generateEnvironment(work_item, 0)
        # Ensure directories exist and serialize the work item
        self.createJobDirsAndSerializeWorkItems(work_item)
        # replace any job vars, and then any Houdini vars ($HFS etc)
        item_command = expand_vars(item_command, job_env)
        item_command = os.path.expandvars(item_command)
        self.workItemStartCook(item_id, -1)
        # ensure no unicode is in the command, because we want to use shlex
        if sys.version_info.major < 3:
            item_command = item_command.encode('ascii', 'ignore')
        argv = shlex.split(item_command)
        try:
            log = subprocess.check_output(argv, shell=True, env=job_env)
            with tempfile.NamedTemporaryFile(delete=False) as f:
                f.write(log)
                work_item.setStringAttrib("custom_log", f.name)
            self.workItemSucceeded(item_id, -1, 0)
        except subprocess.CalledProcessError:
            self.workItemFailed(item_id, -1, 0)
        return pdg.scheduleResult.Succeeded
    def getLogURI(self, work_item):
        log = work_item.stringAttribValue('custom_log')
        if log:
            return parse.urlunparse(('file','', log, '', '', ''))
        return ''

1. Inside __init__.py, register your new scheduler type.

from . import customscheduler 

def registerTypes(type_registry):
    print("Registering customscheduler") 
    type_registry.registerScheduler(customscheduler.CustomScheduler, label="Test Custom Scheduler") 

1. When PDG initializes it will import your customscheduler package, which will register your new scheduler type. It will show up in the TOPs TAB Menu as customscheduler.

1. Create a Python Scheduler node in your TOP network.

   This node provides fields for callback functions that define the behavior of the scheduler. The node will function just like any other scheduler, and will evaluate and execute the python code entered into its callback function fields. This is very useful to let you prototype your scheduler code in a working network.

2. See implementing the callbacks below for how to fill out the various Python snippets on the node’s tabs.

Use the Python snippets on the Initialization and cleanup tab for code that should run when the scheduler first starts, should run when the TOP network starts cooking, should run when the network stops cooking, and/or should run when the scheduler is no longer needed.

The working directory is a directory where the TOP network’s code runs, and is usually where generated files are saved. In a farm set-up, the working directory should be on a shared network filesystem. PDG needs to know the local path to the working directory, as well as the remote. These correspond to a path which is valid on the user’s file-system (where the network is cooking), and a path which is valid on a remote render farm host respectively. The paths supplied to PDG will be used to convert file paths between local and remote (referred to as localize and delocalize)

The code to generate/translate local and remote paths is scheduler-specific. For example, your scheduler software may have its own API for discovering the shared directory, or it might have set system environment variables (for example, $REPO_DIR).

The Python Scheduler node has a Working directory parameter that you can access in the code using self["pdg_workingdir"].evaluateString(). By convention, this parm can be either a full local path, or a relative path which is assumed to be relative to the mount point for your farm file system.

This is the key method that does the “real work” of scheduling jobs. It is called by cooking nodes to process and schedule their work items. It receives a pdg.WorkItem object as work_item.

This is where you must implement the scheduler-specific logic to take WorkItem data and use it to create a scheduled job in whatever custom system you are integrating. This may involve running the command line in an environment with certain variables defined, copying dependent files, calling scheduler APIs, and so on.

• The command string attribute contains the command line the job should run.

• If the scheduler needs/accepts a per-job unique key to look up the job later, use WorkItem.name.

• The snippet should return a pdg.scheduleResult value once the job is scheduled (or encounters an error).

Often you will need to monitor the progress of a job using polling of your farm scheduler’s API. This can be done by implementing the onTick callback. The return value pdg.tickResult of this tick function will inform PDG of your binding’s state.

For example you can return SchedulerCancelCook if you've lost connection to your farm scheduler and there’s no way to recover. SchedulerBusy and SchedulerReady let PDG know if you are ready to handle more onSchedule callbacks. You can use this to rate-limit item submission to your farm in order to control load on the server and other resources.

tickPeriod and maxItems scheduler node parameters can be used to configure the rate of calls to onTick and onSchedule

See onScheduleStatic. This is only used for static cook mode, which is not currently exposed in the TOP UI.

You must implement a method to communicate job status changes back to the TOP network. How to get notifications back from the scheduler software is implementation specific. However Houdini ships with python implementations of several scheduler bindings which code you can likely re-use. It is highly recommended to examine the source code in $HFS/houdini/pdg/types/schedulers and $HHP/pdg.

Note that if the scheduler requires polling to detect status changes, you should do this in the onTick method.

Remember to keep a reference to self (the Scheduler object) in any threads or callbacks you use.

However you detect status changes in the scheduler software, you would then call one of the following methods on pdg.Scheduler (self):

This is called if a user tries to submit an entire TOP network as a single job. This is optional functionality you can choose to implement. For example, the HQueue Scheduler has a Submit graph as job button.

• The graph_file variable contains the path to a HIP file containing the TOP network to cook. This path is relative to the working directory on the machine cooking the network.

• The node_path variable contains a node path of the TOP network node to cook inside the HIP file.

If you want to support this functionality, it is up to your code to create and schedule a script job that runs Houdini, loads the specified HIP file, and cooks the specified network.

Called when a file dependency at file_path should be copied from the local machine to a remote location (e.g. a shared network drive). This is used for files that are necessary to work scripts, so they are copied to the working directory. This snippet should return True if the file copied successfully.

These snippets provide a pdg.WorkItem object in the work_item variable. If you can use work item’s data, in combination with the scheduler’s API if necessary, to calculate the location of the corresponding job’s logs and/or status page, the TOPs interface can then display them as part of the work item’s information.

Instead of a fully static job, it may be possible to instead pre-populate the farm job with all the child Tasks that are going to eventually be run, as placeholders, which are in a suspended state. Then as the PDG cook proceeds as normal the custom scheduler will activate those jobs as they become ready via the onSchedule callback.

With the approach the farm wrangler will get an overview of the full scope of the cook up front as they would with a static job description.

In order to pre-populate the Job, the custom scheduler should inspect the statically generated PDG task graph at the start of the cook. This task graph can be accessed with pdg.SchedulerBase.dependencyGraph. Note that the TOP graph should be fully generated before querying the task graph.

Here’s an example of how to generate a DOT format graph of a given task graph, which can be viewed with Graphviz software.

def taskgraphDOT(scheduler, dotfile=None):
    import sys
    dependencies, dependents, ready = scheduler.dependencyGraph(True)
    outf = open(dotfile, 'w') if dotfile else sys.stdout
    outf.write("digraph taskgraph {\n")
    for work_item, children in dependents.items():
        if not children:
            continue
        child_list = " ; ".join((child.name for child in children))
        if len(children) > 1:
            outf.write("%s -> {%s};\n" % (work_item.name, child_list))
        else:
            outf.write("{} -> {};\n".format(work_item.name, child_list))
    outf.write("}")

# Get the scheduler of interest, generate the task graph and get the dep graph.
node = hou.node('/obj/topnet1/localscheduler')
sch = node.getPDGNode()
output_node = node.parent().displayNode()
output_node.cookWorkItems(block=True, generate_only=True)
taskGraphDOT(sch, '/tmp/depgraph.dot')