@Test public void testPutIfAbsentThreadSafe() { int numberOfElements = 100; int firstHalf = numberOfElements / 2; for (int i = 0; i < firstHalf; i++) { underTest.putIfAbsent(getEvaluatorExecutor("name" + i), i); } Map<Integer, Boolean> results = new ConcurrentHashMap<>(); new ThreadSafetyTester(numberOfElements) .withBlockToTest(i -> results.put(i, underTest.putIfAbsent(getEvaluatorExecutor("name" + i), i))) .run() .waitUntilFinished(); assertTrue(IntStream.range(0, firstHalf).noneMatch(results::get)); assertTrue(IntStream.range(firstHalf, numberOfElements).allMatch(results::get)); }
boolean putIfAbsent(EvaluatorExecutor evaluator, long resourceId) { long now = clock.getCurrentTimeMillis(); Element elementToSubmit = new Element(evaluator, resourceId, now); Element elementAlreadyPresent = submittedEvaluators.putIfAbsent(elementToSubmit, elementToSubmit); if (elementAlreadyPresent == null) { return true; } else if (now - elementAlreadyPresent.getTimestampMillis() > timeout) { LOGGER.info("Timeout after {}: {} for cluster {} still present, resubmitting", now - elementAlreadyPresent.getTimestampMillis(), elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId()); return true; } LOGGER.debug("Submitting {} for cluster {} failed, it has been present in the system for {} ms", elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId(), now - elementAlreadyPresent.getTimestampMillis()); return false; }
EvaluatorExecutorRegistry { boolean putIfAbsent(EvaluatorExecutor evaluator, long resourceId) { long now = clock.getCurrentTimeMillis(); Element elementToSubmit = new Element(evaluator, resourceId, now); Element elementAlreadyPresent = submittedEvaluators.putIfAbsent(elementToSubmit, elementToSubmit); if (elementAlreadyPresent == null) { return true; } else if (now - elementAlreadyPresent.getTimestampMillis() > timeout) { LOGGER.info("Timeout after {}: {} for cluster {} still present, resubmitting", now - elementAlreadyPresent.getTimestampMillis(), elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId()); return true; } LOGGER.debug("Submitting {} for cluster {} failed, it has been present in the system for {} ms", elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId(), now - elementAlreadyPresent.getTimestampMillis()); return false; } }
EvaluatorExecutorRegistry { boolean putIfAbsent(EvaluatorExecutor evaluator, long resourceId) { long now = clock.getCurrentTimeMillis(); Element elementToSubmit = new Element(evaluator, resourceId, now); Element elementAlreadyPresent = submittedEvaluators.putIfAbsent(elementToSubmit, elementToSubmit); if (elementAlreadyPresent == null) { return true; } else if (now - elementAlreadyPresent.getTimestampMillis() > timeout) { LOGGER.info("Timeout after {}: {} for cluster {} still present, resubmitting", now - elementAlreadyPresent.getTimestampMillis(), elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId()); return true; } LOGGER.debug("Submitting {} for cluster {} failed, it has been present in the system for {} ms", elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId(), now - elementAlreadyPresent.getTimestampMillis()); return false; }  }
EvaluatorExecutorRegistry { boolean putIfAbsent(EvaluatorExecutor evaluator, long resourceId) { long now = clock.getCurrentTimeMillis(); Element elementToSubmit = new Element(evaluator, resourceId, now); Element elementAlreadyPresent = submittedEvaluators.putIfAbsent(elementToSubmit, elementToSubmit); if (elementAlreadyPresent == null) { return true; } else if (now - elementAlreadyPresent.getTimestampMillis() > timeout) { LOGGER.info("Timeout after {}: {} for cluster {} still present, resubmitting", now - elementAlreadyPresent.getTimestampMillis(), elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId()); return true; } LOGGER.debug("Submitting {} for cluster {} failed, it has been present in the system for {} ms", elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId(), now - elementAlreadyPresent.getTimestampMillis()); return false; }  void remove(EvaluatorExecutor evaluator, long clusterId); int activeCount(); }
EvaluatorExecutorRegistry { boolean putIfAbsent(EvaluatorExecutor evaluator, long resourceId) { long now = clock.getCurrentTimeMillis(); Element elementToSubmit = new Element(evaluator, resourceId, now); Element elementAlreadyPresent = submittedEvaluators.putIfAbsent(elementToSubmit, elementToSubmit); if (elementAlreadyPresent == null) { return true; } else if (now - elementAlreadyPresent.getTimestampMillis() > timeout) { LOGGER.info("Timeout after {}: {} for cluster {} still present, resubmitting", now - elementAlreadyPresent.getTimestampMillis(), elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId()); return true; } LOGGER.debug("Submitting {} for cluster {} failed, it has been present in the system for {} ms", elementAlreadyPresent.getEvaluatorExecutor().getName(), elementAlreadyPresent.getResourceId(), now - elementAlreadyPresent.getTimestampMillis()); return false; }  void remove(EvaluatorExecutor evaluator, long clusterId); int activeCount();  }