(self, process_obj)
| 39 | return len(self._fds) - 1 |
| 40 | |
| 41 | def _launch(self, process_obj): |
| 42 | prep_data = spawn.get_preparation_data(process_obj._name) |
| 43 | buf = io.BytesIO() |
| 44 | set_spawning_popen(self) |
| 45 | try: |
| 46 | reduction.dump(prep_data, buf) |
| 47 | reduction.dump(process_obj, buf) |
| 48 | finally: |
| 49 | set_spawning_popen(None) |
| 50 | |
| 51 | self.sentinel, w = forkserver.connect_to_new_process(self._fds) |
| 52 | # Keep a duplicate of the data pipe's write end as a sentinel of the |
| 53 | # parent process used by the child process. |
| 54 | _parent_w = os.dup(w) |
| 55 | self.finalizer = util.Finalize(self, util.close_fds, |
| 56 | (_parent_w, self.sentinel)) |
| 57 | with open(w, 'wb', closefd=True) as f: |
| 58 | f.write(buf.getbuffer()) |
| 59 | self.pid = forkserver.read_signed(self.sentinel) |
| 60 | |
| 61 | def poll(self, flag=os.WNOHANG): |
| 62 | if self.returncode is None: |
nothing calls this directly
no test coverage detected