public void allocate(LogicalSlot slot) {
        if (!GlobalVariable.isQueryQueueDriverHighWaterEffective()) {
            return;
        }

        boolean hasAllocated = !allocatedSlotIds.add(slot.getSlotId());
        if (hasAllocated) {
            return;
        }

        if (!slot.isAdaptiveDop()) {
            numAllocatedDrivers.getAndAdd(slot.getNumDrivers());
            return;
        }

        // If there are too many running drivers, set DOP to 1.
        if (numAllocatedDrivers.get() + slot.getNumFragments() >= GlobalVariable.getQueryQueueDriverHighWater()) {
            slot.setPipelineDop(1);
            numAllocatedDrivers.getAndAdd(slot.getNumDrivers());
            return;
        }

        AllocationRequest allocationRequest = new AllocationRequest(slot);
        allocationRequests.add(allocationRequest);
        allocateInBatch();
        allocationRequest.waitFinish();
    }
@Test
    public void testConcurrency() throws InterruptedException {
        PipelineDriverAllocator allocator = new PipelineDriverAllocator();

        mockCPUCores(PipelineDriverAllocator.NUM_BATCH_SLOTS * 1000);
        GlobalVariable.setQueryQueueDriverHighWater(0);
        GlobalVariable.setQueryQueueDriverLowWater(0);

        final int numSlots = PipelineDriverAllocator.NUM_BATCH_SLOTS * 2;

        List<Thread> threads = new ArrayList<>(numSlots);
        List<LogicalSlot> slots = new ArrayList<>(numSlots);
        for (int i = 0; i < numSlots; i++) {
            LogicalSlot slot = genSlot(2, 0);
            Thread thread = new Thread(() -> allocator.allocate(slot));
            threads.add(thread);
            slots.add(slot);
        }

        for (Thread thread : threads) {
            thread.start();
        }

        for (Thread thread : threads) {
            thread.join();
        }

        for (LogicalSlot slot : slots) {
            assertThat(slot.getPipelineDop()).isPositive();
        }
    }