Fix deadlock in SqlStageExecution.waitForNewExchangesOrBuffers

The waiting code only breaks if all sources are completed, but
once a source completes we must set no more splits on the tasks.
Upon review of the waiting code, this entire method can be replaced
with a simple timed wait.

presto-main/src/main/java/com/facebook/presto/execution/SqlStageExecution.java

@@ -775,7 +775,19 @@ private void updateNewExchangesAndBuffers(boolean waitUntilFinished)
                 return;
             }
 
-            waitForNewExchangesOrBuffers();
+            synchronized (this) {
+                // wait for a state change
+                //
+                // NOTE this must be a wait with a timeout since there is no notification
+                // for new exchanges from the child stages
+                try {
+                    TimeUnit.MILLISECONDS.timedWait(this, 100);
+                }
+                catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    throw Throwables.propagate(e);
+                }
+            }
         }
     }
 
@@ -811,39 +823,6 @@ private boolean addNewExchangesAndBuffers()
         return finished;
     }
 
-    private synchronized void waitForNewExchangesOrBuffers()
-    {
-        while (!getState().isDone()) {
-            // if next loop will finish, don't wait
-            Set<PlanNodeId> completeSources = updateCompleteSources();
-            boolean allSourceComplete = completeSources.containsAll(allSources);
-            if (allSourceComplete && getCurrentOutputBuffers().isNoMoreBufferIds()) {
-                return;
-            }
-            // do we have a new set of output buffers?
-            if (nextOutputBuffers != null) {
-                return;
-            }
-
-            // do we have new exchange locations?
-            if (!getNewExchangeLocations().isEmpty()) {
-                return;
-            }
-
-            // wait for a state change
-            //
-            // NOTE this must be a wait with a timeout since there is no notification
-            // for new exchanges from the child stages
-            try {
-                TimeUnit.MILLISECONDS.timedWait(this, 100);
-            }
-            catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw Throwables.propagate(e);
-            }
-        }
-    }
-
     private Set<PlanNodeId> updateCompleteSources()
     {
         for (RemoteSourceNode remoteSourceNode : fragment.getRemoteSourceNodes()) {