Skip to content
Snippets Groups Projects
Unverified Commit f2134b28 authored by Rodrigo Queiro's avatar Rodrigo Queiro Committed by GitHub
Browse files

Avoid a race condition in RetryingExecutorService (#278)

This fixes #274, which occurs when the task completes before the
`Future` is added to the `callables` map (because the corresponding
submit() is still executing). In that case, `callable` is null, which
causes latches.get(callable) to throw an NPE.

The bug can be reproduced with the added test by adding
`Thread.sleep(1000)` below the `completionService.submit` call.
parent 1dd68e36
No related branches found
No related tags found
No related merge requests found
...@@ -63,7 +63,13 @@ public class RetryingExecutorService { ...@@ -63,7 +63,13 @@ public class RetryingExecutorService {
@Override @Override
public void loop() throws InterruptedException { public void loop() throws InterruptedException {
Future<Boolean> future = completionService.take(); Future<Boolean> future = completionService.take();
final Callable<Boolean> callable = callables.remove(future); Callable<Boolean> callable;
CountDownLatch latch;
// Grab the mutex to make sure submit() of the future that we took is finished.
synchronized (mutex) {
callable = callables.remove(future);
latch = latches.get(callable);
}
boolean retry; boolean retry;
try { try {
retry = future.get(); retry = future.get();
...@@ -74,14 +80,15 @@ public class RetryingExecutorService { ...@@ -74,14 +80,15 @@ public class RetryingExecutorService {
if (DEBUG) { if (DEBUG) {
log.info("Retry requested."); log.info("Retry requested.");
} }
final Callable<Boolean> finalCallable = callable;
scheduledExecutorService.schedule(new Runnable() { scheduledExecutorService.schedule(new Runnable() {
@Override @Override
public void run() { public void run() {
submit(callable); submit(finalCallable);
} }
}, retryDelay, retryTimeUnit); }, retryDelay, retryTimeUnit);
} else { } else {
latches.get(callable).countDown(); latch.countDown();
} }
} }
} }
......
/*
* Copyright (C) 2012 Google Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
* use this file except in compliance with the License. You may obtain a copy of
* the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
* License for the specific language governing permissions and limitations under
* the License.
*/
package org.ros.concurrent;
import static org.mockito.Mockito.*;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import org.junit.Before;
import org.junit.Test;
/**
* @author rodrigoq@google.com (Rodrigo Queiro)
*/
public class RetryingExecutorServiceTest {
private ScheduledExecutorService executorService;
@Before
public void before() {
executorService = Executors.newScheduledThreadPool(4);
}
@Test
public void testNoRetry_calledOnce() throws Exception {
RetryingExecutorService service = new RetryingExecutorService(executorService);
Callable<Boolean> callable = mock(Callable.class);
when(callable.call()).thenReturn(false);
service.submit(callable);
service.shutdown(10, TimeUnit.SECONDS);
verify(callable, times(1)).call();
}
@Test
public void testOneRetry_calledTwice() throws Exception {
RetryingExecutorService service = new RetryingExecutorService(executorService);
service.setRetryDelay(0, TimeUnit.SECONDS);
Callable<Boolean> callable = mock(Callable.class);
when(callable.call()).thenReturn(true).thenReturn(false);
service.submit(callable);
// Call verify() with a timeout before calling shutdown, as shutdown() will prevent further
// retries.
verify(callable, timeout(10000).times(2)).call();
service.shutdown(10, TimeUnit.SECONDS);
}
}
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment