Fix corner case with joining processes/queues (#1473)

If a process places items in a queue and the process is joined before the queue is emptied, it can lead to a deadlock under some circumstances. Closes AIRFLOW-61.

See for example: https://docs.python.org/3/library/multiprocessing.html#all-start-methods ("Joining processes that use queues")
http://stackoverflow.com/questions/31665328/python-3-multiprocessing-queue-deadlock-when-calling-join-before-the-queue-is-em
http://stackoverflow.com/questions/31708646/process-join-and-queue-dont-work-with-large-numbers
http://stackoverflow.com/questions/19071529/python-multiprocessing-125-list-never-finishes

airflow/jobs.py

@@ -751,16 +751,18 @@ def _execute(self):
                 self.logger.info("Starting {} scheduler jobs".format(len(jobs)))
                 for j in jobs:
                     j.start()
+
+                while any(j.is_alive() for j in jobs):
+                    while not tis_q.empty():
+                        ti_key, pickle_id = tis_q.get()
+                        dag = dagbag.dags[ti_key[0]]
+                        task = dag.get_task(ti_key[1])
+                        ti = TI(task, ti_key[2])
+                        self.executor.queue_task_instance(ti, pickle_id=pickle_id)
+
                 for j in jobs:
                     j.join()
 
-                while not tis_q.empty():
-                    ti_key, pickle_id = tis_q.get()
-                    dag = dagbag.dags[ti_key[0]]
-                    task = dag.get_task(ti_key[1])
-                    ti = TI(task, ti_key[2])
-                    self.executor.queue_task_instance(ti, pickle_id=pickle_id)
-
                 self.logger.info("Done queuing tasks, calling the executor's "
                               "heartbeat")
                 duration_sec = (datetime.now() - loop_start_dttm).total_seconds()