Using up to 15 connections to synchronize files from a FTP server with local files

Question

Task: Create a service which will synchronize files on FTP server with files on local drive.

1. FTP server allows up to 15 concurent connections
2. Synchronize the files as fast as possible
3. Files on FTP server are archived after 7 day, we should archive them too at our side
4. Files once put on the FTP server do not change
5. Folder structure of FTP server is unknown upfront
6. There are ~1,5K folders containing ~70K files on the FTP server.

I did the task. The service is running stable and synchronizes files within 10min usually. This was my first big attempt on async/await/parallelism.

I'm especially interesed in comments regarding the ParallelFtpClient. It uses the package FluentFTP.

I've tried to strip unimportant details from the code.

Program.cs and "main" loop:

class Program
{
    private static ILog _logger;
    private Task _mainTask;
    private CancellationTokenSource _cancellationTokenSource;

    public static void Main()
    {
        var dataBase = new DataBase(...);
        var ftpSyncer = new FtpSyncer(config.LocalShare, config.FtpHost, config.FtpPort, config.FtpUser,
            config.FtpPassword, config.MaxConcurrentFtpConnections, config.FtpRootDirectory, dataBase);
        _cancellationTokenSource = new CancellationTokenSource();
        var cancellationToken = _cancellationTokenSource.Token;
        _mainTask = Task.Run(async () =>
        {
            while (!cancellationToken.IsCancellationRequested)
            {
                try
                {
                    LogicalThreadContext.Properties["CorrelationId"] = Guid.NewGuid();
                    _logger.Info("Synchronizing files from FTP...");
                    await ftpSyncer.SynchronizeFiles();
                    _logger.Info("Done synchronizing files from FTP");
                }
                catch (Exception e)
                {
                    _logger.Error("An unhandled excpetion occured while synchronizing files. Execution will continue", e);
                }

                try
                {
                    _logger.Info($"Sleeping for {config.DelayBetweenRunsSeconds}s before next Ftp synchronization");
                    await Task.Delay(TimeSpan.FromSeconds(config.DelayBetweenRunsSeconds), cancellationToken);
                }
                catch (OperationCanceledException)
                {
                    // just exit loop
                }
            }
        });
        _logger.Info("Service started");
    }

    private void Stop()
    {
        _logger.Info("Service stopping...");
        _cancellationTokenSource.Cancel();
        _logger.Info("Waiting for the main task to stop...");
        _mainTask.Wait();
        _logger.Info("Done waiting for the main task to stop...");
        }
}

FtpSyncer.cs does the job of synchronizing files:

public class FtpSyncer
{
    private readonly ILog _logger = LogManager.GetLogger(typeof(FtpSyncer).FullName);
    private readonly string _localShare;
    private readonly string _ftpHost;
    private readonly string _ftpPass;
    private readonly int _ftpPort;
    private readonly string _ftpUser;
    private readonly int _maxConcurentFtpConnections;
    private readonly Database _database;
    private readonly string _ftpRootDirectory;
    private readonly int MAX_DATABASE_CONNECTION_RETRIES = 3;
    private readonly int MAX_DOWNLOAD_RETIRES = 3;

    public FtpSyncer(string localShare, string ftpHost, int ftpPort, string ftpUser, string ftpPassword,
        int maxConcurentFtpConnections, string ftpRootDirectory, Database database)
    {
        _localShare = localShare;
        _ftpHost = ftpHost;
        _ftpPort = ftpPort;
        _ftpUser = ftpUser;
        _ftpPass = ftpPassword;
        _maxConcurentFtpConnections = maxConcurentFtpConnections;
        _database = database;
        _ftpRootDirectory = ftpRootDirectory;
    }

