title: Can very long I/O processes be handled by threads
body: <p>I'm starting here a new topic that will be linked with <a href="https://stackoverflow.com/questions/47250025/qthreadpool-how-to-interrupt-how-to-use-wisely-the-waitfordone-method">this question</a>.</p>
  
  <p>I invite you to just read on the background in order to get the global idea.</p>
  
  <p>So I’ve a download function which relies on a python 3.2 API (developed by a private company). The process can take up to 400 seconds per file.</p>
  
  <p>Obviously, I don't have only one file to download, so I've been trying for days to put every download process in a thread pool. Each thread in the pool should be completely autonomous from the GUI Main Thread. When one of them has finished, it should just send a signal to the GUI.</p>
  
  <p>I did several tests but whatever the technique used, but </p>
  
  <ol>
  <li>the GUI is freezing;</li>
  <li>the result is only given at the end of the processing of all threads and not – as wanted – one by one.</li>
  </ol>
  
  <p>I think that the download method given by the API is a blocking function that can't be threaded.</p>
  
  <p>So my question is simple: how to know if a I/O method can be handled through a thread.</p>
  
  <hr>
  
  <p>November 24,2017 Update</p>
  
  <p>You will find below a first draft (with the tandem multiprocessing.pool / map_async) that partially meets my expectations. As you will see, I unfortunately had to insert a "Busy Waiting Loop" in order to get on the QPlainTextEdit some information on what was going on.</p>
  
  <p>The results of the tasks are given only at the end of the global processing (behaviour map_async). That's not exactly what I'm looking for. I would like to insert a little more real time and see for each completed task its message immediately on the console.</p>
  
  <pre><code>import time
  import multiprocessing
  import private.library as bathy
  from PyQt4 import QtCore, QtGui
  import os
  import sys
  
  user = 'user'
  password = 'password'
  server = 'server'
  basename = 'basename'
  
  workers = multiprocessing.cpu_count()
  
  node = bathy.NodeManager(user, password, server)
  database = node.get_database(basename)
  
  ids = (10547, 3071, 13845, 13846, 13851, 13844, 5639, 4612, 4613, 954,
         961, 962, 4619, 4620, 4622, 4623, 4624, 4627, 4628, 4631,
         4632, 4634, 4635, 4638, 4639, 4640, 4641, 4642, 10722, 1300,
         1301, 1303, 1310, 1319, 1316, 1318, 1321, 1322, 1323, 1324,
         1325, 1347, 1348, 1013, 1015, 1320, 8285, 8286, 8287, 10329,
         9239, 9039, 5006, 5009, 5011, 5012, 5013, 5014, 5015, 5025,
         5026, 4998, 5040, 5041, 5042, 5043, 11811, 2463, 2464, 5045,
         5046, 5047, 5048, 5049, 5053, 5060, 5064, 5065, 5068, 5069,
         5071, 5072, 5075, 5076, 5077, 5079, 5080, 5081, 5082, 5083,
         5084, 5085, 5086, 5087, 5088, 5090, 5091, 5092, 5093)
  
  
  # ---------------------------------------------------------------------------------
  def download(surface_id, index):
      global node
      global database
  
      t = time.time()
      message = 'Surface #%d - Process started\n' % index
  
      surface = database.get_surface(surface_id)
      metadata = surface.get_metadata()
      file_path = os.path.join("C:\\Users\\philippe\\Test_Download",
                               metadata["OBJNAM"] + ".surf")
  
      try:
          surface.download_bathymetry(file_path)
      except RuntimeError as error:
          message += "Error : " + str(error).split('\n')[0] + '\n'
      finally:
          message += ('Process ended : %.2f s\n' % (time.time() - t))
  
      return message
  
  
  # ---------------------------------------------------------------------------------
   def pass_args(args):
      # Method to pass multiple arguments to download (multiprocessing.Pool)
      return download(*args)
  
  
  # ---------------------------------------------------------------------------------
  class Console(QtGui.QDialog):
      def __init__(self):
          super(self.__class__, self).__init__()
  
          self.resize(600, 300)
          self.setMinimumSize(QtCore.QSize(600, 300))
          self.setWindowTitle("Console")
          self.setModal(True)
  
          self.verticalLayout = QtGui.QVBoxLayout(self)
  
          # Text edit
          # -------------------------------------------------------------------------
  
          self.text_edit = QtGui.QPlainTextEdit(self)
          self.text_edit.setReadOnly(True)
          self.text_edit_cursor = QtGui.QTextCursor(self.text_edit.document())
          self.verticalLayout.addWidget(self.text_edit)
  
          # Ok / Close
          # -------------------------------------------------------------------------
          self.button_box = QtGui.QDialogButtonBox(self)
          self.button_box.setStandardButtons(QtGui.QDialogButtonBox.Close | 
                                             QtGui.QDialogButtonBox.Ok)
          self.button_box.setObjectName("button_box")
          self.verticalLayout.addWidget(self.button_box)
  
          # Connect definition
          # -------------------------------------------------------------------------
  
          self.connect(self.button_box.button(QtGui.QDialogButtonBox.Close),
                       QtCore.SIGNAL('clicked()'),
                       self.button_cancel_clicked)
          self.connect(self.button_box.button(QtGui.QDialogButtonBox.Ok),
                       QtCore.SIGNAL('clicked()'),
                       self.button_ok_clicked)
  
          # Post initialization
          # -------------------------------------------------------------------------
          self.pool = multiprocessing.Pool(processes=workers)
  
      # Connect functions
      # -----------------------------------------------------------------------------
      def button_cancel_clicked(self):
          self.close()
  
      def button_ok_clicked(self):
          jobs_args = [(surface_id, index) for index, surface_id in enumerate(ids)]
          async = pool.map_async(pass_args, jobs_args)
          pool.close()
  
          # Busy waiting loop
          while True:
              # pool.map_async has a _number_left attribute, and a ready() method
              if async.ready():
                  self.write_stream("All tasks completed\n")
                  pool.join()
                  for line in async.get():
                      self.write_stream(line)
                  break
  
              remaining = async._number_left
              self.write_stream("Waiting for %d task(s) to complete...\n" % remaining)
              time.sleep(0.5)
  
  
      # Other functions
      # -----------------------------------------------------------------------------
      def write_stream(self, text):
          self.text_edit.insertPlainText(text)
          cursor = self.text_edit.textCursor()
          self.text_edit.setTextCursor(cursor)
          app.processEvents()
  
  
  # ---------------------------------------------------------------------------------
  if __name__ == '__main__':
      app = QtGui.QApplication(sys.argv)
      window = Console()
      window.show()
      app.exec_()
  </code></pre>
  
  <p><strong>Questions</strong></p>
  
  <ol>
  <li>At first glance does the above code for you present conceptual errors?</li>
  <li>Do I have to use the apply_async method in this specific case to get something more interactive?</li>
  <li>Could you guide me on how to use a callback function to post a custom event to update the console (methodology suggested by @ekhumoro)?</li>
  </ol>
  
  <hr>
  
  <p>November 25,2017 Update</p>
  
  <p>I had a try with apply_async:</p>
  
  <pre><code>def button_ok_clicked(self):
      # Pool.apply_async - the call returns immediately instead of 
      # waiting for the result
      for index, surface_id in enumerate(ids):
          async = pool.apply_async(download, 
                                   args=(surface_id, index),
                                   callback=self.write_stream)
      pool.close()
  </code></pre>
  
  <p>with a callback:</p>
  
  <pre><code>def write_stream(self, text):
      # This is called whenever pool.apply_async(i) returns a result
      self.text_edit.insertPlainText(text)
      cursor = self.text_edit.textCursor()
      self.text_edit.setTextCursor(cursor)
      # Update the text edit
      app.processEvents()
  </code></pre>
  
  <p>Unfortunately, by doing this way the application crashes. I think I'll have to put a lock mechanism to prevent all the tasks from writing in the text edit at the same time.</p>