lsst.ctrl.pool.test.demoTask.DemoTask Class Reference

Inheritance diagram for lsst.ctrl.pool.test.demoTask.DemoTask:

Public Member Functions
def	batchWallTime (cls, time, parsedCmd, numCores)
def	runDataRef (self, visitRef)
def	run (self, cache, dataId, butler=None)
def	parseAndRun (cls, *args, **kwargs)
def	parseAndSubmit (cls, args=None, **kwargs)
def	batchCommand (cls, args)
	Return command to run CmdLineTask. More...
def	logOperation (self, operation, catch=False, trace=True)
	Provide a context manager for logging an operation. More...

Static Public Attributes
	ConfigClass = Config

Detailed Description

Task for demonstrating the BatchPoolTask functionality

Definition at line 12 of file demoTask.py.

Member Function Documentation

batchCommand()

def lsst.ctrl.pool.parallel.BatchCmdLineTask.batchCommand (   cls,   args  )

inherited

Return command to run CmdLineTask.

    @param cls: Class
    @param args: Parsed batch job arguments (from BatchArgumentParser)

Definition at line 476 of file parallel.py.

    def batchCommand(cls, args):
        """!Return command to run CmdLineTask
 
        @param cls: Class
        @param args: Parsed batch job arguments (from BatchArgumentParser)
        """
        job = args.job if args.job is not None else "job"
        module = cls.__module__
        script = ("import os; os.umask(%#05o); " +
                  "import lsst.base; lsst.base.disableImplicitThreading(); " +
                  "import lsst.ctrl.pool.log; lsst.ctrl.pool.log.jobLog(\"%s\"); ") % (UMASK, job)
 
        if args.batchStats:
            script += ("import lsst.ctrl.pool.parallel; import atexit; " +
                       "atexit.register(lsst.ctrl.pool.parallel.printProcessStats); ")
 
        script += "import %s; %s.%s.parseAndRun();" % (module, module, cls.__name__)
 
        profilePre = "import cProfile; import os; cProfile.run(\"\"\""
        profilePost = "\"\"\", filename=\"profile-" + job + "-%s-%d.dat\" % (os.uname()[1], os.getpid()))"
 
        return ("python -c '" + (profilePre if args.batchProfile else "") + script +
                (profilePost if args.batchProfile else "") + "' " + shCommandFromArgs(args.leftover) +
                " --noExit")
 

batchWallTime()

def lsst.ctrl.pool.test.demoTask.DemoTask.batchWallTime	(		cls,
			time,
			parsedCmd,
			numCores
	)

Return walltime request for batch job

Subclasses should override if the walltime should be calculated
differently (e.g., addition of some serial time).

@param time: Requested time per iteration
@param parsedCmd: Results of argument parsing
@param numCores: Number of cores

Reimplemented from lsst.ctrl.pool.parallel.BatchCmdLineTask.

Definition at line 26 of file demoTask.py.

    def batchWallTime(cls, time, parsedCmd, numCores):
        """Return walltime request for batch job
 
        Subclasses should override if the walltime should be calculated
        differently (e.g., addition of some serial time).
 
        @param time: Requested time per iteration
        @param parsedCmd: Results of argument parsing
        @param numCores: Number of cores
        """
        numTargets = [sum(1 for ccdRef in visitRef.subItems("ccd") if ccdRef.datasetExists("raw")) for
                      visitRef in parsedCmd.id.refList]
        return time*sum(math.ceil(tt/numCores) for tt in numTargets)
 

logOperation()

def lsst.ctrl.pool.parallel.BatchCmdLineTask.logOperation (   self,   operation,   catch = False,   trace = True  )

inherited

Provide a context manager for logging an operation.

    @param operation: description of operation (string)
    @param catch: Catch all exceptions?
    @param trace: Log a traceback of caught exception?

    Note that if 'catch' is True, all exceptions are swallowed, but there may
    be other side-effects such as undefined variables.

Definition at line 502 of file parallel.py.

    def logOperation(self, operation, catch=False, trace=True):
        """!Provide a context manager for logging an operation
 
        @param operation: description of operation (string)
        @param catch: Catch all exceptions?
        @param trace: Log a traceback of caught exception?
 
        Note that if 'catch' is True, all exceptions are swallowed, but there may
        be other side-effects such as undefined variables.
        """
        self.log.info("%s: Start %s" % (NODE, operation))
        try:
            yield
        except Exception:
            if catch:
                cls, e, _ = sys.exc_info()
                self.log.warn("%s: Caught %s while %s: %s" % (NODE, cls.__name__, operation, e))
                if trace:
                    self.log.info("%s: Traceback:\n%s" % (NODE, traceback.format_exc()))
                return
            raise
        finally:
            self.log.info("%s: Finished %s" % (NODE, operation))
 
 

