IngestTasksScheduler.java

Go to the documentation of this file.

/*
 * Autopsy Forensic Browser
 *
 * Copyright 2012-2021 Basis Technology Corp.
 * Contact: carrier <at> sleuthkit <dot> org
 *
 * Licensed 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.sleuthkit.autopsy.ingest;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Deque;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Queue;
import java.util.TreeSet;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.logging.Level;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import org.sleuthkit.autopsy.casemodule.Case;
import org.sleuthkit.autopsy.coreutils.Logger;
import org.sleuthkit.datamodel.AbstractFile;
import org.sleuthkit.datamodel.Blackboard;
import org.sleuthkit.datamodel.Content;
import org.sleuthkit.datamodel.DataArtifact;
import org.sleuthkit.datamodel.DataSource;
import org.sleuthkit.datamodel.FileSystem;
import org.sleuthkit.datamodel.TskCoreException;
import org.sleuthkit.datamodel.TskData;

@ThreadSafe
final class IngestTasksScheduler {

    private static final int FAT_NTFS_FLAGS = TskData.TSK_FS_TYPE_ENUM.TSK_FS_TYPE_FAT12.getValue() | TskData.TSK_FS_TYPE_ENUM.TSK_FS_TYPE_FAT16.getValue() | TskData.TSK_FS_TYPE_ENUM.TSK_FS_TYPE_FAT32.getValue() | TskData.TSK_FS_TYPE_ENUM.TSK_FS_TYPE_NTFS.getValue();
    private static final Logger logger = Logger.getLogger(IngestTasksScheduler.class.getName());
    @GuardedBy("IngestTasksScheduler.this")
    private static IngestTasksScheduler instance;
    private final IngestTaskTrackingQueue dataSourceIngestTasksQueue;
    @GuardedBy("this")
    private final TreeSet<FileIngestTask> topLevelFileIngestTasksQueue;
    @GuardedBy("this")
    private final Deque<FileIngestTask> batchedFileIngestTasksQueue;
    @GuardedBy("this")
    private final Queue<FileIngestTask> streamedFileIngestTasksQueue;
    private final IngestTaskTrackingQueue fileIngestTasksQueue;
    private final IngestTaskTrackingQueue artifactIngestTasksQueue;

    synchronized static IngestTasksScheduler getInstance() {
        if (IngestTasksScheduler.instance == null) {
            IngestTasksScheduler.instance = new IngestTasksScheduler();
        }
        return IngestTasksScheduler.instance;
    }

    private IngestTasksScheduler() {
        dataSourceIngestTasksQueue = new IngestTaskTrackingQueue();
        topLevelFileIngestTasksQueue = new TreeSet<>(new RootDirectoryTaskComparator());
        batchedFileIngestTasksQueue = new LinkedList<>();
        fileIngestTasksQueue = new IngestTaskTrackingQueue();
        streamedFileIngestTasksQueue = new LinkedList<>();
        artifactIngestTasksQueue = new IngestTaskTrackingQueue();
    }

    BlockingIngestTaskQueue getDataSourceIngestTaskQueue() {
        return dataSourceIngestTasksQueue;
    }

    BlockingIngestTaskQueue getFileIngestTaskQueue() {
        return fileIngestTasksQueue;
    }

    BlockingIngestTaskQueue getResultIngestTaskQueue() {
        return artifactIngestTasksQueue;
    }

    synchronized void scheduleIngestTasks(IngestJobPipeline ingestPipeline) {
        if (!ingestPipeline.isCancelled()) {
            if (ingestPipeline.hasDataSourceIngestModules()) {
                scheduleDataSourceIngestTask(ingestPipeline);
            }
            if (ingestPipeline.hasFileIngestModules()) {
                scheduleFileIngestTasks(ingestPipeline, Collections.emptyList());
            }
            if (ingestPipeline.hasDataArtifactIngestModules()) {
                scheduleDataArtifactIngestTasks(ingestPipeline);
            }
        }
    }

    synchronized void scheduleDataSourceIngestTask(IngestJobPipeline ingestPipeline) {
        if (!ingestPipeline.isCancelled()) {
            DataSourceIngestTask task = new DataSourceIngestTask(ingestPipeline);
            try {
                dataSourceIngestTasksQueue.putLast(task);
            } catch (InterruptedException ex) {
                IngestTasksScheduler.logger.log(Level.INFO, String.format("Ingest tasks scheduler interrupted while blocked adding a task to the data source level ingest task queue (pipelineId={%d)", ingestPipeline.getId()), ex);
                Thread.currentThread().interrupt();
            }
        }
    }

    synchronized void scheduleFileIngestTasks(IngestJobPipeline ingestPipeline, Collection<AbstractFile> files) {
        if (!ingestPipeline.isCancelled()) {
            Collection<AbstractFile> candidateFiles;
            if (files.isEmpty()) {
                candidateFiles = getTopLevelFiles(ingestPipeline.getDataSource());
            } else {
                candidateFiles = files;
            }
            for (AbstractFile file : candidateFiles) {
                FileIngestTask task = new FileIngestTask(ingestPipeline, file);
                if (IngestTasksScheduler.shouldEnqueueFileTask(task)) {
                    topLevelFileIngestTasksQueue.add(task);
                }
            }
            refillFileIngestTasksQueue();
        }
    }

    synchronized void scheduleStreamedFileIngestTasks(IngestJobPipeline ingestPipeline, List<Long> fileIds) {
        if (!ingestPipeline.isCancelled()) {
            for (long id : fileIds) {
                /*
                 * Create the file ingest task. Note that we do not do the
                 * shouldEnqueueFileTask() check here in order to delay querying
                 * the case database to construct the AbstractFile object. The
                 * file filter will be applied before the file task makes it to
                 * the task queue consumed by the file ingest threads.
                 */
                FileIngestTask task = new FileIngestTask(ingestPipeline, id);
                streamedFileIngestTasksQueue.add(task);
            }
            refillFileIngestTasksQueue();
        }
    }

    synchronized void fastTrackFileIngestTasks(IngestJobPipeline ingestPipeline, Collection<AbstractFile> files) {
        if (!ingestPipeline.isCancelled()) {
            /*
             * Put the files directly into the queue for the file ingest
             * threads, if they pass the file filter for the job. The files are
             * added to the queue for the ingest threads BEFORE the other queued
             * tasks because the use case for this method is scheduling new
             * carved or derived files from a high priority task that is already
             * in progress.
             */
            for (AbstractFile file : files) {
                FileIngestTask fileTask = new FileIngestTask(ingestPipeline, file);
                if (shouldEnqueueFileTask(fileTask)) {
                    try {
                        fileIngestTasksQueue.putFirst(fileTask);
                    } catch (InterruptedException ex) {
                        DataSource dataSource = ingestPipeline.getDataSource();
                        logger.log(Level.WARNING, String.format("Interrupted while enqueuing file tasks for %s (data source object ID = %d)", dataSource.getName(), dataSource.getId()), ex); //NON-NLS
                        Thread.currentThread().interrupt();
                        return;
                    }
                }
            }
        }
    }

    synchronized void scheduleDataArtifactIngestTasks(IngestJobPipeline ingestPipeline) {
        if (!ingestPipeline.isCancelled()) {
            Blackboard blackboard = Case.getCurrentCase().getSleuthkitCase().getBlackboard();
            try {
                List<DataArtifact> artifacts = blackboard.getDataArtifacts(ingestPipeline.getDataSource().getId(), null);
                scheduleDataArtifactIngestTasks(ingestPipeline, artifacts);
            } catch (TskCoreException ex) {
                DataSource dataSource = ingestPipeline.getDataSource();
                logger.log(Level.SEVERE, String.format("Failed to retrieve data artifacts for %s (data source object ID = %d)", dataSource.getName(), dataSource.getId()), ex); //NON-NLS
            }
        }
    }

    synchronized void scheduleDataArtifactIngestTasks(IngestJobPipeline ingestPipeline, List<DataArtifact> artifacts) {
        if (!ingestPipeline.isCancelled()) {
            for (DataArtifact artifact : artifacts) {
                DataArtifactIngestTask task = new DataArtifactIngestTask(ingestPipeline, artifact);
                try {
                    this.artifactIngestTasksQueue.putLast(task);
                } catch (InterruptedException ex) {
                    DataSource dataSource = ingestPipeline.getDataSource();
                    logger.log(Level.WARNING, String.format("Interrupted while enqueuing data artifact tasks for %s (data source object ID = %d)", dataSource.getName(), dataSource.getId()), ex); //NON-NLS
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
    }

    synchronized void notifyTaskCompleted(DataSourceIngestTask task) {
        dataSourceIngestTasksQueue.taskCompleted(task);
    }

    synchronized void notifyTaskCompleted(FileIngestTask task) {
        fileIngestTasksQueue.taskCompleted(task);
        refillFileIngestTasksQueue();
    }

    synchronized void notifyTaskCompleted(DataArtifactIngestTask task) {
        artifactIngestTasksQueue.taskCompleted(task);
    }

    synchronized boolean currentTasksAreCompleted(IngestJobPipeline ingestPipeline) {
        long pipelineId = ingestPipeline.getId();
        return !(dataSourceIngestTasksQueue.hasTasksForJob(pipelineId)
                || hasTasksForJob(topLevelFileIngestTasksQueue, pipelineId)
                || hasTasksForJob(batchedFileIngestTasksQueue, pipelineId)
                || hasTasksForJob(streamedFileIngestTasksQueue, pipelineId)
                || fileIngestTasksQueue.hasTasksForJob(pipelineId)
                || artifactIngestTasksQueue.hasTasksForJob(pipelineId));
    }

    synchronized void cancelPendingFileTasksForIngestJob(IngestJobPipeline ingestJobPipeline) {
        long jobId = ingestJobPipeline.getId();
        removeTasksForJob(topLevelFileIngestTasksQueue, jobId);
        removeTasksForJob(batchedFileIngestTasksQueue, jobId);
        removeTasksForJob(streamedFileIngestTasksQueue, jobId);
    }

    private static List<AbstractFile> getTopLevelFiles(Content dataSource) {
        List<AbstractFile> topLevelFiles = new ArrayList<>();
        Collection<AbstractFile> rootObjects = dataSource.accept(new GetRootDirectoryVisitor());
        if (rootObjects.isEmpty() && dataSource instanceof AbstractFile) {
            // The data source is itself a file to be processed.
            topLevelFiles.add((AbstractFile) dataSource);
        } else {
            for (AbstractFile root : rootObjects) {
                List<Content> children;
                try {
                    children = root.getChildren();
                    if (children.isEmpty()) {
                        // Add the root object itself, it could be an unallocated
                        // space file, or a child of a volume or an image.
                        topLevelFiles.add(root);
                    } else {
                        // The root object is a file system root directory, get
                        // the files within it.
                        for (Content child : children) {
                            if (child instanceof AbstractFile) {
                                topLevelFiles.add((AbstractFile) child);
                            }
                        }
                    }
                } catch (TskCoreException ex) {
                    logger.log(Level.SEVERE, "Could not get children of root to enqueue: " + root.getId() + ": " + root.getName(), ex); //NON-NLS
                }
            }
        }
        return topLevelFiles;
    }

    synchronized private void refillFileIngestTasksQueue() {
        try {
            takeFromStreamingFileTasksQueue();
            takeFromBatchTasksQueues();
        } catch (InterruptedException ex) {
            IngestTasksScheduler.logger.log(Level.INFO, "Ingest tasks scheduler interrupted while blocked adding a task to the file level ingest task queue", ex);
            Thread.currentThread().interrupt();
        }
    }

    synchronized private void takeFromStreamingFileTasksQueue() throws InterruptedException {
        while (fileIngestTasksQueue.isEmpty()) {
            int taskCount = 0;
            while (taskCount < IngestManager.getInstance().getNumberOfFileIngestThreads()) {
                final FileIngestTask streamingTask = streamedFileIngestTasksQueue.poll();
                if (streamingTask == null) {
                    return; // No streaming tasks are queued right now
                }
                if (shouldEnqueueFileTask(streamingTask)) {
                    fileIngestTasksQueue.putLast(streamingTask);
                    taskCount++;
                }
            }
        }
    }

    synchronized private void takeFromBatchTasksQueues() throws InterruptedException {

        while (fileIngestTasksQueue.isEmpty()) {
            /*
             * If the batched file task queue is empty, move the highest
             * priority top level file task into it.
             */
            if (batchedFileIngestTasksQueue.isEmpty()) {
                final FileIngestTask topLevelTask = topLevelFileIngestTasksQueue.pollFirst();
                if (topLevelTask != null) {
                    batchedFileIngestTasksQueue.addLast(topLevelTask);
                }
            }

            /*
             * Try to move the next task from the batched file tasks queue into
             * the queue for the file ingest threads.
             */
            final FileIngestTask nextTask = batchedFileIngestTasksQueue.pollFirst();
            if (nextTask == null) {
                return;
            }
            if (shouldEnqueueFileTask(nextTask)) {
                fileIngestTasksQueue.putLast(nextTask);
            }

            /*
             * If the task that was just queued for the file ingest threads has
             * children, queue tasks for the children as well.
             */
            AbstractFile file = null;
            try {
                file = nextTask.getFile();
                for (Content child : file.getChildren()) {
                    if (child instanceof AbstractFile) {
                        AbstractFile childFile = (AbstractFile) child;
                        FileIngestTask childTask = new FileIngestTask(nextTask.getIngestJobPipeline(), childFile);
                        if (childFile.hasChildren()) {
                            batchedFileIngestTasksQueue.add(childTask);
                        } else if (shouldEnqueueFileTask(childTask)) {
                            fileIngestTasksQueue.putLast(childTask);
                        }
                    }
                }
            } catch (TskCoreException ex) {
                if (file != null) {
                    logger.log(Level.SEVERE, String.format("Error getting the children of %s (object ID = %d)", file.getName(), file.getId()), ex); //NON-NLS
                } else {
                    logger.log(Level.SEVERE, "Error loading file with object ID = {0}", nextTask.getFileId()); //NON-NLS
                }
            }
        }
    }

    private static boolean shouldEnqueueFileTask(final FileIngestTask task) {
        AbstractFile file;
        try {
            file = task.getFile();
        } catch (TskCoreException ex) {
            logger.log(Level.SEVERE, "Error loading file with ID {0}", task.getFileId());
            return false;
        }

        // Skip the task if the file is actually the pseudo-file for the parent
        // or current directory.
        String fileName = file.getName();

        if (fileName.equals(".") || fileName.equals("..")) {
            return false;
        }

        /*
         * Ensures that all directories, files which are members of the ingest
         * file filter, and unallocated blocks (when processUnallocated is
         * enabled) all continue to be processed. AbstractFiles which do not
         * meet one of these criteria will be skipped.
         *
         * An additional check to see if unallocated space should be processed
         * is part of the FilesSet.fileIsMemberOf() method.
         *
         * This code may need to be updated when
         * TSK_DB_FILES_TYPE_ENUM.UNUSED_BLOCKS comes into use by Autopsy.
         */
        if (!file.isDir() && !shouldBeCarved(task) && !fileAcceptedByFilter(task)) {
            return false;
        }

        // Skip the task if the file is one of a select group of special, large
        // NTFS or FAT file system files.
        if (file instanceof org.sleuthkit.datamodel.File) {
            final org.sleuthkit.datamodel.File f = (org.sleuthkit.datamodel.File) file;

            // Get the type of the file system, if any, that owns the file.
            TskData.TSK_FS_TYPE_ENUM fsType = TskData.TSK_FS_TYPE_ENUM.TSK_FS_TYPE_UNSUPP;
            try {
                FileSystem fs = f.getFileSystem();
                if (fs != null) {
                    fsType = fs.getFsType();
                }
            } catch (TskCoreException ex) {
                logger.log(Level.SEVERE, "Error querying file system for " + f, ex); //NON-NLS
            }

            // If the file system is not NTFS or FAT, don't skip the file.
            if ((fsType.getValue() & FAT_NTFS_FLAGS) == 0) {
                return true;
            }

            // Find out whether the file is in a root directory. 
            boolean isInRootDir = false;
            try {
                AbstractFile parent = f.getParentDirectory();
                if (parent == null) {
                    isInRootDir = true;
                } else {
                    isInRootDir = parent.isRoot();
                }
            } catch (TskCoreException ex) {
                logger.log(Level.WARNING, "Error querying parent directory for" + f.getName(), ex); //NON-NLS
            }

            // If the file is in the root directory of an NTFS or FAT file 
            // system, check its meta-address and check its name for the '$'
            // character and a ':' character (not a default attribute).
            if (isInRootDir && f.getMetaAddr() < 32) {
                String name = f.getName();
                if (name.length() > 0 && name.charAt(0) == '$' && name.contains(":")) {
                    return false;
                }
            }
        }

        return true;
    }

    private static boolean shouldBeCarved(final FileIngestTask task) {
        try {
            AbstractFile file = task.getFile();
            return task.getIngestJobPipeline().shouldProcessUnallocatedSpace() && file.getType().equals(TskData.TSK_DB_FILES_TYPE_ENUM.UNALLOC_BLOCKS);
        } catch (TskCoreException ex) {
            return false;
        }
    }

    private static boolean fileAcceptedByFilter(final FileIngestTask task) {
        try {
            AbstractFile file = task.getFile();
            return !(task.getIngestJobPipeline().getFileIngestFilter().fileIsMemberOf(file) == null);
        } catch (TskCoreException ex) {
            return false;
        }
    }

    synchronized private static boolean hasTasksForJob(Collection<? extends IngestTask> tasks, long pipelineId) {
        for (IngestTask task : tasks) {
            if (task.getIngestJobPipeline().getId() == pipelineId) {
                return true;
            }
        }
        return false;
    }

    private static void removeTasksForJob(Collection<? extends IngestTask> tasks, long pipelineId) {
        Iterator<? extends IngestTask> iterator = tasks.iterator();
        while (iterator.hasNext()) {
            IngestTask task = iterator.next();
            if (task.getIngestJobPipeline().getId() == pipelineId) {
                iterator.remove();
            }
        }
    }

    private static int countTasksForJob(Collection<? extends IngestTask> tasks, long pipelineId) {
        int count = 0;
        for (IngestTask task : tasks) {
            if (task.getIngestJobPipeline().getId() == pipelineId) {
                count++;
            }
        }
        return count;
    }

    synchronized IngestJobTasksSnapshot getTasksSnapshotForJob(long jobId) {
        return new IngestJobTasksSnapshot(jobId, dataSourceIngestTasksQueue.countQueuedTasksForJob(jobId),
                countTasksForJob(topLevelFileIngestTasksQueue, jobId),
                countTasksForJob(batchedFileIngestTasksQueue, jobId),
                fileIngestTasksQueue.countQueuedTasksForJob(jobId),
                dataSourceIngestTasksQueue.countRunningTasksForJob(jobId) + fileIngestTasksQueue.countRunningTasksForJob(jobId) + artifactIngestTasksQueue.countRunningTasksForJob(jobId),
                countTasksForJob(streamedFileIngestTasksQueue, jobId),
                artifactIngestTasksQueue.countQueuedTasksForJob(jobId));
    }

    private static class RootDirectoryTaskComparator implements Comparator<FileIngestTask> {

        @Override
        public int compare(FileIngestTask q1, FileIngestTask q2) {
            // In practice the case where one or both calls to getFile() fails
            // should never occur since such tasks would not be added to the queue.
            AbstractFile file1 = null;
            AbstractFile file2 = null;
            try {
                file1 = q1.getFile();
            } catch (TskCoreException ex) {
                // Do nothing - the exception has been logged elsewhere
            }

            try {
                file2 = q2.getFile();
            } catch (TskCoreException ex) {
                // Do nothing - the exception has been logged elsewhere
            }

            if (file1 == null) {
                if (file2 == null) {
                    return (int) (q2.getFileId() - q1.getFileId());
                } else {
                    return 1;
                }
            } else if (file2 == null) {
                return -1;
            }

            AbstractFilePriority.Priority p1 = AbstractFilePriority.getPriority(file1);
            AbstractFilePriority.Priority p2 = AbstractFilePriority.getPriority(file2);
            if (p1 == p2) {
                return (int) (file2.getId() - file1.getId());
            } else {
                return p2.ordinal() - p1.ordinal();
            }
        }

        private static class AbstractFilePriority {

            private AbstractFilePriority() {
            }

            enum Priority {

                LAST, LOW, MEDIUM, HIGH
            }

            static final List<Pattern> LAST_PRI_PATHS = new ArrayList<>();

            static final List<Pattern> LOW_PRI_PATHS = new ArrayList<>();

            static final List<Pattern> MEDIUM_PRI_PATHS = new ArrayList<>();

            static final List<Pattern> HIGH_PRI_PATHS = new ArrayList<>();

            /*
             * prioritize root directory folders based on the assumption that we
             * are looking for user content. Other types of investigations may
             * want different priorities.
             */
            static /*
             * prioritize root directory folders based on the assumption that we
             * are looking for user content. Other types of investigations may
             * want different priorities.
             */ {
                // these files have no structure, so they go last
                //unalloc files are handled as virtual files in getPriority()
                //LAST_PRI_PATHS.schedule(Pattern.compile("^\\$Unalloc", Pattern.CASE_INSENSITIVE));
                //LAST_PRI_PATHS.schedule(Pattern.compile("^\\Unalloc", Pattern.CASE_INSENSITIVE));
                LAST_PRI_PATHS.add(Pattern.compile("^pagefile", Pattern.CASE_INSENSITIVE));
                LAST_PRI_PATHS.add(Pattern.compile("^hiberfil", Pattern.CASE_INSENSITIVE));
                // orphan files are often corrupt and windows does not typically have
                // user content, so put them towards the bottom
                LOW_PRI_PATHS.add(Pattern.compile("^\\$OrphanFiles", Pattern.CASE_INSENSITIVE));
                LOW_PRI_PATHS.add(Pattern.compile("^Windows", Pattern.CASE_INSENSITIVE));
                // all other files go into the medium category too
                MEDIUM_PRI_PATHS.add(Pattern.compile("^Program Files", Pattern.CASE_INSENSITIVE));
                // user content is top priority
                HIGH_PRI_PATHS.add(Pattern.compile("^Users", Pattern.CASE_INSENSITIVE));
                HIGH_PRI_PATHS.add(Pattern.compile("^Documents and Settings", Pattern.CASE_INSENSITIVE));
                HIGH_PRI_PATHS.add(Pattern.compile("^home", Pattern.CASE_INSENSITIVE));
                HIGH_PRI_PATHS.add(Pattern.compile("^ProgramData", Pattern.CASE_INSENSITIVE));
            }

            static AbstractFilePriority.Priority getPriority(final AbstractFile abstractFile) {
                if (!abstractFile.getType().equals(TskData.TSK_DB_FILES_TYPE_ENUM.FS)) {
                    //quickly filter out unstructured content
                    //non-fs virtual files and dirs, such as representing unalloc space
                    return AbstractFilePriority.Priority.LAST;
                }
                //determine the fs files priority by name
                final String path = abstractFile.getName();
                if (path == null) {
                    return AbstractFilePriority.Priority.MEDIUM;
                }
                for (Pattern p : HIGH_PRI_PATHS) {
                    Matcher m = p.matcher(path);
                    if (m.find()) {
                        return AbstractFilePriority.Priority.HIGH;
                    }
                }
                for (Pattern p : MEDIUM_PRI_PATHS) {
                    Matcher m = p.matcher(path);
                    if (m.find()) {
                        return AbstractFilePriority.Priority.MEDIUM;
                    }
                }
                for (Pattern p : LOW_PRI_PATHS) {
                    Matcher m = p.matcher(path);
                    if (m.find()) {
                        return AbstractFilePriority.Priority.LOW;
                    }
                }
                for (Pattern p : LAST_PRI_PATHS) {
                    Matcher m = p.matcher(path);
                    if (m.find()) {
                        return AbstractFilePriority.Priority.LAST;
                    }
                }
                //default is medium
                return AbstractFilePriority.Priority.MEDIUM;
            }
        }
    }

    @ThreadSafe
    private class IngestTaskTrackingQueue implements BlockingIngestTaskQueue {

        private final BlockingDeque<IngestTask> taskQueue = new LinkedBlockingDeque<>();
        @GuardedBy("this")
        private final List<IngestTask> queuedTasks = new LinkedList<>();
        @GuardedBy("this")
        private final List<IngestTask> tasksInProgress = new LinkedList<>();

        void putFirst(IngestTask task) throws InterruptedException {
            synchronized (this) {
                this.queuedTasks.add(task);
            }
            try {
                this.taskQueue.putFirst(task);
            } catch (InterruptedException ex) {
                synchronized (this) {
                    this.queuedTasks.remove(task);
                }
                throw ex;
            }
        }

        void putLast(IngestTask task) throws InterruptedException {
            synchronized (this) {
                this.queuedTasks.add(task);
            }
            try {
                this.taskQueue.putLast(task);
            } catch (InterruptedException ex) {
                synchronized (this) {
                    this.queuedTasks.remove(task);
                }
                throw ex;
            }
        }

        @Override
        public IngestTask getNextTask() throws InterruptedException {
            IngestTask task = taskQueue.takeFirst();
            synchronized (this) {
                this.queuedTasks.remove(task);
                this.tasksInProgress.add(task);
            }
            return task;
        }

        boolean isEmpty() {
            synchronized (this) {
                return this.queuedTasks.isEmpty();
            }
        }

        void taskCompleted(IngestTask task) {
            synchronized (this) {
                this.tasksInProgress.remove(task);
            }
        }

        boolean hasTasksForJob(long pipelineId) {
            synchronized (this) {
                return IngestTasksScheduler.hasTasksForJob(queuedTasks, pipelineId) || IngestTasksScheduler.hasTasksForJob(tasksInProgress, pipelineId);
            }
        }

        int countQueuedTasksForJob(long pipelineId) {
            synchronized (this) {
                return IngestTasksScheduler.countTasksForJob(queuedTasks, pipelineId);
            }
        }

        int countRunningTasksForJob(long pipelineId) {
            synchronized (this) {
                return IngestTasksScheduler.countTasksForJob(tasksInProgress, pipelineId);
            }
        }

    }

    static final class IngestJobTasksSnapshot implements Serializable {

        private static final long serialVersionUID = 1L;
        private final long jobId;
        private final long dsQueueSize;
        private final long rootQueueSize;
        private final long dirQueueSize;
        private final long fileQueueSize;
        private final long runningListSize;
        private final long streamingQueueSize;
        private final long artifactsQueueSize;

        IngestJobTasksSnapshot(long jobId, long dsQueueSize, long rootQueueSize, long dirQueueSize, long fileQueueSize,
                long runningListSize, long streamingQueueSize, long artifactsQueueSize) {
            this.jobId = jobId;
            this.dsQueueSize = dsQueueSize;
            this.rootQueueSize = rootQueueSize;
            this.dirQueueSize = dirQueueSize;
            this.fileQueueSize = fileQueueSize;
            this.runningListSize = runningListSize;
            this.streamingQueueSize = streamingQueueSize;
            this.artifactsQueueSize = artifactsQueueSize;
        }

        long getJobId() {
            return jobId;
        }

        long getRootQueueSize() {
            return rootQueueSize;
        }

        long getDirectoryTasksQueueSize() {
            return dirQueueSize;
        }

        long getFileQueueSize() {
            return fileQueueSize;
        }

        long getStreamingQueueSize() {
            return streamingQueueSize;
        }

        long getDsQueueSize() {
            return dsQueueSize;
        }

        long getRunningListSize() {
            return runningListSize;
        }
        
        long getArtifactsQueueSize() {
            return artifactsQueueSize;
        }

    }

}