<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nifi.controller.queue;

import org.apache.nifi.controller.repository.FlowFileRecord;
import org.apache.nifi.controller.repository.FlowFileSwapManager;
import org.apache.nifi.controller.repository.IncompleteSwapFileException;
import org.apache.nifi.controller.repository.SwapContents;
import org.apache.nifi.controller.repository.SwapSummary;
import org.apache.nifi.controller.repository.claim.ResourceClaim;
import org.apache.nifi.controller.swap.StandardSwapSummary;
import org.apache.nifi.events.EventReporter;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.FlowFileFilter;
import org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult;
import org.apache.nifi.reporting.Severity;
import org.apache.nifi.util.concurrency.TimedLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;


public class SwappablePriorityQueue {
    private static final Logger logger = LoggerFactory.getLogger(SwappablePriorityQueue.class);
    private static final int SWAP_RECORD_POLL_SIZE = 10_000;
    private static final int MAX_EXPIRED_RECORDS_PER_ITERATION = 10_000;

    private final int swapThreshold;
    private final FlowFileSwapManager swapManager;
    private final EventReporter eventReporter;
    private final FlowFileQueue flowFileQueue;
    private final DropFlowFileAction dropAction;
    private final List<FlowFilePrioritizer> priorities = new ArrayList<>();
    private final String swapPartitionName;

    private final List<String> swapLocations = new ArrayList<>();
    private final AtomicReference<FlowFileQueueSize> size = new AtomicReference<>(new FlowFileQueueSize(0, 0L, 0, 0L, 0, 0, 0L));
    private final TimedLock readLock;
    private final TimedLock writeLock;

    // We keep an "active queue" and a "swap queue" that both are able to hold records in heap. When
    // FlowFiles are added to this FlowFileQueue, we first check if we are in "swap mode" and if so
    // we add to the 'swap queue' instead of the 'active queue'. The code would be much simpler if we
    // eliminated the 'swap queue' and instead just used the active queue and swapped out the 10,000
    // lowest priority FlowFiles from that. However, doing that would cause problems with the ordering
    // of FlowFiles. If we swap out some FlowFiles, and then allow a new FlowFile to be written to the
    // active queue, then we would end up processing the newer FlowFile before the swapped FlowFile. By
    // keeping these separate, we are able to guarantee that FlowFiles are swapped in in the same order
    // that they are swapped out.
    // Guarded by lock.
    private PriorityQueue<FlowFileRecord> activeQueue;
    private ArrayList<FlowFileRecord> swapQueue;
    private boolean swapMode = false;

    public SwappablePriorityQueue(final FlowFileSwapManager swapManager, final int swapThreshold, final EventReporter eventReporter, final FlowFileQueue flowFileQueue,
        final DropFlowFileAction dropAction, final String swapPartitionName) {
        this.swapManager = swapManager;
        this.swapThreshold = swapThreshold;

        this.activeQueue = new PriorityQueue<>(20, new QueuePrioritizer(Collections.emptyList()));
        this.swapQueue = new ArrayList<>();
        this.eventReporter = eventReporter;
        this.flowFileQueue = flowFileQueue;
        this.dropAction = dropAction;
        this.swapPartitionName = swapPartitionName;

        final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
        readLock = new TimedLock(lock.readLock(), flowFileQueue.getIdentifier() + " Read Lock", 100);
        writeLock = new TimedLock(lock.writeLock(), flowFileQueue.getIdentifier() + " Write Lock", 100);
    }

    private String getQueueIdentifier() {
        return flowFileQueue.getIdentifier();
    }