parseAndRun()

def lsst.ctrl.pool.parallel.BatchPoolTask.parseAndRun (   cls, *  args, **  kwargs  )

inherited

Run with a MPI process pool

Definition at line 534 of file parallel.py.

    def parseAndRun(cls, *args, **kwargs):
        """Run with a MPI process pool"""
        pool = startPool()
        super(BatchPoolTask, cls).parseAndRun(*args, **kwargs)
        pool.exit()
 
 

parseAndSubmit()

def lsst.ctrl.pool.parallel.BatchCmdLineTask.parseAndSubmit (   cls,   args = None, **  kwargs  )

inherited

Definition at line 435 of file parallel.py.

    def parseAndSubmit(cls, args=None, **kwargs):
        taskParser = cls._makeArgumentParser(doBatch=True, add_help=False)
        batchParser = BatchArgumentParser(parent=taskParser)
        batchArgs = batchParser.parse_args(config=cls.ConfigClass(), args=args, override=cls.applyOverrides,
                                           **kwargs)
 
        if not cls.RunnerClass(cls, batchArgs.parent).precall(batchArgs.parent):  # Write config, schema
            taskParser.error("Error in task preparation")
 
        setBatchType(batchArgs.batch)
 
        if batchArgs.batch is None:     # don't use a batch system
            sys.argv = [sys.argv[0]] + batchArgs.leftover  # Remove all batch arguments
 
            return cls.parseAndRun()
        else:
            if batchArgs.walltime > 0:
                walltime = batchArgs.walltime
            else:
                numCores = batchArgs.cores if batchArgs.cores > 0 else batchArgs.nodes*batchArgs.procs
                walltime = cls.batchWallTime(batchArgs.time, batchArgs.parent, numCores)
 
            command = cls.batchCommand(batchArgs)
            batchArgs.batch.run(command, walltime=walltime)
 

run()

def lsst.ctrl.pool.test.demoTask.DemoTask.run	(		self,
			cache,
			dataId,
			butler = None
	)

Read image and return number of pixels

Only the slave nodes run this method.

Definition at line 58 of file demoTask.py.

    def run(self, cache, dataId, butler=None):
        """Read image and return number of pixels
 
        Only the slave nodes run this method.
        """
        assert butler is not None
        with self.logOperation("read %s" % (dataId,)):
            raw = butler.get("raw", dataId, immediate=True)
            dims = raw.getDimensions()
            num = dims.getX()*dims.getY()
            self.log.info("Read %d pixels for %s" % (num, dataId,))
            return num
 

runDataRef()

def lsst.ctrl.pool.test.demoTask.DemoTask.runDataRef	(		self,
			visitRef
	)

Main entry-point

Only the master node runs this method.  It will dispatch jobs to the
slave nodes.

Definition at line 40 of file demoTask.py.

    def runDataRef(self, visitRef):
        """Main entry-point
 
        Only the master node runs this method.  It will dispatch jobs to the
        slave nodes.
        """
        pool = Pool("test")
 
        # Less overhead to transfer the butler once rather than in each dataRef
        dataIdList = dict([(ccdRef.get("ccdExposureId"), ccdRef.dataId)
                           for ccdRef in visitRef.subItems("ccd") if ccdRef.datasetExists("raw")])
        dataIdList = collections.OrderedDict(sorted(dataIdList.items()))
 
        with self.logOperation("master"):
            total = pool.reduce(operator.add, self.run, list(dataIdList.values()),
                                butler=visitRef.getButler())
        self.log.info("Total number of pixels read: %d" % (total,))
 

Member Data Documentation

ConfigClass

lsst.ctrl.pool.test.demoTask.DemoTask.ConfigClass = Config

static

Definition at line 14 of file demoTask.py.

---

The documentation for this class was generated from the following file:

• /j/snowflake/release/lsstsw/stack/lsst-scipipe-0.7.0/Linux64/ctrl_pool/22.0.1-2-g92698f7+9f079a9461/python/lsst/ctrl/pool/test/demoTask.py