@Test public void testTakeWithLimits() throws OutOfCapacityException, BrokenBarrierException, InterruptedException { Map<String, Object> properties = new HashMap<>(); properties.put(ResourceManager.QUERY_WORKER_CONFIG_KEY, 40); properties.put(ResourceManager.QUERY_RUNNER_CONFIG_KEY, 10); properties.put(ResourceLimitPolicy.TABLE_THREADS_SOFT_LIMIT, 20); properties.put(ResourceLimitPolicy.TABLE_THREADS_HARD_LIMIT, 80); PinotConfiguration configuration = new PinotConfiguration(properties); PolicyBasedResourceManager rm = new PolicyBasedResourceManager(configuration); MultiLevelPriorityQueue queue = createQueue(configuration, rm); queue.put(createQueryRequest(groupOne, metrics)); queue.put(createQueryRequest(groupOne, metrics)); queue.put(createQueryRequest(groupTwo, metrics)); TestSchedulerGroup testGroupOne = groupFactory.groupMap.get(groupOne); TestSchedulerGroup testGroupTwo = groupFactory.groupMap.get(groupTwo); testGroupOne.addReservedThreads(rm.getTableThreadsSoftLimit() + 1); QueueReader reader = new QueueReader(queue); reader.startAndWaitForRead(); assertEquals(reader.readQueries.size(), 1); assertEquals(reader.readQueries.poll().getSchedulerGroup().name(), groupTwo); queue.put(createQueryRequest(groupTwo, metrics)); reader = new QueueReader(queue); reader.startAndWaitForRead(); assertEquals(reader.readQueries.size(), 1); assertEquals(reader.readQueries.poll().getSchedulerGroup().name(), groupTwo); queue.put(createQueryRequest(groupTwo, metrics)); testGroupTwo.addReservedThreads(testGroupOne.totalReservedThreads() + 1); reader = new QueueReader(queue); reader.startAndWaitForRead(); assertEquals(reader.readQueries.size(), 1); assertEquals(reader.readQueries.poll().getSchedulerGroup().name(), groupOne); testGroupOne.addReservedThreads(rm.getTableThreadsHardLimit()); reader = new QueueReader(queue); reader.startAndWaitForRead(); assertEquals(reader.readQueries.size(), 1); assertEquals(reader.readQueries.poll().getSchedulerGroup().name(), groupTwo); queue.put(createQueryRequest(groupTwo, metrics)); queue.put(createQueryRequest(groupTwo, metrics)); queue.put(createQueryRequest(groupOne, metrics)); testGroupTwo.addReservedThreads(rm.getTableThreadsHardLimit()); reader = new QueueReader(queue); reader.startAndWaitForQueueWakeup(); assertEquals(reader.readQueries.size(), 0); sleepForQueueWakeup(queue); assertEquals(reader.readQueries.size(), 0); testGroupTwo.releasedReservedThreads(testGroupTwo.totalReservedThreads()); sleepForQueueWakeup(queue); assertEquals(reader.readQueries.size(), 1); }
@Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } MultiLevelPriorityQueue(@Nonnull PinotConfiguration config, @Nonnull ResourceManager resourceManager,
      @Nonnull SchedulerGroupFactory groupFactory, @Nonnull SchedulerGroupMapper groupMapper); }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } MultiLevelPriorityQueue(@Nonnull PinotConfiguration config, @Nonnull ResourceManager resourceManager,
      @Nonnull SchedulerGroupFactory groupFactory, @Nonnull SchedulerGroupMapper groupMapper); @Override void put(@Nonnull SchedulerQueryContext query); @Nullable @Override SchedulerQueryContext take(); @Nonnull @Override List<SchedulerQueryContext> drain(); }
MultiLevelPriorityQueue implements SchedulerPriorityQueue { @Override public void put(@Nonnull SchedulerQueryContext query) throws OutOfCapacityException { Preconditions.checkNotNull(query); queueLock.lock(); String groupName = groupSelector.getSchedulerGroupName(query); try { SchedulerGroup groupContext = getOrCreateGroupContext(groupName); checkGroupHasCapacity(groupContext); query.setSchedulerGroupContext(groupContext); groupContext.addLast(query); queryReaderCondition.signal(); } finally { queueLock.unlock(); } } MultiLevelPriorityQueue(@Nonnull PinotConfiguration config, @Nonnull ResourceManager resourceManager,
      @Nonnull SchedulerGroupFactory groupFactory, @Nonnull SchedulerGroupMapper groupMapper); @Override void put(@Nonnull SchedulerQueryContext query); @Nullable @Override SchedulerQueryContext take(); @Nonnull @Override List<SchedulerQueryContext> drain(); static final String QUERY_DEADLINE_SECONDS_KEY; static final String MAX_PENDING_PER_GROUP_KEY; static final String QUEUE_WAKEUP_MICROS; }