Fix executor deadlock on IO errors

This change ensures that the executor does not deadlock on IO errors,
for example when IO does not support non-ascii characters.

poetry/installation/executor.py

@@ -131,32 +131,29 @@ class Executor(object):
             return
 
         if self._io.is_debug():
-            self._lock.acquire()
+            with self._lock:
                 section = self._sections[id(operation)]
                 section.write_line(line)
-            self._lock.release()
 
             return
 
-        self._lock.acquire()
+        with self._lock:
             section = self._sections[id(operation)]
             section.output.clear()
             section.write(line)
-        self._lock.release()
 
     def _execute_operation(self, operation):
         try:
             if self.supports_fancy_output():
                 if id(operation) not in self._sections:
                     if self._should_write_operation(operation):
-                        self._lock.acquire()
+                        with self._lock:
                             self._sections[id(operation)] = self._io.section()
                             self._sections[id(operation)].write_line(
                                 "  <fg=blue;options=bold>•</> {message}: <fg=blue>Pending...</>".format(
                                     message=self.get_operation_message(operation),
                                 ),
                             )
-                        self._lock.release()
             else:
                 if self._should_write_operation(operation):
                     if not operation.skipped:
@@ -190,6 +187,7 @@ class Executor(object):
             if result == -2:
                 raise KeyboardInterrupt
         except Exception as e:
+            try:
                 from clikit.ui.components.exception_trace import ExceptionTrace
 
                 if not self.supports_fancy_output():
@@ -201,15 +199,15 @@ class Executor(object):
                     self._write(operation, message)
                     io = self._sections.get(id(operation), self._io)
 
-            self._lock.acquire()
-
+                with self._lock:
                     trace = ExceptionTrace(e)
                     trace.render(io)
                     io.write_line("")
-
+            finally:
+                with self._lock:
                     self._shutdown = True
-            self._lock.release()
         except KeyboardInterrupt:
+            try:
                 message = "  <warning>•</warning> {message}: <warning>Cancelled</warning>".format(
                     message=self.get_operation_message(operation, warning=True),
                 )
@@ -217,10 +215,9 @@ class Executor(object):
                     self._io.write_line(message)
                 else:
                     self._write(operation, message)
-
-            self._lock.acquire()
+            finally:
+                with self._lock:
                     self._shutdown = True
-            self._lock.release()
 
     def _do_execute_operation(self, operation):
         method = operation.job_type
@@ -266,15 +263,13 @@ class Executor(object):
         return result
 
     def _increment_operations_count(self, operation, executed):
-        self._lock.acquire()
+        with self._lock:
             if executed:
                 self._executed_operations += 1
                 self._executed[operation.job_type] += 1
             else:
                 self._skipped[operation.job_type] += 1
 
-        self._lock.release()
-
     def run_pip(self, *args, **kwargs):  # type: (...) -> int
         try:
             self._env.run_pip(*args, **kwargs)
@@ -622,9 +617,8 @@ class Executor(object):
                 progress.set_format(message + " <b>%percent%%</b>")
 
         if progress:
-            self._lock.acquire()
+            with self._lock:
                 progress.start()
-            self._lock.release()
 
         done = 0
         archive = self._chef.get_cache_directory_for_link(link) / link.filename
@@ -637,16 +631,14 @@ class Executor(object):
                 done += len(chunk)
 
                 if progress:
-                    self._lock.acquire()
+                    with self._lock:
                         progress.set_progress(done)
-                    self._lock.release()
 
                 f.write(chunk)
 
         if progress:
-            self._lock.acquire()
+            with self._lock:
                 progress.finish()
-            self._lock.release()
 
         return archive
 

tests/installation/test_executor.py

@@ -185,6 +185,32 @@ Package operations: 1 install, 0 updates, 0 removals
     assert expected == io.fetch_output()
 
 
+def test_execute_should_gracefully_handle_io_error(config, mocker, io):
+    env = MockEnv()
+    executor = Executor(env, pool, config, io)
+    executor.verbose()
+
+    original_write_line = executor._io.write_line
+
+    def write_line(string, flags=None):
+        # Simulate UnicodeEncodeError
+        string.encode("ascii")
+        original_write_line(string, flags)
+
+    mocker.patch.object(io, "write_line", side_effect=write_line)
+
+    assert 1 == executor.execute([Install(Package("clikit", "0.2.3"))])
+
+    expected = r"""
+Package operations: 1 install, 0 updates, 0 removals
+
+
+\s*Unicode\w+Error
+"""
+
+    assert re.match(expected, io.fetch_output())
+
+
 def test_executor_should_delete_incomplete_downloads(
     config, io, tmp_dir, mocker, pool, mock_file_downloads
 ):