@@ -100,16 +100,16 @@ def _parse_lines_from_buffer(buf): |
100 | 100 | # end |
101 | 101 |
|
102 | 102 | def _read_lines_from_fno(fno, last_buf_list): |
103 | | - buf = os.read(fno, mmap.PAGESIZE) |
104 | | - buf = last_buf_list[0] + buf |
| 103 | + for buf in iter(lambda: os.read(fno, mmap.PAGESIZE), b''): |
| 104 | + buf = last_buf_list[0] + buf |
105 | 105 |
|
106 | | - bi = 0 |
107 | | - for bi, line in _parse_lines_from_buffer(buf): |
108 | | - yield line |
109 | | - # for each line to parse from the buffer |
| 106 | + bi = 0 |
| 107 | + for bi, line in _parse_lines_from_buffer(buf): |
| 108 | + yield line |
| 109 | + # for each line to parse from the buffer |
110 | 110 |
|
111 | | - # keep remainder |
112 | | - last_buf_list[0] = buf[bi:] |
| 111 | + # keep remainder |
| 112 | + last_buf_list[0] = buf[bi:] |
113 | 113 |
|
114 | 114 | def _dispatch_single_line(line, handler): |
115 | 115 | line = line.decode(defenc) |
@@ -193,14 +193,24 @@ def _deplete_buffer(fno, handler, buf_list, wg=None): |
193 | 193 | else: |
194 | 194 | # Oh ... probably we are on windows. select.select() can only handle sockets, we have files |
195 | 195 | # The only reliable way to do this now is to use threads and wait for both to finish |
| 196 | + def _handle_lines(fd, handler, wg): |
| 197 | + for line in fd: |
| 198 | + line = line.decode(defenc) |
| 199 | + if line and handler: |
| 200 | + handler(line) |
| 201 | + if wg: |
| 202 | + wg.done() |
| 203 | + |
196 | 204 | # Since the finalizer is expected to wait, we don't have to introduce our own wait primitive |
197 | 205 | # NO: It's not enough unfortunately, and we will have to sync the threads |
198 | 206 | wg = WaitGroup() |
199 | | - for fno, (handler, buf_list) in fdmap.items(): |
| 207 | + for fd, handler in zip((process.stdout, process.stderr), |
| 208 | + (stdout_handler, stderr_handler)): |
200 | 209 | wg.add(1) |
201 | | - t = threading.Thread(target=lambda: _deplete_buffer(fno, handler, buf_list, wg)) |
| 210 | + t = threading.Thread(target=_handle_lines, args=(fd, handler, wg)) |
| 211 | + t.setDaemon(True) |
202 | 212 | t.start() |
203 | | - # end |
| 213 | + |
204 | 214 | # NOTE: Just joining threads can possibly fail as there is a gap between .start() and when it's |
205 | 215 | # actually started, which could make the wait() call to just return because the thread is not yet |
206 | 216 | # active |
|