lsst.ctrl.pool.pool.PoolSlave Class Reference

Inheritance diagram for lsst.ctrl.pool.pool.PoolSlave:

Public Member Functions
def	log (self, msg, *args)
def	run (self)
def	reduce (self)
def	mapNoBalance (self)
def	mapToPrevious (self)
def	storeSet (self)
def	storeDel (self)
def	storeClear (self)
def	cacheClear (self)
def	cacheList (self)
def	storeList (self)
def	exit (self)
def	isMaster (self)
def	storeSet (self, context, **kwargs)
def	storeDel (self, context, *nameList)
def	storeClear (self, context)
def	cacheClear (self, context)
def	cacheList (self, context)
def	storeList (self, context)
def	__call__ (cls, *args, **kwargs)

Public Attributes
	comm
	rank
	root
	size
	debugger
	node

Detailed Description

Slave node instance of MPI process pool

Definition at line 1048 of file pool.py.

Member Function Documentation

__call__()

def lsst.ctrl.pool.pool.SingletonMeta.__call__ (   cls, *  args, **  kwargs  )

inherited

Definition at line 387 of file pool.py.

    def __call__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super(SingletonMeta, cls).__call__(*args, **kwargs)
        return cls._instance
 
 

cacheClear() [1/2]

def lsst.ctrl.pool.pool.PoolSlave.cacheClear

(

self

)

Reset cache

Definition at line 1157 of file pool.py.

    def cacheClear(self):
        """Reset cache"""
        context = self.comm.broadcast(None, root=self.root)
        super(PoolSlave, self).cacheClear(context)
 

cacheClear() [2/2]

def lsst.ctrl.pool.pool.PoolNode.cacheClear (   self,   context  )

inherited

Reset cache for a particular context

Reimplemented in lsst.ctrl.pool.pool.PoolMaster.

Definition at line 607 of file pool.py.

    def cacheClear(self, context):
        """Reset cache for a particular context"""
        self.log("clearing cache", context)
        if context not in self._cache:
            return
        self._cache[context] = {}
 

cacheList() [1/2]

def lsst.ctrl.pool.pool.PoolSlave.cacheList

(

self

)

List cache contents

Definition at line 1162 of file pool.py.

    def cacheList(self):
        """List cache contents"""
        context = self.comm.broadcast(None, root=self.root)
        super(PoolSlave, self).cacheList(context)
 

cacheList() [2/2]

def lsst.ctrl.pool.pool.PoolNode.cacheList (   self,   context  )

inherited

List contents of cache

Reimplemented in lsst.ctrl.pool.pool.PoolMaster.

Definition at line 614 of file pool.py.

    def cacheList(self, context):
        """List contents of cache"""
        cache = self._cache[context] if context in self._cache else {}
        sys.stderr.write("Cache on %s (%s): %s\n" % (self.node, context, cache))
 

exit()

def lsst.ctrl.pool.pool.PoolSlave.exit

(

self

)

Allow exit from loop in 'run'

Definition at line 1172 of file pool.py.

    def exit(self):
        """Allow exit from loop in 'run'"""
        return True
 
 

isMaster()

def lsst.ctrl.pool.pool.PoolNode.isMaster (   self )

inherited

Definition at line 526 of file pool.py.

    def isMaster(self):
        return self.rank == self.root
 

log()

def lsst.ctrl.pool.pool.PoolSlave.log	(		self,
			msg,
		*	args
	)

Log a debugging message

Reimplemented from lsst.ctrl.pool.pool.PoolNode.

Definition at line 1051 of file pool.py.

    def log(self, msg, *args):
        """Log a debugging message"""
        assert self.rank != self.root, "This is not the master node."
        self.debugger.log("Slave %d" % self.rank, msg, *args)
 

mapNoBalance()

def lsst.ctrl.pool.pool.PoolSlave.mapNoBalance

(

self

)

Process bulk scattered data and return results

Definition at line 1102 of file pool.py.

    def mapNoBalance(self):
        """Process bulk scattered data and return results"""
        self.log("waiting for instruction")
        tags, func, args, kwargs, context = self.comm.broadcast(None, root=self.root)
        self.log("waiting for job")
        queue = self.comm.recv(tag=tags.work, source=self.root)
 
        resultList = []
        for index, data in queue:
            self.log("running job", index)
            result = self._processQueue(context, func, [(index, data)], *args, **kwargs)[0]
            resultList.append(result)
 
        self.comm.send(resultList, self.root, tag=tags.result)
        self.log("done")
 

mapToPrevious()

def lsst.ctrl.pool.pool.PoolSlave.mapToPrevious

(

self

)

Process the same scattered data processed previously

Definition at line 1119 of file pool.py.

    def mapToPrevious(self):
        """Process the same scattered data processed previously"""
        self.log("waiting for instruction")
        tags, func, args, kwargs, context = self.comm.broadcast(None, root=self.root)
        queue = list(self._cache[context].keys()) if context in self._cache else None
        index = queue.pop(0) if queue else -1
        self.log("request job", index)
        self.comm.gather(index, root=self.root)
        self.log("waiting for job")
        data = self.comm.scatter(None, root=self.root)
 
        while index >= 0:
            self.log("running job")
            result = func(self._getCache(context, index), data, *args, **kwargs)
            self.log("pending", queue)
            nextIndex = queue.pop(0) if queue else -1
            self.comm.send((index, result, nextIndex), self.root, tag=tags.result)
            index = nextIndex
            if index >= 0:
                data = self.comm.recv(tag=tags.work, source=self.root)
 
        self.log("done")
 

reduce()

def lsst.ctrl.pool.pool.PoolSlave.reduce

(

self

)

Reduce scattered data and return results

Definition at line 1076 of file pool.py.

    def reduce(self):
        """Reduce scattered data and return results"""
        self.log("waiting for instruction")
        tags, func, reducer, args, kwargs, context = self.comm.broadcast(None, root=self.root)
        self.log("waiting for job")
        job = self.comm.scatter(None, root=self.root)
 
        out = None  # Reduction result
        while not isinstance(job, NoOp):
            index, data = job
            self.log("running job")
            result = self._processQueue(context, func, [(index, data)], *args, **kwargs)[0]
            if reducer is None:
                report = (index, result)
            else:
                report = None
                out = reducer(out, result) if out is not None else result
            self.comm.send(report, self.root, tag=tags.request)
            self.log("waiting for job")
            job = self.comm.recv(tag=tags.work, source=self.root)
 
        if reducer is not None:
            self.comm.gather(out, root=self.root)
        self.log("done")
 

run()

def lsst.ctrl.pool.pool.PoolSlave.run

(

self

)

Serve commands of master node

Slave accepts commands, which are the names of methods to execute.
This exits when a command returns a true value.

Definition at line 1057 of file pool.py.

    def run(self):
        """Serve commands of master node
 
        Slave accepts commands, which are the names of methods to execute.
        This exits when a command returns a true value.
        """
        menu = dict((cmd, getattr(self, cmd)) for cmd in ("reduce", "mapNoBalance", "mapToPrevious",
                                                          "storeSet", "storeDel", "storeClear", "storeList",
                                                          "cacheList", "cacheClear", "exit",))
        self.log("waiting for command from", self.root)
        command = self.comm.broadcast(None, root=self.root)
        self.log("command", command)
        while not menu[command]():
            self.log("waiting for command from", self.root)
            command = self.comm.broadcast(None, root=self.root)
            self.log("command", command)
        self.log("exiting")
 

storeClear() [1/2]

def lsst.ctrl.pool.pool.PoolSlave.storeClear

(

self

)

Reset data store

Definition at line 1152 of file pool.py.

    def storeClear(self):
        """Reset data store"""
        context = self.comm.broadcast(None, root=self.root)
        super(PoolSlave, self).storeClear(context)
 

storeClear() [2/2]

def lsst.ctrl.pool.pool.PoolNode.storeClear (   self,   context  )

inherited

Clear stored data for a particular context

Reimplemented in lsst.ctrl.pool.pool.PoolMaster.

Definition at line 600 of file pool.py.

    def storeClear(self, context):
        """Clear stored data for a particular context"""
        self.log("clearing store", context)
        if context not in self._store:
            raise KeyError("No such context: %s" % context)
        self._store[context] = {}
 

storeDel() [1/2]

def lsst.ctrl.pool.pool.PoolSlave.storeDel

(

self

)

Delete value in store

Definition at line 1147 of file pool.py.

    def storeDel(self):
        """Delete value in store"""
        context, nameList = self.comm.broadcast(None, root=self.root)
        super(PoolSlave, self).storeDel(context, *nameList)
 

storeDel() [2/2]

def lsst.ctrl.pool.pool.PoolNode.storeDel (   self,   context, *  nameList  )

inherited

Delete value in store for a particular context

Reimplemented in lsst.ctrl.pool.pool.PoolMaster.

Definition at line 592 of file pool.py.

    def storeDel(self, context, *nameList):
        """Delete value in store for a particular context"""
        self.log("deleting from store", context, nameList)
        if context not in self._store:
            raise KeyError("No such context: %s" % context)
        for name in nameList:
            del self._store[context][name]
 

storeList() [1/2]

def lsst.ctrl.pool.pool.PoolSlave.storeList

(

self

)

List store contents

Definition at line 1167 of file pool.py.

    def storeList(self):
        """List store contents"""
        context = self.comm.broadcast(None, root=self.root)
        super(PoolSlave, self).storeList(context)
 

storeList() [2/2]

def lsst.ctrl.pool.pool.PoolNode.storeList (   self,   context  )

inherited

List contents of store for a particular context

Reimplemented in lsst.ctrl.pool.pool.PoolMaster.

Definition at line 619 of file pool.py.

    def storeList(self, context):
        """List contents of store for a particular context"""
        if context not in self._store:
            raise KeyError("No such context: %s" % context)
        sys.stderr.write("Store on %s (%s): %s\n" % (self.node, context, self._store[context]))
 
 

storeSet() [1/2]

def lsst.ctrl.pool.pool.PoolSlave.storeSet

(

self

)

Set value in store

Definition at line 1142 of file pool.py.

    def storeSet(self):
        """Set value in store"""
        context, kwargs = self.comm.broadcast(None, root=self.root)
        super(PoolSlave, self).storeSet(context, **kwargs)
 

storeSet() [2/2]

def lsst.ctrl.pool.pool.PoolNode.storeSet (   self,   context, **  kwargs  )

inherited

Set values in store for a particular context

Reimplemented in lsst.ctrl.pool.pool.PoolMaster.

Definition at line 584 of file pool.py.

    def storeSet(self, context, **kwargs):
        """Set values in store for a particular context"""
        self.log("storing", context, kwargs)
        if context not in self._store:
            self._store[context] = {}
        for name, value in kwargs.items():
            self._store[context][name] = value
 

Member Data Documentation

comm

lsst.ctrl.pool.pool.PoolNode.comm

inherited

Definition at line 497 of file pool.py.

debugger

lsst.ctrl.pool.pool.PoolNode.debugger

inherited

Definition at line 503 of file pool.py.

node

lsst.ctrl.pool.pool.PoolNode.node

inherited

Definition at line 504 of file pool.py.

rank

lsst.ctrl.pool.pool.PoolNode.rank

inherited

Definition at line 498 of file pool.py.

root

lsst.ctrl.pool.pool.PoolNode.root

inherited

Definition at line 499 of file pool.py.

size

lsst.ctrl.pool.pool.PoolNode.size

inherited

Definition at line 500 of file pool.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+dcf3732eb2/python/lsst/ctrl/pool/pool.py