FSLeafQueue extends FSQueue { void addApp(FSAppAttempt app, boolean runnable) { writeLock.lock(); try { if (runnable) { runnableApps.add(app); } else { nonRunnableApps.add(app); } assignedApps.remove(app.getApplicationId()); incUsedResource(app.getResourceUsage()); } finally { writeLock.unlock(); } } FSLeafQueue(String name, FairScheduler scheduler,
      FSParentQueue parent); @Override void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override Resource getDemand(); @Override void updateDemand(); @Override Resource assignContainer(FSSchedulerNode node); @Override List<FSQueue> getChildQueues(); @Override List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); @Override int getNumRunnableApps(); int getNumPendingApps(); int getNumAssignedApps(); int getNumActiveApps(); @Override ActiveUsersManager getAbstractUsersManager(); boolean canRunAppAM(Resource amResource); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt schedulerAttempt, RMContainer rmContainer); void setWeights(float weight); @Override Resource getMaximumContainerAllocation(); void addAssignedApp(ApplicationId applicationId); void removeAssignedApp(ApplicationId applicationId);  }
@Test public void testConcurrentAccess() { conf.set(FairSchedulerConfiguration.ASSIGN_MULTIPLE, "false"); resourceManager = new MockRM(conf); resourceManager.start(); scheduler = (FairScheduler) resourceManager.getResourceScheduler(); String queueName = "root.queue1"; final FSLeafQueue schedulable = scheduler.getQueueManager(). getLeafQueue(queueName, true); ApplicationAttemptId applicationAttemptId = createAppAttemptId(1, 1); RMContext rmContext = resourceManager.getRMContext(); final FSAppAttempt app = new FSAppAttempt(scheduler, applicationAttemptId, "user1", schedulable, null, rmContext); int testThreads = 2; List<Runnable> runnables = new ArrayList<Runnable>(); runnables.add(new Runnable() { @Override public void run() { for (int i=0; i < 500; i++) { schedulable.addApp(app, true); } } }); runnables.add(new Runnable() { @Override public void run() { for (int i=0; i < 500; i++) { schedulable.getResourceUsage(); } } }); final List<Throwable> exceptions = Collections.synchronizedList( new ArrayList<Throwable>()); final ExecutorService threadPool = HadoopExecutors.newFixedThreadPool( testThreads); try { final CountDownLatch allExecutorThreadsReady = new CountDownLatch(testThreads); final CountDownLatch startBlocker = new CountDownLatch(1); final CountDownLatch allDone = new CountDownLatch(testThreads); for (final Runnable submittedTestRunnable : runnables) { threadPool.submit(new Runnable() { public void run() { allExecutorThreadsReady.countDown(); try { startBlocker.await(); submittedTestRunnable.run(); } catch (final Throwable e) { exceptions.add(e); } finally { allDone.countDown(); } } }); } allExecutorThreadsReady.await(); startBlocker.countDown(); int testTimeout = 2; assertTrue("Timeout waiting for more than " + testTimeout + " seconds", allDone.await(testTimeout, TimeUnit.SECONDS)); } catch (InterruptedException ie) { exceptions.add(ie); } finally { threadPool.shutdownNow(); } assertTrue("Test failed with exception(s)" + exceptions, exceptions.isEmpty()); }