lsst.ctrl.pool.pool.Comm Class Reference

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

Public Member Functions
def	__new__ (cls, comm=mpi.COMM_WORLD, recvSleep=0.1, barrierSleep=0.1)
	Construct an MPI.Comm wrapper. More...
def	recv (self, obj=None, source=0, tag=0, status=None)
def	send (self, obj=None, *args, **kwargs)
def	Barrier (self, tag=0)
def	broadcast (self, value, root=0)
def	scatter (self, dataList, root=0, tag=0)
def	Free (self)

Detailed Description

Wrapper to mpi4py's MPI.Intracomm class to avoid busy-waiting.

As suggested by Lisandro Dalcin at:
* http://code.google.com/p/mpi4py/issues/detail?id=4 and
* https://groups.google.com/forum/?fromgroups=#!topic/mpi4py/nArVuMXyyZI

Definition at line 243 of file pool.py.

Member Function Documentation

__new__()

def lsst.ctrl.pool.pool.Comm.__new__	(		cls,
			comm = mpi.COMM_WORLD,
			recvSleep = 0.1,
			barrierSleep = 0.1
	)

Construct an MPI.Comm wrapper.

    @param cls             Class
    @param comm            MPI.Intracomm to wrap a duplicate of
    @param recvSleep       Sleep time (seconds) for recv()
    @param barrierSleep    Sleep time (seconds) for Barrier()

Definition at line 251 of file pool.py.

    def __new__(cls, comm=mpi.COMM_WORLD, recvSleep=0.1, barrierSleep=0.1):
        """!Construct an MPI.Comm wrapper
 
        @param cls             Class
        @param comm            MPI.Intracomm to wrap a duplicate of
        @param recvSleep       Sleep time (seconds) for recv()
        @param barrierSleep    Sleep time (seconds) for Barrier()
        """
        self = super(Comm, cls).__new__(cls, comm.Dup())
        self._barrierComm = None  # Duplicate communicator used for Barrier point-to-point checking
        self._recvSleep = recvSleep
        self._barrierSleep = barrierSleep
        return self
 

Barrier()

def lsst.ctrl.pool.pool.Comm.Barrier	(		self,
			tag = 0
	)

Version of comm.Barrier() that doesn't busy-wait

A duplicate communicator is used so as not to interfere with the user's own communications.

Definition at line 281 of file pool.py.

    def Barrier(self, tag=0):
        """Version of comm.Barrier() that doesn't busy-wait
 
        A duplicate communicator is used so as not to interfere with the user's own communications.
        """
        self._checkBarrierComm()
        size = self._barrierComm.Get_size()
        if size == 1:
            return
        rank = self._barrierComm.Get_rank()
        mask = 1
        while mask < size:
            dst = (rank + mask) % size
            src = (rank - mask + size) % size
            req = self._barrierComm.isend(None, dst, tag)
            while not self._barrierComm.Iprobe(src, tag):
                time.sleep(self._barrierSleep)
            self._barrierComm.recv(None, src, tag)
            req.Wait()
            mask <<= 1
 

broadcast()

def lsst.ctrl.pool.pool.Comm.broadcast	(		self,
			value,
			root = 0
	)

Definition at line 302 of file pool.py.

    def broadcast(self, value, root=0):
        with PickleHolder(value):
            return super(Comm, self).bcast(value, root=root)
 

Free()

def lsst.ctrl.pool.pool.Comm.Free

(

self

)

Definition at line 330 of file pool.py.

    def Free(self):
        if self._barrierComm is not None:
            self._barrierComm.Free()
        super(Comm, self).Free()
 
 

recv()

def lsst.ctrl.pool.pool.Comm.recv	(		self,
			obj = None,
			source = 0,
			tag = 0,
			status = None
	)

Version of comm.recv() that doesn't busy-wait

Definition at line 265 of file pool.py.

    def recv(self, obj=None, source=0, tag=0, status=None):
        """Version of comm.recv() that doesn't busy-wait"""
        sts = mpi.Status()
        while not self.Iprobe(source=source, tag=tag, status=sts):
            time.sleep(self._recvSleep)
        return super(Comm, self).recv(buf=obj, source=sts.source, tag=sts.tag, status=status)
 

scatter()

def lsst.ctrl.pool.pool.Comm.scatter	(		self,
			dataList,
			root = 0,
			tag = 0
	)

Scatter data across the nodes

The default version apparently pickles the entire 'dataList',
which can cause errors if the pickle size grows over 2^31 bytes
due to fundamental problems with pickle in python 2. Instead,
we send the data to each slave node in turn; this reduces the
pickle size.

@param dataList  List of data to distribute; one per node
    (including root)
@param root  Index of root node
@param tag  Message tag (integer)
@return  Data for this node

Definition at line 306 of file pool.py.

    def scatter(self, dataList, root=0, tag=0):
        """Scatter data across the nodes
 
        The default version apparently pickles the entire 'dataList',
        which can cause errors if the pickle size grows over 2^31 bytes
        due to fundamental problems with pickle in python 2. Instead,
        we send the data to each slave node in turn; this reduces the
        pickle size.
 
        @param dataList  List of data to distribute; one per node
            (including root)
        @param root  Index of root node
        @param tag  Message tag (integer)
        @return  Data for this node
        """
        if self.Get_rank() == root:
            for rank, data in enumerate(dataList):
                if rank == root:
                    continue
                self.send(data, rank, tag=tag)
            return dataList[root]
        else:
            return self.recv(source=root, tag=tag)
 

send()

def lsst.ctrl.pool.pool.Comm.send	(		self,
			obj = None,
		*	args,
		**	kwargs
	)

Definition at line 272 of file pool.py.

    def send(self, obj=None, *args, **kwargs):
        with PickleHolder(obj):
            return super(Comm, self).send(obj, *args, **kwargs)
 

---

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/pool.py