Changeset 18821 in josm for trunk

Timestamp:

2023-08-24T17:23:11+02:00 (14 months ago)

Author:

taylor.smock

Message:

Fix #23140: RejectedExecutionException when MultiFetchServerObjectReader is cancelled while creating download jobs

This was caused by a race condition.
User starts a download, and cancels it as jobs are submitted to the thread pool.
The next job to be submitted will cause a RejectedExecutionException. In order
to fix this, we ensure that we shutdown the thread pool inside a synchronized
block, and add jobs to the executor inside a synchronized block.

Location:

trunk

Files:

• src/org/openstreetmap/josm/io/MultiFetchServerObjectReader.java (modified) (9 diffs)
• test/functional/org/openstreetmap/josm/io/MultiFetchServerObjectReaderTest.java (modified) (12 diffs)
• test/unit/org/openstreetmap/josm/testutils/annotations/OsmApi.java (added)
• test/unit/org/openstreetmap/josm/testutils/annotations/TestUser.java (added)

trunk/src/org/openstreetmap/josm/io/MultiFetchServerObjectReader.java

@@ -48 +48 @@
  * Retrieves a set of {@link OsmPrimitive}s from an OSM server using the so called
  * Multi Fetch API.
- *
+ * <p>
  * Usage:
  * <pre>
@@ -128 +128 @@
      * Remembers an {@link OsmPrimitive}'s id. The id will
      * later be fetched as part of a Multi Get request.
-     *
+     * <p>
      * Ignore the id if it represents a new primitives.
      *
@@ -326 +326 @@
         CompletionService<FetchResult> ecs = new ExecutorCompletionService<>(exec);
         List<Future<FetchResult>> jobs = new ArrayList<>();
