exchanger.py 文件源码

def push_message(self, dest_rank, count_arr, recorder=None):

        '''
        push message:
        push params_i and alpha_i to the choosen rank
        '''

        # detect if any other worker is pushing to self at the same time to prevent deadlock
        while self.comm.Iprobe(source=MPI.ANY_SOURCE, tag=700): 
            if self.test: print('a potential deadlock prevented')
            self.process_messages(count_arr, recorder)

        if recorder: recorder.start()

        # 0. blocking request

        if self.test: print('%d pushing msg to %d'  % (self.rank,dest_rank))

        self.comm.Send(buf=count_arr,dest=dest_rank, tag=700)  

        if self.test: print('%d requested to %d'  % (self.rank,dest_rank))

        # 1. push

        self.gpucomm, dest_gpurank, self_gpurank = self.get_gpucomm_with(dest_rank)

        self._push_params(self_gpurank, dest_rank)

        if self.test: print('%d msg pushed'  % self.rank)

        if recorder: recorder.end('comm')