_broker.py 文件源码

def adopt_sessions(self):
        if not self.fdtx_path:
            return
        fdtx = FdTx(None)
        fdtx.connect(self.fdtx_path, 5)
        # receive open file descriptors together with their associated authkeys
        while True:
            try: # future safety: double the needed message length, two fd's
                authkey, fd = fdtx.get(AUTHKEY_LENGTH*2, 2)
            except ConnectionClosed:
                break # other end sent all it had
            if authkey not in self.adoption:
                continue # session died during handover
            self.adoption[authkey]['fd'] = fd[0] # just one fd per message
        for authkey in self.adoption:
            if 'fd' not in self.adoption[authkey]:
                continue # session died during handover
            # reconnect the session
            fd      = self.adoption[authkey]['fd']
            pid     = self.adoption[authkey]['pid']
            address = tuple(self.adoption[authkey]['address'])
            # represent the adopted session by an AdoptedSession instance that
            # only implements .pid and .terminate()
            local   = AdoptedSession(pid)
            sock    = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
            remote  = RemoteSession(address, authkey, timeout=1, sock=sock)
            self.add_connection(remote._connection, authkey)
            self.sessions[authkey] = (local, remote)

            # recreate the allocation records
            alloc   = self.adoption[authkey]['allocations']
            for a in alloc:
                resource   = a['profile']
                collateral = a['collateral']
                self.allocators['local'].allocate(resource, remote, collateral)