-        while (!toFetch.isEmpty() && !isCanceled()) {
-            jobs.add(ecs.submit(new Fetcher(type, extractIdPackage(toFetch), progressMonitor)));
+        // There exists a race condition where this is cancelled after isCanceled is called, such that
+        // the exec ThreadPool has been shut down. This can cause a RejectedExecutionException.
+        synchronized (this) {
+            while (!toFetch.isEmpty() && !isCanceled()) {
+                jobs.add(ecs.submit(new Fetcher(type, extractIdPackage(toFetch), progressMonitor)));
+            }
         }
         // Run the fetchers
@@ -348 +352 @@
                 }
             } catch (InterruptedException | ExecutionException e) {
+                if (e instanceof InterruptedException) {
+                    Thread.currentThread().interrupt();
+                }
                 Logging.error(e);
                 if (e.getCause() instanceof OsmTransferException)
@@ -369 +376 @@
      * the latest version of the primitive (if any), even if the primitive is not visible (i.e. if
      * visible==false).
-     *
+     * <p>
      * Invoke {@link #getMissingPrimitives()} to get a list of primitives which have not been
      * found on  the server (the server response code was 404)
@@ -589 +596 @@
                     }
                     if (pkg.size() == 1) {
-                        FetchResult res = new FetchResult(new DataSet(), new HashSet<PrimitiveId>());
+                        FetchResult res = new FetchResult(new DataSet(), new HashSet<>());
                         res.missingPrimitives.add(new SimplePrimitiveId(pkg.iterator().next(), type));
                         return res;
@@ -668 +675 @@
          * invokes a sequence of Multi Gets for individual ids in a set of ids and a given {@link OsmPrimitiveType}.
          * The retrieved primitives are merged to {@link #outputDataSet}.
-         *
+         * <p>
          * This method is used if one of the ids in pkg doesn't exist (the server replies with return code 404).
          * If the set is fetched with this method it is possible to find out which of the ids doesn't exist.
@@ -682 +689 @@
         protected FetchResult singleGetIdPackage(OsmPrimitiveType type, Set<Long> pkg, ProgressMonitor progressMonitor)
                 throws OsmTransferException {
-            FetchResult result = new FetchResult(new DataSet(), new HashSet<PrimitiveId>());
+            FetchResult result = new FetchResult(new DataSet(), new HashSet<>());
             String baseUrl = OsmApi.getOsmApi().getBaseUrl();
             for (long id : pkg) {
@@ -713 +720 @@
     public void cancel() {
         super.cancel();
-        if (exec != null)
-            exec.shutdownNow();
+        // Synchronized to avoid a RejectedExecutionException in fetchPrimitives
+        // We don't want to synchronize on the super.cancel() call.
+        synchronized (this) {
+            if (exec != null) {
+                exec.shutdownNow();
+            }
+        }
     }
 }

trunk/test/functional/org/openstreetmap/josm/io/MultiFetchServerObjectReaderTest.java

@@ -5 +5 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assumptions.assumeTrue;
 
 import java.io.File;
@@ -23 +23 @@
 import java.util.Random;
 import java.util.TreeSet;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Logger;
 
+import org.awaitility.Awaitility;
+import org.awaitility.Durations;
+import org.awaitility.core.ConditionTimeoutException;
+import org.hamcrest.Matchers;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
-import org.junit.jupiter.api.extension.RegisterExtension;
-import org.openstreetmap.josm.JOSMFixture;
-import org.openstreetmap.josm.TestUtils;
 import org.openstreetmap.josm.data.coor.LatLon;
 import org.openstreetmap.josm.data.osm.Changeset;
@@ -43 +48 @@
 import org.openstreetmap.josm.data.osm.Way;
 import org.openstreetmap.josm.gui.progress.NullProgressMonitor;
-import org.openstreetmap.josm.spi.preferences.Config;
-import org.openstreetmap.josm.testutils.JOSMTestRules;
+import org.openstreetmap.josm.gui.util.GuiHelper;
+import org.openstreetmap.josm.testutils.annotations.TestUser;
+import org.openstreetmap.josm.tools.Logging;
 
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
@@ -53 +59 @@
 @SuppressFBWarnings(value = "CRLF_INJECTION_LOGS")
 @Timeout(value = 1, unit = TimeUnit.MINUTES)
+@org.openstreetmap.josm.testutils.annotations.OsmApi(org.openstreetmap.josm.testutils.annotations.OsmApi.APIType.DEV)
+@TestUser
 class MultiFetchServerObjectReaderTest {
     private static final Logger logger = Logger.getLogger(MultiFetchServerObjectReader.class.getName());
-
-    /**
-     * Setup test.
-     */
-    @RegisterExtension
-    @SuppressFBWarnings(value = "URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD")
-    public JOSMTestRules test = new JOSMTestRules().preferences();
 
     /**
@@ -159 +160 @@
     @BeforeAll
     public static void init() throws Exception {
-        if (!TestUtils.areCredentialsProvided()) {
-            logger.severe("OSM DEV API credentials not provided. Please define them with -Dosm.username and -Dosm.password");
-            return;
-        }
         logger.info("initializing ...");
-        JOSMFixture.createFunctionalTestFixture().init();
-
-        Config.getPref().put("osm-server.auth-method", "basic");
-
-        // don't use atomic upload, the test API server can't cope with large diff uploads
-        Config.getPref().putBoolean("osm-server.atomic-upload", false);
 
         File dataSetCacheOutputFile = new File(System.getProperty("java.io.tmpdir"),
@@ -213 +204 @@
     @BeforeEach
     public void setUp() throws IOException, IllegalDataException, FileNotFoundException {
-        if (!TestUtils.areCredentialsProvided()) {
-            return;
-        }
         File f = new File(System.getProperty("java.io.tmpdir"), MultiFetchServerObjectReaderTest.class.getName() + ".dataset");
         logger.info(MessageFormat.format("reading cached dataset ''{0}''", f.toString()));
@@ -230 +218 @@
     @Test
     void testMultiGet10Nodes() throws OsmTransferException {
-        assumeTrue(TestUtils.areCredentialsProvided());
         MultiFetchServerObjectReader reader = new MultiFetchServerObjectReader();
         ArrayList<Node> nodes = new ArrayList<>(ds.getNodes());
@@ -252 +239 @@
     @Test
     void testMultiGet10Ways() throws OsmTransferException {
-        assumeTrue(TestUtils.areCredentialsProvided());
         MultiFetchServerObjectReader reader = new MultiFetchServerObjectReader();
         ArrayList<Way> ways = new ArrayList<>(ds.getWays());
@@ -275 +261 @@
     @Test
     void testMultiGet10Relations() throws OsmTransferException {
-        assumeTrue(TestUtils.areCredentialsProvided());
         MultiFetchServerObjectReader reader = new MultiFetchServerObjectReader();
         ArrayList<Relation> relations = new ArrayList<>(ds.getRelations());
@@ -298 +283 @@
     @Test
     void testMultiGet800Nodes() throws OsmTransferException {
-        assumeTrue(TestUtils.areCredentialsProvided());
         MultiFetchServerObjectReader reader = new MultiFetchServerObjectReader();
         ArrayList<Node> nodes = new ArrayList<>(ds.getNodes());
@@ -320 +304 @@
     @Test
     void testMultiGetWithNonExistingNode() throws OsmTransferException {
-        assumeTrue(TestUtils.areCredentialsProvided());
         MultiFetchServerObjectReader reader = new MultiFetchServerObjectReader();
         ArrayList<Node> nodes = new ArrayList<>(ds.getNodes());
@@ -349 +332 @@
         assertEquals("ways?ways=123,126,130", requestString);
     }
+
+    /**
+     * This is a non-regression test for #23140: Cancelling `MultiFetchServerObjectReader` while it is adding jobs
+     * to the executor causes a {@link RejectedExecutionException}.
+     * This was caused by a race condition between {@link MultiFetchServerObjectReader#cancel()} and queuing download
+     * jobs.
+     */
+    @Test
+    void testCancelDuringJobAdd() {
+        final AtomicBoolean parsedData = new AtomicBoolean();
+        final AtomicBoolean continueAddition = new AtomicBoolean();
+        final AtomicInteger callCounter = new AtomicInteger();
+        final AtomicReference<Throwable> thrownFailure = new AtomicReference<>();
+        // We have 5 + 10 maximum (5 previous calls, 10 calls when iterating through the nodes).
+        final int expectedCancelCalls = 5;
+        final MultiFetchServerObjectReader reader = new MultiFetchServerObjectReader() {
+            @Override
+            public boolean isCanceled() {
+                final boolean result = super.isCanceled();
+                // There are some calls prior to the location where we are interested
+                if (callCounter.incrementAndGet() >= expectedCancelCalls) {
+                    // This will throw a ConditionTimeoutException.
+                    // By blocking here until cancel() is called, we block cancel (since we are interested in a loop).
+                    Awaitility.await().timeout(Durations.FIVE_HUNDRED_MILLISECONDS).untilTrue(continueAddition);
+                }
+                return result;
+            }
+        };
+        ArrayList<Node> nodes = new ArrayList<>(ds.getNodes());
+        for (int i = 0; i < 10; i++) {
+            reader.append(nodes.get(i));
+        }
+        GuiHelper.runInEDT(() -> {
+                try {
+                    reader.parseOsm(NullProgressMonitor.INSTANCE);
+                } catch (ConditionTimeoutException timeoutException) {
+                    // This is expected due to the synchronization, so we just swallow it.
+                    Logging.trace(timeoutException);
+                } catch (Exception failure) {
+                    thrownFailure.set(failure);
+                } finally {
+                    parsedData.set(true);
+                }
+            });
+        // cancel, then continue
+        Awaitility.await().untilAtomic(callCounter, Matchers.greaterThanOrEqualTo(expectedCancelCalls));
+        reader.cancel();
+        continueAddition.set(true);
+        Awaitility.await().untilTrue(parsedData);
+        if (thrownFailure.get() != null) {
+            Logging.error(thrownFailure.get());
+        }
+        assertNull(thrownFailure.get());
+        assertEquals(expectedCancelCalls, callCounter.get());
+    }
 }