    public async Task SynchronizeFiles()
    {

        IFtpClient factory() => new FtpClient(_ftpHost, _ftpPort, _ftpUser, _ftpPass);
        using (var parallelFtpClient = new ParallelFtpClient(_maxConcurentFtpConnections, factory))
        {
            var gettingRemoteFiles = Task.Run(() =>
            {
                _logger.Info("Connecting to FTP...");
                parallelFtpClient.Connect();

                _logger.Info($"Querying for directories under {_ftpHost}{_ftpRootDirectory} with files on FTP (excluding Archive directories)...");
                var directoriesWithFiles = parallelFtpClient.ListNonArchiveDirectoresWithFiles(_ftpRootDirectory);
                _logger.Info($"There are {directoriesWithFiles.Count} directories with files on {_ftpHost}{_ftpRootDirectory}");

                _logger.Info("Querying for files on FTP...");
                var allFtpFiles = parallelFtpClient.ListFilesInDirectories(directoriesWithFiles);
                _logger.Info($"There are {allFtpFiles.Count} files under {_ftpHost}{_ftpRootDirectory}");
                return allFtpFiles;
            });

            var gettingLocalFiles = Task.Run(() =>
            {
                _logger.Info("Querying for files already in local share...");
                var allFilesOnShare = Directory.GetFiles(Path.Combine(_localShare, "20160301"), "*",
                    SearchOption.AllDirectories);
                _logger.Info($"There are {allFilesOnShare.Length} files in {_localShare}");

                // unify paths to format used by FTP
                var allFilesOnShareFtpPathFormat = allFilesOnShare
                    .Select(x => x.Replace(_localShare, ""))
                    .Select(x => x.Replace(@"\", "/"))
                    .ToArray();

                return allFilesOnShareFtpPathFormat;
            });

            var filesOnFtp = await gettingRemoteFiles;
            var filesOnShare = await gettingLocalFiles;

            var toBeDownloaded = filesOnFtp
                .Except(filesOnShare, StringComparer.OrdinalIgnoreCase) // windows paths are case-insensitive
                // remove leading slash '/' from filePath (if you don't remove it Path.Combine() will not combine paths)
                .Select(ftpFilePath => (ftpFilePath, Path.Combine(_localShare, ftpFilePath.Substring(1))))
                .ToList();
            _logger.Info($"There are {toBeDownloaded.Count} files to be downloaded");

            _logger.Info($"Downloading files to {_localShare}...");
            List<(string, string)> failedDownloads;
            var downloadedFiles = DownloadFilesWithRetires(parallelFtpClient, toBeDownloaded, out failedDownloads);
            if (failedDownloads.Any())
            {
                var dump = string.Join("\n", failedDownloads);
                _logger.Error($"{failedDownloads.Count} files failed to be downloaded after {MAX_DOWNLOAD_RETIRES} attempts. " +
                            $"List:\n" + dump);
            }
            _logger.Info($"Done downloading files, {downloadedFiles.Count} files downloaded");

            _logger.Info("Disconnecting FTP clinets...");
            parallelFtpClient.Disconnect();
            _logger.Info("Done disconnecting FTP clinets");

            _logger.Info("Registering files...");
            var registerInfos = MapFileToDatabaseRegisterInfo.Map(downloadedFiles);
            var toNotBeRegistered = registerInfos.Where(x => x.FileTypes.Count == 0).ToArray();
            var toBeRegistered = registerInfos.Where(x => x.FileTypes.Count == 1).ToArray();
            var filesWithMultipleMatchingFileTypes = registerInfos.Where(x => x.FileTypes.Count == 2).ToArray();
            _logger.Info($"There are {toBeRegistered.Length} files to be registered");
            _logger.Info($"There are {toNotBeRegistered.Length} files which will not be registered");

            if (filesWithMultipleMatchingFileTypes.Any())
            {
                _logger.Warn($"There are {filesWithMultipleMatchingFileTypes.Length} with multiple matching Database FileTypes");
                foreach (var registerInfo in filesWithMultipleMatchingFileTypes)
                {
                    var listing = string.Join(", ", registerInfo.FileTypes);
                    _logger.Warn($"File {registerInfo.FullFilePath} was matched with FileTypes: {listing}");
                }
            }

            var filesAreRegistered = RegisterInDatabaseWithRetries(toBeRegistered);
            if (filesAreRegistered)
            {
                _logger.Info("Done registering files");
            }
            else
            {
                var dump = string.Join("\n", toBeRegistered.Select(x => x.ToJson()));
                _logger.Error($"Could not register files after {MAX_DATABASE_CONNECTION_RETRIES} retries. " +
                            $"Dumping list of files which should be registered: \n{dump}");
            }

            _logger.Info("Archiving files...");
            var toBeArchived = filesOnShare
                .Except(filesOnFtp, StringComparer.OrdinalIgnoreCase)  // windows paths are case-insensitive
                .Select(x => x.Substring(1)) // remove leading '/' so the path will be treated as relative
                .Select(x => Path.Combine(_localShare, x))
                .ToArray();
            _logger.Info($"There are {toBeArchived.Length} files to be archived");
            MeteologicaFilesArchiver.Archive(toBeArchived);
            _logger.Info("Done Archiving files...");
        }
    }

    private List<string> DownloadFilesWithRetires(ParallelFtpClient parallelFtpClient, List<(string, string)> toBeDownloaded, out List<(string, string)> failedDownloads)
    {
        List<string> allDownloadedFiles = new List<string>();
        var attempt = 1;
        do
        {
            _logger.Info($"Downloading files - attempt {attempt}/{MAX_DOWNLOAD_RETIRES}");
            var downloadedFiles = parallelFtpClient.DownloadFilesParallel(toBeDownloaded, out failedDownloads);
            allDownloadedFiles.AddRange(downloadedFiles);
            if (failedDownloads.Any())
            {
                _logger.Warn($"{failedDownloads.Count} files were not downloaded. This was the {attempt}/{MAX_DOWNLOAD_RETIRES} attempt");
                toBeDownloaded = failedDownloads;
                attempt++;
            }
        } while (attempt <= MAX_DOWNLOAD_RETIRES && failedDownloads.Any());

        return allDownloadedFiles;
    }

    private bool RegisterInDatabaseWithRetries(RegisterInfo[] toBeRegistered)
    {
        var retryCount = 1;
        var filesAreRegistered = false;
        while (filesAreRegistered == false && retryCount <= MAX_DATABASE_CONNECTION_RETRIES)
        {
            try
            {
                _database.Register(toBeRegistered);
                filesAreRegistered = true;
            }
            catch (SqlException e) // catch this excetpion because it might be a timeout
            {
                _logger.Warn($"An exception occured while trying to register files in Database. " +
                            $"This was the {retryCount} attempt out of max={MAX_DATABASE_CONNECTION_RETRIES} attempts.", e);
                retryCount++;
                if (retryCount <= MAX_DATABASE_CONNECTION_RETRIES)
                {
                    var sleepTime = 10 * Math.Pow(2, retryCount);
                    _logger.Info($"Sleeping {sleepTime}s before next attempt...");
                    Thread.Sleep(TimeSpan.FromSeconds(sleepTime));
                }
            }
        }
        return filesAreRegistered;
    }
}

ParallelFtpClient.cs usses 15 connections to get info/data from the FTP server as fast as possible:

public class ParallelFtpClient : IDisposable
{
    private readonly ILog _logger = LogManager.GetLogger(...);
    private readonly int _maxConcurentConnections;
    private readonly Func<IFtpClient> _factory;
    private readonly List<IFtpClient> _ftpClients;

    public ParallelFtpClient(int maxConcurentConnections, Func<IFtpClient> factory)
    {
        _maxConcurentConnections = maxConcurentConnections;
        _factory = factory;
        _ftpClients = new List<IFtpClient>();
    }

    public void Connect()
    {
        var tasks = new List<Task>();
        for (int i = 0; i < _maxConcurentConnections; i++)
        {
            var ftpClient = _factory();
            _ftpClients.Add(ftpClient);
            var connectTask = ftpClient.ConnectAsync();
            tasks.Add(connectTask);
        }
        Task.WaitAll(tasks.ToArray());
    }

    public void Disconnect()
    {
        var tasks = _ftpClients
            .Select(x => x.DisconnectAsync())
            .ToArray();
        Task.WaitAll(tasks);
    }

    public List<string> ListFilesInDirectories(IEnumerable<string> directories)
    {
        var items = new ConcurrentBag<FtpListItem>();
        var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);

        Parallel.ForEach(directories,
            new ParallelOptions() {MaxDegreeOfParallelism = _maxConcurentConnections},
            (path) =>
            {
                IFtpClient client;
                clients.TryDequeue(out client);
                var listing = client.GetListing(path);
                foreach (var item in listing)
                {
                    items.Add(item);
                }
                clients.Enqueue(client);
            });

        var list = items
            .Where(x => x.Type == FtpFileSystemObjectType.File)
            .Select(x => x.FullName)
            .ToList();
        return list;
    }

    public List<string> DownloadFilesParallel(IEnumerable<(string,string)> ftpPathLocalPathPairs, out List<(string, string)> failedDownloads)
    {
        var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);
        var downloadedFiles = new ConcurrentBag<string>();
        var failedDownloadsBag = new ConcurrentBag<(string,string)>();

        Parallel.ForEach(ftpPathLocalPathPairs,
            new ParallelOptions() {MaxDegreeOfParallelism = _maxConcurentConnections},
            (ftpPathLocalPathPair) =>
            {
                IFtpClient client;
                clients.TryDequeue(out client);
                var ftpFilePath = ftpPathLocalPathPair.Item1;
                var destinationFilePath = ftpPathLocalPathPair.Item2;
                _logger.Debug($"Downloading {ftpFilePath} to {destinationFilePath}...");
                try
                {
                    var memoryStream = new MemoryStream();
                    client.Download(memoryStream, ftpFilePath);
                    memoryStream.Position = 0;
                    string destinationDirectory = Path.GetDirectoryName(destinationFilePath);
                    if (!Directory.Exists(destinationDirectory))
                    {
                        _logger.Info($"Creating new directory {destinationDirectory}");
                        Directory.CreateDirectory(destinationDirectory);
                    }
                    File.WriteAllBytes(destinationFilePath, memoryStream.ToArray());
                    downloadedFiles.Add(destinationFilePath);
                }
                catch (Exception e)
                {
                    _logger.Warn($"An unhandled excetpion occured while downloading file {ftpFilePath}", e);
                    failedDownloadsBag.Add(ftpPathLocalPathPair);
                }

                clients.Enqueue(client);
            }
        );

        failedDownloads = failedDownloadsBag.ToList();

        return downloadedFiles.ToList();
    }

