//******************************************************************************************************//  FileTracker.cs - Gbtc////  Copyright © 2014, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), 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.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  04/29/2014 - Stephen C. Wills//       Generated original version of source code.////****************************************************************************************************** using System;using System.Collections.Generic;using System.Diagnostics;using System.IO;using System.Linq;using System.Text;using System.Threading;using System.Timers;using GSF.Collections;using GSF.Threading;using CancellationToken = GSF.Threading.CancellationToken;using Timer = System.Timers.Timer; namespace GSF.IO{    /// <summary>    /// Defines strategies for enumerating    /// files in the watch directories.    /// </summary>    public enum FileEnumerationStrategy    {        /// <summary>        /// Enumerate all files sequentially.        /// </summary>        Sequential,         /// <summary>        /// Enumerates the watch directories in parallel,        /// but subdirectories are processed sequentially.        /// </summary>        ParallelWatchDirectories,         /// <summary>        /// Enumerates every directory, including subdirectories, in parallel.        /// </summary>        ParallelSubdirectories,         /// <summary>        /// Does not enumerate directories, relying only on        /// the file watcher to handle file processing events.        /// </summary>        None    }     /// <summary>    /// Arguments to events triggered by the <see cref="FileProcessor"/>.    /// </summary>    public class FileProcessorEventArgs : EventArgs    {        #region [ Members ]         // Fields        private readonly string m_fullPath;        private readonly bool m_alreadyProcessed;        private bool m_requeue;         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new instance of the <see cref="FileProcessorEventArgs"/> class.        /// </summary>        /// <param name="fullPath">The full path to the file to be processed.</param>        /// <param name="alreadyProcessed">Flag indicating whether this file has been processed before.</param>        public FileProcessorEventArgs(string fullPath, bool alreadyProcessed)        {            m_fullPath = fullPath;            m_alreadyProcessed = alreadyProcessed;        }         #endregion         #region [ Properties ]         /// <summary>        /// Gets the full path to the file to be processed.        /// </summary>        public string FullPath        {            get            {                return m_fullPath;            }        }         /// <summary>        /// Gets the flag that indicates whether this file has been processed before.        /// </summary>        public bool AlreadyProcessed        {            get            {                return m_alreadyProcessed;            }        }         /// <summary>        /// Gets or sets a flag that indicates whether the        /// file should be requeued by the file processor.        /// </summary>        public bool Requeue        {            get            {                return m_requeue;            }            set            {                m_requeue = value;            }        }         #endregion    }     /// <summary>    /// Tracks files processed in a list of directories, and    /// notifies when new files become available to be processed.    /// </summary>    public sealed class FileProcessor : IDisposable    {        #region [ Members ]         // Nested Types         private class EnumerableWrapper : IDisposable        {            #region [ Members ]             // Fields            private IEnumerable<string> m_enumerable;            private IEnumerator<string> m_enumerator;            private bool m_lastMove;            private string m_current;             private CancellationToken m_cancellationToken;            private bool m_disposed;             #endregion             #region [ Constructors ]             /// <summary>            /// Creates a new instance of the <see cref="EnumerableWrapper"/> class.            /// </summary>            /// <param name="enumerable">The enumerable to be wrapped.</param>            /// <param name="cancellationToken">The token used to cancel the enumeration operation.</param>            public EnumerableWrapper(IEnumerable<string> enumerable, CancellationToken cancellationToken)            {                if ((object)enumerable == null)                    throw new ArgumentNullException(nameof(enumerable));                 m_enumerable = enumerable;                m_enumerator = enumerable.GetEnumerator();                m_cancellationToken = cancellationToken;            }             #endregion             #region [ Properties ]             /// <summary>            /// Gets the value returned by the            /// last call to <see cref="MoveNext"/>.            /// </summary>            public bool LastMove            {                get                {                    return m_lastMove;                }            }             /// <summary>            /// Gets the current value of the enumerator.            /// </summary>            public string Current            {                get                {                    return m_current;                }            }             /// <summary>            /// Gets the cancellation token for this enumerator.            /// </summary>            public CancellationToken CancellationToken            {                get                {                    return m_cancellationToken;                }            }             #endregion             #region [ Methods ]             /// <summary>            /// Advances the enumerator to the next element of the collection.            /// </summary>            /// <returns>True if the enumerator was successfully advanced to the next element; false if the enumerator has passed the end of the collection.</returns>            public bool MoveNext()            {                m_lastMove = (!m_cancellationToken.IsCancelled) && m_enumerator.MoveNext();                m_current = m_lastMove ? m_enumerator.Current : null;                return m_lastMove;            }             /// <summary>            /// Releases all the resources used by the enumerable.            /// </summary>            public void Dispose()            {                if (!m_disposed)                {                    try                    {                        (m_enumerator as IDisposable)?.Dispose();                        (m_enumerable as IDisposable)?.Dispose();                    }                    finally                    {                        m_lastMove = false;                        m_current = null;                        m_disposed = true;                    }                }            }             #endregion        }         private class FileEnumerator        {            #region [ Members ]             // Fields            private FileProcessor m_fileProcessor;            private Dictionary<string, CancellationToken> m_cancellationTokens;            private int m_enumerationThreads;             private LogicalThread m_sequentialEnumerationThread;            private LogicalThreadLocal<bool> m_isActive;            private LogicalThreadLocal<Stack<Action>> m_wrapperStack;            private LogicalThreadLocal<Queue<Action>> m_directoryQueue;             private LogicalThread m_cleanProcessedFilesThread;            private LogicalThreadOperation m_cleanProcessedFilesOperation;             #endregion             #region [ Properties ]             /// <summary>            /// Gets the number of active enumeration threads.            /// </summary>            public int EnumerationThreads            {                get                {                    return Interlocked.CompareExchange(ref m_enumerationThreads, 0, 0);                }            }             #endregion             #region [ Constructors ]             /// <summary>            /// Creates a new instance of the <see cref="FileEnumerator"/> class.            /// </summary>            /// <param name="fileProcessor">The file processor that created the file enumerator.</param>            public FileEnumerator(FileProcessor fileProcessor)            {                m_fileProcessor = fileProcessor;                m_cancellationTokens = new Dictionary<string, CancellationToken>();                 m_sequentialEnumerationThread = m_fileProcessor.m_threadScheduler.CreateThread();                m_isActive = new LogicalThreadLocal<bool>();                m_wrapperStack = new LogicalThreadLocal<Stack<Action>>(() => new Stack<Action>());                m_directoryQueue = new LogicalThreadLocal<Queue<Action>>(() => new Queue<Action>());                 m_cleanProcessedFilesThread = m_fileProcessor.m_threadScheduler.CreateThread();                m_cleanProcessedFilesOperation = new LogicalThreadOperation(m_cleanProcessedFilesThread, GetProcessedFiles, false);            }             #endregion             #region [ Methods ]             /// <summary>            /// Initiates enumeration of the watch directories.            /// </summary>            /// <param name="directory">The directory to be enumerated.</param>            public void Enumerate(string directory)            {                CancellationToken cancellationToken;                 lock (m_cancellationTokens)                {                    if (m_cancellationTokens.TryGetValue(directory, out cancellationToken))                        cancellationToken.Cancel();                     cancellationToken = new CancellationToken();                    m_cancellationTokens[directory] = cancellationToken;                }                 switch (m_fileProcessor.EnumerationStrategy)                {                    // Sequential enumeration strategy kicks                    // off all its processing on a single thread                    case FileEnumerationStrategy.Sequential:                        m_sequentialEnumerationThread.Push(() =>                        {                            // Create the enumerable wrappers for file and directory enumeration                            EnumerableWrapper fileWrapper = new EnumerableWrapper(Directory.EnumerateFiles(directory), cancellationToken);                            EnumerableWrapper directoryWrapper = new EnumerableWrapper(Directory.EnumerateDirectories(directory), cancellationToken);                             if (m_isActive.Value)                            {                                // If the thread is already active,                                // push processing into the directory queue                                m_directoryQueue.Value.Enqueue(() =>                                {                                    m_wrapperStack.Value.Push(() => EnumerateNextFile(fileWrapper));                                    EnumerateNextDirectory(directoryWrapper);                                });                            }                            else                            {                                // If the thread is inactive, mark it as active                                // and then begin processing the directory wrapper                                ActivateThread();                                m_wrapperStack.Value.Push(() => EnumerateNextFile(fileWrapper));                                EnumerateNextDirectory(directoryWrapper);                            }                        });                        break;                     // Parallel processing kicks off new                    // enumerations on their own thread                    case FileEnumerationStrategy.ParallelWatchDirectories:                    case FileEnumerationStrategy.ParallelSubdirectories:                        m_fileProcessor.m_threadScheduler.CreateThread().Push(() =>                        {                            EnumerableWrapper fileWrapper = new EnumerableWrapper(Directory.EnumerateFiles(directory), cancellationToken);                            EnumerableWrapper directoryWrapper = new EnumerableWrapper(Directory.EnumerateDirectories(directory), cancellationToken);                            m_wrapperStack.Value.Push(() => EnumerateNextFile(fileWrapper));                            LogicalThread.CurrentThread.Push(() => EnumerateNextDirectory(directoryWrapper));                            ActivateThread();                        });                        break;                }                 // Initiate the process to clean up the files                // in the collection of processed files                m_cleanProcessedFilesOperation.RunOnceAsync();            }             /// <summary>            /// Cancels enumeration for a specific directory.            /// </summary>            /// <param name="directory">The directory for which enumeration should be cancelled.</param>            public void Cancel(string directory)            {                CancellationToken cancellationToken;                 lock (m_cancellationTokens)                {                    if (!m_cancellationTokens.TryGetValue(directory, out cancellationToken))                        return;                     m_cancellationTokens.Remove(directory);                }                 cancellationToken.Cancel();            }             /// <summary>            /// Cancels all running enumeration operations.            /// </summary>            public void Cancel()            {                CancellationToken[] cancellationTokens;                 lock (m_cancellationTokens)                {                    cancellationTokens = m_cancellationTokens.Values.ToArray();                    m_cancellationTokens.Clear();                }                 foreach (CancellationToken cancellationToken in cancellationTokens)                    cancellationToken.Cancel();            }             // Handle the next directory in the enumeration.            private void EnumerateNextDirectory(EnumerableWrapper wrapper)            {                EnumerableWrapper fileWrapper;                EnumerableWrapper directoryWrapper;                string directory;                 try                {                    // Advance directory enumeration                    if (!wrapper.MoveNext())                    {                        // No more directories, so dispose                        // and then move to the next wrapper                        wrapper.Dispose();                        EnumerateNextWrapper();                        return;                    }                     // Initialize the fileWrapper                    // and directoryWrapper                    directory = wrapper.Current;                    fileWrapper = null;                    directoryWrapper = null;                }                catch                {                    // If an error occurs, dispose of the                    // wrapper and then move to the next wrapper                    wrapper.Dispose();                    EnumerateNextWrapper();                    throw;                }                 try                {                    switch (m_fileProcessor.EnumerationStrategy)                    {                        // Sequential and ParallelWatchDirectories strategies                        // place subdirectories on the current thread                        case FileEnumerationStrategy.Sequential:                        case FileEnumerationStrategy.ParallelWatchDirectories:                            // Create the fileWrapper and directoryWrapper objects                            fileWrapper = new EnumerableWrapper(Directory.EnumerateFiles(directory), wrapper.CancellationToken);                            directoryWrapper = new EnumerableWrapper(Directory.EnumerateDirectories(directory), wrapper.CancellationToken);                             // Push the current directory wrapper onto the stack                            m_wrapperStack.Value.Push(() => EnumerateNextDirectory(wrapper));                             // Push the subdirectory's file wrapper onto the stack                            m_wrapperStack.Value.Push(() => EnumerateNextFile(fileWrapper));                             // Continue enumeration with the directory wrapper for the subdirectory                            LogicalThread.CurrentThread.Push(() => EnumerateNextDirectory(directoryWrapper));                            break;                         // ParallelSubdirectories strategy spawns new threads for subdirectories                        case FileEnumerationStrategy.ParallelSubdirectories:                            // Create the fileWrapper and directoryWrapper objects                            fileWrapper = new EnumerableWrapper(Directory.EnumerateFiles(directory), wrapper.CancellationToken);                            directoryWrapper = new EnumerableWrapper(Directory.EnumerateDirectories(directory), wrapper.CancellationToken);                             // Create a new thread, push the file wrapper onto the new thread's                            // stack, then enumerate the directory wrapper on the new thread                            m_fileProcessor.m_threadScheduler.CreateThread().Push(() =>                            {                                ActivateThread();                                m_wrapperStack.Value.Push(() => EnumerateNextFile(fileWrapper));                                EnumerateNextDirectory(directoryWrapper);                            });                             // Continue enumeration on this thread with the current directory wrapper                            LogicalThread.CurrentThread.Push(() => EnumerateNextDirectory(wrapper));                            break;                         // The only other file enumeration                        // strategy is no strategy at all                        default:                            // Clear out the stack and queue,

Lines 1–500 of 1534. Load more