    public synchronized List<FlowFilePrioritizer> getPriorities() {
        readLock.lock();
        try {
            return Collections.unmodifiableList(priorities);
        } finally {
            readLock.unlock("getPriorities");
        }<commit_msg>NIFI-5736: Removed 'synchronized' keyword from SwappablePriorityQueue#getProperties - the method should not be synchronized, as it guards access to member variables via a ReadLock. I am guessing that the synchronized keyword is the result of refactoring from 'synchronized' to a read lock and inadvertently leaving the 'synchronized' keyword in.<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.nifi.controller.queue;

import org.apache.nifi.controller.repository.FlowFileRecord;
import org.apache.nifi.controller.repository.FlowFileSwapManager;
import org.apache.nifi.controller.repository.IncompleteSwapFileException;
import org.apache.nifi.controller.repository.SwapContents;
import org.apache.nifi.controller.repository.SwapSummary;
import org.apache.nifi.controller.repository.claim.ResourceClaim;
import org.apache.nifi.controller.swap.StandardSwapSummary;
import org.apache.nifi.events.EventReporter;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.FlowFileFilter;
import org.apache.nifi.processor.FlowFileFilter.FlowFileFilterResult;
import org.apache.nifi.reporting.Severity;
import org.apache.nifi.util.concurrency.TimedLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;


public class SwappablePriorityQueue {
    private static final Logger logger = LoggerFactory.getLogger(SwappablePriorityQueue.class);
    private static final int SWAP_RECORD_POLL_SIZE = 10_000;
    private static final int MAX_EXPIRED_RECORDS_PER_ITERATION = 10_000;

    private final int swapThreshold;
    private final FlowFileSwapManager swapManager;
    private final EventReporter eventReporter;
    private final FlowFileQueue flowFileQueue;
    private final DropFlowFileAction dropAction;
    private final List<FlowFilePrioritizer> priorities = new ArrayList<>();
    private final String swapPartitionName;

    private final List<String> swapLocations = new ArrayList<>();
    private final AtomicReference<FlowFileQueueSize> size = new AtomicReference<>(new FlowFileQueueSize(0, 0L, 0, 0L, 0, 0, 0L));
    private final TimedLock readLock;
    private final TimedLock writeLock;

    // We keep an "active queue" and a "swap queue" that both are able to hold records in heap. When
    // FlowFiles are added to this FlowFileQueue, we first check if we are in "swap mode" and if so
    // we add to the 'swap queue' instead of the 'active queue'. The code would be much simpler if we
    // eliminated the 'swap queue' and instead just used the active queue and swapped out the 10,000
    // lowest priority FlowFiles from that. However, doing that would cause problems with the ordering
    // of FlowFiles. If we swap out some FlowFiles, and then allow a new FlowFile to be written to the
    // active queue, then we would end up processing the newer FlowFile before the swapped FlowFile. By
    // keeping these separate, we are able to guarantee that FlowFiles are swapped in in the same order
    // that they are swapped out.
    // Guarded by lock.
    private PriorityQueue<FlowFileRecord> activeQueue;
    private ArrayList<FlowFileRecord> swapQueue;
    private boolean swapMode = false;

    public SwappablePriorityQueue(final FlowFileSwapManager swapManager, final int swapThreshold, final EventReporter eventReporter, final FlowFileQueue flowFileQueue,
        final DropFlowFileAction dropAction, final String swapPartitionName) {
        this.swapManager = swapManager;
        this.swapThreshold = swapThreshold;

        this.activeQueue = new PriorityQueue<>(20, new QueuePrioritizer(Collections.emptyList()));
        this.swapQueue = new ArrayList<>();
        this.eventReporter = eventReporter;
        this.flowFileQueue = flowFileQueue;
        this.dropAction = dropAction;
        this.swapPartitionName = swapPartitionName;

        final ReentrantReadWriteLock lock = new ReentrantReadWriteLock(true);
        readLock = new TimedLock(lock.readLock(), flowFileQueue.getIdentifier() + " Read Lock", 100);
        writeLock = new TimedLock(lock.writeLock(), flowFileQueue.getIdentifier() + " Write Lock", 100);
    }

    private String getQueueIdentifier() {
        return flowFileQueue.getIdentifier();
    }

    public List<FlowFilePrioritizer> getPriorities() {
        readLock.lock();
        try {
            return Collections.unmodifiableList(priorities);
        } finally {
            readLock.unlock("getPriorities");
        }
{'fraction_non_alphanumeric': 0.07398439601829432, 'fraction_numerical': 0.006815532239261053, 'mean_word_length': 4.272813238770685, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 25, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5534171', 'n_tokens_mistral': 3009, 'n_tokens_neox': 2906, 'n_words': 1009}