    public List<string> ListNonArchiveDirectoresWithFiles(string root)
    {
        var directoriesWithFiles = new List<string>();
        var directoiresToQuery = new Queue<string>();
        var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);

        directoiresToQuery.Enqueue(root);
        var tasks = new List<Task<Result>>();
        while (directoiresToQuery.Any() || tasks.Any())
        {
            while (clients.Any() && directoiresToQuery.Any())
            {
                var path = directoiresToQuery.Dequeue();
                IFtpClient client;
                clients.TryDequeue(out client);
                var task = Task.Run(() =>
                {
                    var childItems = client.GetListing(path);
                    clients.Enqueue(client);
                    return new Result()
                    {
                        FullName = path,
                        Children = childItems
                    };
                });
                tasks.Add(task);
            }

            var array = tasks.ToArray();
            var finishedTaskIndex = Task.WaitAny(array);
            var finishedTask = array[finishedTaskIndex];
            var finishedTaskResult = finishedTask.Result;
            if (finishedTaskResult.Children.Any(i => i.Type == FtpFileSystemObjectType.File))
            {
                directoriesWithFiles.Add(finishedTaskResult.FullName);
            }
            var subdirectoriesToQuery = finishedTaskResult
                .Children
                .Where(x => x.Type == FtpFileSystemObjectType.Directory)
                .Where(x => x.Name != "Archive")
                .Select(x => x.FullName);
            foreach (var subdirectoryToQuery in subdirectoriesToQuery)
            {
                directoiresToQuery.Enqueue(subdirectoryToQuery);
            }

            tasks = array
                .Except(new[] {finishedTask})
                .ToList();
        }

        return directoriesWithFiles;
    }

    public void Dispose()
    {
        foreach (var ftpClient in _ftpClients)
        {
            ftpClient.Dispose();
        }
    }
}

public class Result
{
    public string FullName { get; set; }
    public FtpListItem[] Children { get; set; }
}