Update ShellImpl task scheduling

- Prevent execTask starvation: tasks scheduled through execTask are
  now queued along with submitted tasks, executing in order of
  submission
- waitAndClose now properly waits for all tasks to complete, including
  both synchronous and asynchronous tasks

core/src/main/java/com/topjohnwu/superuser/internal/ShellImpl.java

@@ -21,6 +21,7 @@ import static java.nio.charset.StandardCharsets.UTF_8;
 import android.text.TextUtils;
 
 import androidx.annotation.NonNull;
+import androidx.annotation.Nullable;
 
 import com.topjohnwu.superuser.Shell;
 import com.topjohnwu.superuser.ShellUtils;
@@ -38,16 +39,48 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.FutureTask;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.ReentrantLock;
 
 class ShellImpl extends Shell {
     private volatile int status;
 
-    private final Process proc;
+    private final Process process;
     private final NoCloseOutputStream STDIN;
     private final NoCloseInputStream STDOUT;
     private final NoCloseInputStream STDERR;
+
+    // Guarded by scheduleLock
+    private final ReentrantLock scheduleLock = new ReentrantLock();
+    private final Condition idle = scheduleLock.newCondition();
     private final ArrayDeque<Task> tasks = new ArrayDeque<>();
-    private boolean runningTasks = false;
+    private boolean isRunningTask = false;
+
+    private static final class SyncTask implements Task {
+
+        private final Condition condition;
+        private boolean set = false;
+
+        SyncTask(Condition c) {
+            condition = c;
+        }
+
+        void signal() {
+            set = true;
+            condition.signal();
+        }
+
+        void await() {
+            while (!set) {
+                try {
+                    condition.await();
+                } catch (InterruptedException ignored) {}
+            }
+        }
+
+        @Override
+        public void run(OutputStream stdin, InputStream stdout, InputStream stderr) {}
+    }
 
     private static class NoCloseInputStream extends FilterInputStream {
 
@@ -84,12 +117,12 @@ class ShellImpl extends Shell {
         }
     }
 
-    ShellImpl(BuilderImpl builder, Process process) throws IOException {
+    ShellImpl(BuilderImpl builder, Process proc) throws IOException {
         status = UNKNOWN;
-        proc = process;
+        process = proc;
-        STDIN = new NoCloseOutputStream(process.getOutputStream());
+        STDIN = new NoCloseOutputStream(proc.getOutputStream());
-        STDOUT = new NoCloseInputStream(process.getInputStream());
+        STDOUT = new NoCloseInputStream(proc.getInputStream());
-        STDERR = new NoCloseInputStream(process.getErrorStream());
+        STDERR = new NoCloseInputStream(proc.getErrorStream());
 
         // Shell checks might get stuck indefinitely
         FutureTask<Integer> check = new FutureTask<>(this::shellCheck);
@@ -117,7 +150,7 @@ class ShellImpl extends Shell {
 
     private Integer shellCheck() throws IOException {
         try {
-            proc.exitValue();
+            process.exitValue();
             throw new IOException("Created process has terminated");
         } catch (IllegalThreadStateException ignored) {
             // Process is alive
@@ -156,7 +189,7 @@ class ShellImpl extends Shell {
         try { STDIN.close0(); } catch (IOException ignored) {}
         try { STDERR.close0(); } catch (IOException ignored) {}
         try { STDOUT.close0(); } catch (IOException ignored) {}
-        proc.destroy();
+        process.destroy();
     }
 
     @Override
@@ -164,19 +197,16 @@ class ShellImpl extends Shell {
         if (status < 0)
             return true;
 
-        synchronized (tasks) {
+        scheduleLock.lock();
-            if (runningTasks) {
+        try {
-                tasks.clear();
+            if (isRunningTask && !idle.await(timeout, unit))
-                tasks.wait(unit.toMillis(timeout));
+                return false;
-            }
+            close();
-            if (!runningTasks) {
+        } finally {
-                release();
+            scheduleLock.unlock();
-                return true;
-            }
         }
 
-        status = UNKNOWN;
+        return true;
-        return false;
     }
 
     @Override
@@ -198,8 +228,9 @@ class ShellImpl extends Shell {
             return false;
 
         try {
-            proc.exitValue();
+            process.exitValue();
             // Process is dead, shell is not alive
+            release();
             return false;
         } catch (IllegalThreadStateException e) {
             // Process is still running
@@ -228,43 +259,71 @@ class ShellImpl extends Shell {
     }
 
     private void processTasks() {
-        for (;;) {
+        Task task;
-            Task task;
+        while ((task = processNextTask(false)) != null) {
-            synchronized (tasks) {
-                if ((task = tasks.poll()) == null) {
-                    runningTasks = false;
-                    tasks.notifyAll();
-                    return;
-                }
-            }
             try {
                 exec0(task);
             } catch (IOException ignored) {}
         }
     }
 
+    @Nullable
+    private Task processNextTask(boolean fromExec) {
+        scheduleLock.lock();
+        try {
+            final Task task = tasks.poll();
+            if (task == null) {
+                isRunningTask = false;
+                idle.signalAll();
+                return null;
+            }
+            if (task instanceof SyncTask) {
+                ((SyncTask) task).signal();
+                return null;
+            }
+            if (fromExec) {
+                // Put the task back in front of the queue
+                tasks.offerFirst(task);
+            } else {
+                return task;
+            }
+        } finally {
+            scheduleLock.unlock();
+        }
+        EXECUTOR.execute(this::processTasks);
+        return null;
+    }
+
     @Override
     public void submitTask(@NonNull Task task) {
-        synchronized (tasks) {
+        scheduleLock.lock();
+        try {
             tasks.offer(task);
-            if (!runningTasks) {
+            if (!isRunningTask) {
-                runningTasks = true;
+                isRunningTask = true;
                 EXECUTOR.execute(this::processTasks);
             }
+        } finally {
+            scheduleLock.unlock();
         }
     }
 
     @Override
     public void execTask(@NonNull Task task) throws IOException {
-        synchronized (tasks) {
+        scheduleLock.lock();
-            while (runningTasks) {
+        try {
-                // Wait until all existing tasks are done
+            if (isRunningTask) {
-                try {
+                SyncTask sync = new SyncTask(scheduleLock.newCondition());
-                    tasks.wait();
+                tasks.offer(sync);
-                } catch (InterruptedException ignored) {}
+                // Wait until it's our turn
+                sync.await();
             }
+            isRunningTask = true;
+        } finally {
+            scheduleLock.unlock();
         }
         exec0(task);
+        processNextTask(true);
     }
 
     @NonNull

example/src/main/java/com/topjohnwu/libsuexample/MainActivity.java

@@ -260,6 +260,20 @@ public class MainActivity extends Activity implements Handler.Callback {
         binding.testAsync.setOnClickListener(v ->
                 Shell.cmd("test_async").to(consoleList).submit());
 
+        binding.testQueue.setOnClickListener(v -> {
+            Shell.getShell(Shell.EXECUTOR, s -> {
+                Log.i(TAG, "Queue: 1");
+                s.newJob().to(consoleList).add("sleep 1", "echo 1").submit();
+                Log.i(TAG, "Queue: 2");
+                s.newJob().to(consoleList).add("echo 2").exec();
+                Log.i(TAG, "Queue: 3");
+                s.newJob().to(consoleList).add("sleep 1", "echo 3").submit();
+                Log.i(TAG, "Queue: 4");
+                s.newJob().to(consoleList).add("echo 4").submit();
+                Log.i(TAG, "Queue: done");
+            });
+        });
+
         binding.clear.setOnClickListener(v -> binding.console.setText(""));
 
         binding.stressTest.setOnClickListener(v -> StressTest.perform(remoteFS));

example/src/main/res/layout/activity_main.xml

@@ -44,7 +44,7 @@
             android:layout_width="0dp"
             android:layout_height="wrap_content"
             android:layout_weight="1"
-            android:text="Shell Sync" />
+            android:text="Sync CMD" />
 
         <Button
             android:id="@+id/test_async"
@@ -52,7 +52,15 @@
             android:layout_width="0dp"
             android:layout_height="wrap_content"
             android:layout_weight="1"
-            android:text="Shell Async" />
+            android:text="Async CMD" />
+
+        <Button
+            android:id="@+id/test_queue"
+            style="?android:borderlessButtonStyle"
+            android:layout_width="0dp"
+            android:layout_height="wrap_content"
+            android:layout_weight="1"
+            android:text="Test Queue" />
 
         <Button
             android:id="@+id/close_shell"