fix: fix interrupt spiral in grpc ReadObject drainQueue - #2850
Conversation
Sorry, something went wrong.
If our thread is interrupted while attempting to drainQueue poll will throw an InterruptedException, instead of setting the flag back on the thread immediately we need to defer setting it until we complete our draining. If we don't defer setting it, we can never actually drain our queue.
There was a problem hiding this comment.
LGTM, 1 question before approval.
Sorry, something went wrong.
| @@ -234,35 +234,42 @@ public void close() throws IOException { | |||
| } | |||
|
|
|||
| private void drainQueue() throws IOException { | |||
There was a problem hiding this comment.
IIUC, we added this queuing code as a response to b/364531464. Can you please clarify what was missed or why the previous implementation didn't work as expected? Just want to better understand the cause/intended fix/new fix.
Also, Can we add some tests?
Sorry, something went wrong.
There was a problem hiding this comment.
The sentences before the code sample in the description outline why this is necessary and was was missed the first time around. The code attached to this PR simulates the scenario which caused the interrupt spiral, unfortunately there isn't a reliable way for us to force the interrupt to happen during drainQueue while the loop is processing. It's a multithreaded race scenario.
Sorry, something went wrong.
There was a problem hiding this comment.
Some extra context for posterity from chatting with Ben + Syd: The code here was created to replace gax and add some extra memory management features.
The actual bug is in the while loop that causes the interruption to catch its own exception indefinitely (hence moving the InterruptedException inside the loop)
Sorry, something went wrong.
If our thread is interrupted while attempting to drainQueue poll will throw an InterruptedException, instead of setting the flag back on the thread immediately we need to defer setting it until we complete our draining. If we don't defer setting it, we can never actually drain our queue.
Manually tested repro and fix with the following code sample, which mirrors the pattern in drainQueue.
import java.io.IOException; import java.io.InterruptedIOException; import java.util.concurrent.ArrayBlockingQueue; final class InterruptSpiral { public static void main(String[] args) { final ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<>(2); Thread main = Thread.currentThread(); Thread thread = new Thread(() -> { try { queue.add("String 1"); main.interrupt(); Thread.sleep(2500); queue.add("String 2"); } catch (InterruptedException ignore) {} }); IOException ioException = null; boolean shouldInterupt = false; thread.start(); int taken = 0; for (int i = 0; i < 5; i++) { System.out.println("while"); System.out.println("main.isInterrupted() = " + main.isInterrupted()); try { System.out.println("pre = " + System.nanoTime()); String take = queue.take(); taken++; System.out.println("post = " + System.nanoTime()); System.out.println("take = " + take); if (taken == 2) { break; } } catch (InterruptedException e) { // Thread.currentThread().interrupt(); <-- the bug shouldInterupt = true; if (ioException == null) { ioException = new InterruptedIOException(); } else { ioException.addSuppressed(e); } } } if (shouldInterupt) { Thread.currentThread().interrupt(); } System.out.println("main.isInterrupted() = " + main.isInterrupted()); thread.interrupt(); ioException.printStackTrace(); } }Without the fix, prints the following:
With the fix, prints the following: