exchanger.py 文件源码

def process_messages(self, count_arr, recorder=None):

        if recorder: recorder.start()

        status = MPI.Status()

        s = self.comm.Iprobe(source=MPI.ANY_SOURCE, tag=700, status=status)

        # if self.test: print '%d probed, got %s' % (self.rank,s)

        while s:

            src_rank=status.source

            self.comm.Recv(buf=count_arr, source=src_rank, tag=700, status=status)

            self.gpucomm, src_gpurank, self_gpurank = self.get_gpucomm_with(src_rank)

            if self.test: print('%d merging with %d' % (self.rank, src_rank))

            self._merge_params_from(src_gpurank, src_rank)

            s = self.comm.Iprobe(source=MPI.ANY_SOURCE, tag=700, status=status)

            if self.test: print('%d probed again, got %s' % (self.rank,s))

        if recorder: recorder.end('comm')