diff --git a/invoke/runners.py b/invoke/runners.py
index f1c888f44..d5c382a90 100644
--- a/invoke/runners.py
+++ b/invoke/runners.py
@@ -895,7 +895,8 @@ def handle_stdin(
                 if self.program_finished.is_set() and not data:
                     break
                 # Take a nap so we're not chewing CPU.
-                time.sleep(self.input_sleep)
+                if data is None:
+                    time.sleep(self.input_sleep)
 
     def should_echo_stdin(self, input_: IO, output: IO) -> bool:
         """
diff --git a/tests/runners.py b/tests/runners.py
index 94c63d8b3..fc06f6c07 100644
--- a/tests/runners.py
+++ b/tests/runners.py
@@ -8,6 +8,7 @@
 import types
 from io import StringIO
 from io import BytesIO
+from io import TextIOBase
 from itertools import chain, repeat
 
 from pytest import raises, skip
@@ -1098,16 +1099,51 @@ def subclasses_can_override_input_sleep(self):
             class MyRunner(_Dummy):
                 input_sleep = 0.007
 
+            def fake_stdin_stream():
+                # The value "foo" is eventually returned.
+                yield "f"
+                # None values simulate waiting for input on stdin.
+                yield None
+                yield "o"
+                yield None
+                yield "o"
+                yield None
+                # Once the stream is closed, stdin continues to return empty strings.
+                while True:
+                    yield ''
+
+            class FakeStdin(TextIOBase):
+                def __init__(self, stdin):
+                    self.stream = stdin
+
+                def read(self, size):
+                    return next(self.stream)
+
             with patch("invoke.runners.time") as mock_time:
                 MyRunner(Context()).run(
                     _,
-                    in_stream=StringIO("foo"),
+                    in_stream=FakeStdin(fake_stdin_stream()),
                     out_stream=StringIO(),  # null output to not pollute tests
                 )
-            # Just make sure the first few sleeps all look good. Can't know
-            # exact length of list due to stdin worker hanging out til end of
-            # process. Still worth testing more than the first tho.
-            assert mock_time.sleep.call_args_list[:3] == [call(0.007)] * 3
+            # Just make sure the sleeps all look good.
+            # There are three here because we return three "None" in fake_stdin_stream.
+            assert mock_time.sleep.call_args_list == [call(0.007)] * 3
+
+        @mock_subprocess()
+        def populated_streams_do_not_sleep(self):
+            class MyRunner(_Dummy):
+                read_chunk_size = 1
+
+            runner = MyRunner(Context())
+            with patch("invoke.runners.time") as mock_time:
+                with patch.object(runner, "wait"):
+                    runner.run(
+                        _,
+                        in_stream=StringIO("lots of bytes to read"),
+                        out_stream=StringIO(),  # null output to not pollute tests
+                    )
+            # Sleep should not be called before we break.
+            assert len(mock_time.sleep.call_args_list) == 0
 
     class stdin_mirroring:
         def _test_mirroring(self, expect_mirroring, **kwargs):