Есть ли какой-нибудь способ конвертировать массив или коллекцию List в synchronizedCollection?

Question

[Cod Rob]

Вообще пытаюсь использовать ConcurrentBag в своей задаче, но если получится его заменить synchronizedCollection по некоторым причинам, то улучшу читабельность кода.

Comments

[Сергей Соловьев]

если бы еще контекст проблемы знали

[Cod Rob]

Сергей Соловьев, У меня есть такой класс:

internal class EmailReaderWriter : IEmailReaderWriter
{
    public ConcurrentBag<string> ReadEmails { get; private set; }
    public ConcurrentBag<string> WrittenEmails { get; private set; }
    public string FolderPath { get; }

    private delegate void EmailReaderWriterHandler(string message);
    private event EmailReaderWriterHandler SuccessNotification, ErrorNotification;
    private FileOptions customFileOptions;

    public EmailReaderWriter(string folderpath)
    {
        FolderPath = folderpath;
        ReadEmails = new ConcurrentBag<string>();
        WrittenEmails = new ConcurrentBag<string>();
        SuccessNotification += OperationSuccessNotifier.Success;
        ErrorNotification += OperationSuccessNotifier.Error;
        customFileOptions = FileOptions.Asynchronous | FileOptions.SequentialScan;
    }

    public async Task<ConcurrentBag<string>?> ReadAsync(string filepath)
    {
        if (!File.Exists(filepath))
        {
            ErrorNotification($"File {filepath} doesn't exist!");
            return null;
        }

        FileStream fstream = new FileStream(filepath, FileMode.Open,
            FileAccess.Read, FileShare.None, 0, customFileOptions);
        StreamReader reader = new StreamReader(fstream);

        var allEmails = new ConcurrentBag<string>(
            (
            await reader.ReadToEndAsync()).
            Split(new string[] { " ", ". ", ", " }, StringSplitOptions.RemoveEmptyEntries).
            Where(x => x.Contains("@gmail.com"))
            );

        if (allEmails.Count() == 0)
        {
            reader.Close();
            fstream.Close();
            ErrorNotification($"0 emails were found!");
            return null;
        }

        ConcurrentBag<string> nonReadEmails = new ConcurrentBag<string>();
        int readEmailsAddedCounter = 0;

        foreach (var email in allEmails)
        {
            if (ReadEmails.Contains(email))
            {
                ErrorNotification($"Email {email} is already read!");
                continue;
            }
            else
            {
                readEmailsAddedCounter++;
                nonReadEmails.Add(email);
                ReadEmails.Add(email);
                SuccessNotification($"Email {email} is read!");
            }
        }

        if (readEmailsAddedCounter > 0)
            SuccessNotification($"Such quantity of emails was read: {readEmailsAddedCounter}!");
        reader.Close();
        fstream.Close();
        allEmails.Clear();
        return nonReadEmails;
    }

    public async Task<bool> WriteAsync(string filepath, string email)
    {
        if (WrittenEmails.Contains(email))
        {
            ErrorNotification($"Email {email} is already written on file!");
            return false;
        }

        using (FileStream fstream = new FileStream(filepath, FileMode.Append,
            FileAccess.Write, FileShare.None, 0, customFileOptions |
            FileOptions.WriteThrough))
        {
            using (StreamWriter writer = new StreamWriter(fstream))
            {
                await writer.WriteLineAsync(email);
                WrittenEmails.Add(email);
            }
        }

        SuccessNotification($"Email {email} is successfully written on file!");
        return true;
    }
}

Не обращайте внимания на класс OperationSuccessNotifier в конструкторе. Вопрос у меня в том, как я могу метод ReadAsync, который из файла выбирает имэйлы и возвращает их, сделать проще.

[Сергей Соловьев]

Cod Rob, здесь вообще никакого Concurrency нет.
Во-первых, внутри метода достаточно использовать List
Во-вторых, для ReadEmails и других полей объекта используй обычные List, а когда захочешь их обновлять/читать - просто используй на них lock

Проблемы из пустого места

P.S. почитай что такое using

[Cod Rob]

Сергей Соловьев, дело в том, что чуть позже я это Concurrency добавлю. Функционал не до конца сделан

[Cod Rob]

Сергей Соловьев, Если коротко, то весь этот сыр бор устроен из-за того, что будет три потока, с помощью которых из кучи текстовых файлов будут выбираться имэйлы. И просто, например, то же свойство ReadEmails - почему я сделал его ConcurrentBag? А потому что проверка на то, есть ли в этом свойстве имэйл, который мы считали, может произойти параллельно со вторым потоком, который этот имэйл якобы будет в первый раз считывать. В общем, если там будет не ConcurrentBag, то проверка может происходить некорректно

[Сергей Соловьев]

Cod Rob, где его можно добавить? внутри метода все синхронно происходит - никаких гонок нет, а если нужно просто параллельно добавлять записи в список, то лок на объект будет гораздо эффективнее, чем ConcurrentBag

[Cod Rob]

Сергей Соловьев, Сергей Соловьев, Тип, у мя задача такая: есть много текстовых файлов, заполненных рандомным текстом, в котором иногда встречаются имэйлы. Нужно, используя три потока, параллельно считывать данных с этих файлов, при этом после того, как какой-то поток считал имэйлы, он должен записать их на один файл, при этом всех три потока должны записывать считаннные данные на этот один файл, не на разные файлы.

[Cod Rob]

Сергей Соловьев, не спрашую решение задачи, ибо сам хочу решить. ПРосто хотел сделать метод ReadAsync более лаконичным

[Cod Rob]

Сергей Соловьев, типа, сделать lock внутри using'a? Например, пусть будет:

public List<string> ReadEmails { get; private set; }

Тогда:

public async Task<ConcurrentBag<string>?> ReadAsync(string filepath)
    {
        if (!File.Exists(filepath))
        {
            ErrorNotification($"File {filepath} doesn't exist!");
            return null;
        }

        FileStream fstream = new FileStream(filepath, FileMode.Open,
            FileAccess.Read, FileShare.None, 0, customFileOptions);
        StreamReader reader = new StreamReader(fstream);

        var allEmails = new ConcurrentBag<string>(
            (
            await reader.ReadToEndAsync()).
            Split(new string[] { " ", ". ", ", " }, StringSplitOptions.RemoveEmptyEntries).
            Where(x => x.Contains("@gmail.com"))
            );

        if (allEmails.Count() == 0)
        {
            reader.Close();
            fstream.Close();
            ErrorNotification($"0 emails were found!");
            return null;
        }

        ConcurrentBag<string> nonReadEmails = new ConcurrentBag<string>();
        int readEmailsAddedCounter = 0;

        foreach (var email in allEmails)
        {
            if (ReadEmails.Contains(email))
            {
                ErrorNotification($"Email {email} is already read!");
                continue;
            }
            else
            {
                readEmailsAddedCounter++;
                nonReadEmails.Add(email);
lock(locker)
{
                ReadEmails.Add(email);
}
                SuccessNotification($"Email {email} is read!");
            }
        }

        if (readEmailsAddedCounter > 0)
            SuccessNotification($"Such quantity of emails was read: {readEmailsAddedCounter}!");
        reader.Close();
        fstream.Close();
        allEmails.Clear();
        return nonReadEmails;
    }

[Cod Rob]

Cod Rob, ну, это как пример

[Cod Rob]

Сергей Соловьев, типа, а в самом классе сделать приватный object locker?

[Cod Rob]

Сергей Соловьев, а, стоп, или вы имеете в виду поместить всю конструкцию if-else в локер?

lock(locker)
{
if (ReadEmails.Contains(email))
            {
                ErrorNotification($"Email {email} is already read!");
                continue;
            }
            else
            {
                readEmailsAddedCounter++;
                nonReadEmails.Add(email);
                ReadEmails.Add(email);
                SuccessNotification($"Email {email} is read!");
            }
}

[Сергей Соловьев]

Cod Rob, значит лучше сделать такую реализацию:
1. Весь список файлов делишь между этими 3 потоками
2. Каждый поток синхронно работает с этими файлами - блокировка не нужна
3.1. Для записи в файл результирующий пусть у тебя будет отдельный объект, который через лок это и делает
3.2. Либо это будет отдельный поток, а синхронизация между ними реализована с помощью BlockingCollection

P.S. через 3.2. мне больше нравится,

[Cod Rob]

Сергей Соловьев, да, можно и синхронно работать с файлами, если бы не одна проблема: в двух разных файлах могут быть одинаковые имэйлы, из-за чего, если делать синхронно и делать параллельно через блокировку и потокобезопасные коллекции, то результат вывода считывания может быть разным. Синхронный подход в данном случае выглядит не очень надёжным и корректным

[Cod Rob]

Сергей Соловьев, типа, мне нужно считать имэйл, когда я его в первый раз встретил, и больше не считывать. А если синхронно сделать, то может случиться так, что имэйл считается, например, вторым потоком в 5-ом файле первый раз в то время, как он мог считаться первым потоком в 3-ем файле

[Сергей Соловьев]

Cod Rob, так запись происходит в другом месте (другой объект/поток) - он пусть это и отловит

[Cod Rob]

Сергей Соловьев, запись - да, но с методом ReadAsync будет работать несколько потоков, и, типа, может возникнуть ситуация, что, например, есть одинаковый имэйл в файле №3 и файле №5, и вот из-за синхронности (на сколько я понимаю) проверка по наличию считанных файлов в свойстве ReadEmails может сработать некорректно

[Cod Rob]

Сергей Соловьев, UPD: методом ReadAsync и объектом ReadEmails

[Cod Rob]

Сергей Соловьев, или, думаете, лока для этой части кода в методе:, где как раз происходит считывание и работа с ReadEmails

var allEmails = new ConcurrentBag<string>(
            (
            await reader.ReadToEndAsync()).
            Split(new string[] { " ", ". ", ", " }, StringSplitOptions.RemoveEmptyEntries).
            Where(x => x.Contains("@gmail.com"))
            );

        if (allEmails.Count() == 0)
        {
            reader.Close();
            fstream.Close();
            ErrorNotification($"0 emails were found!");
            return null;
        }

        ConcurrentBag<string> nonReadEmails = new ConcurrentBag<string>();
        int readEmailsAddedCounter = 0;

        foreach (var email in allEmails)
        {
            if (ReadEmails.Contains(email))
            {
                ErrorNotification($"Email {email} is already read!");
                continue;
            }
            else
            {
                readEmailsAddedCounter++;
                nonReadEmails.Add(email);
                ReadEmails.Add(email);
                SuccessNotification($"Email {email} is read!");
            }
        }

        if (readEmailsAddedCounter > 0)
            SuccessNotification($"Such quantity of emails was read: {readEmailsAddedCounter}!");
        reader.Close();
        fstream.Close();
        allEmails.Clear();

будет достаточно?
Я просто щас вот призадумался над вашей идеей... Можно в теории поставить локер, но стоит ли на такой большой участок кода ставить локер? Это практикуется?

[Сергей Соловьев]

Cod Rob, используй ConcurrentDictionary и если вставка удалась - добавляй в результирующий файл. Самое простое решение

[Cod Rob]

Сергей Соловьев, можно, конечно, заменить ConcurrentBag на ConcurrentDictionary, но разве это не равносильно будет? Типа, речь пока не идёт о записи в файл, давайте забудем об этом. Я имею в виду щас метод ReadAsync. Стоит ли заменить все ConcurrentBag Листами и поставить локер, как я поставил выше, или такое использование локера не рекомендуется? Просто, если поставить локер, то будет приблизительно так:

Я только не заменил ConcurrentBag на лист, это просто для примера

[Сергей Соловьев]

Cod Rob, предлагаю сначала выучить теорию по параллельности, многопоточности и структурах данных для них. вижу многое не понятно пока, а подставлять объекты и думать что все заработает магическим образом - плохая затея

[Cod Rob]

Сергей Соловьев, так я и пытаюсь разобраться, какой вариант лучше: с ConcurrentBag или с Lock'ом и List'ом, потому и задаю вопросы.

[Сергей Соловьев]

Cod Rob, тебе надо просто верхнеуровнево нарисовать диаграмму работы программы, чтобы понять. что нужно использовать. тут вообще можно одним локом обойтись

[Cod Rob]

Сергей Соловьев, заменил на лок и листы таким образом:

Выбило исключение, что коллекция emails was modified. Переместил её создание и возврат результата из функции внутрь локера - всё равно исключение.

[Сергей Соловьев]

Cod Rob, потому что нельзя изменять коллекцию по которой в данный момент итерируешься.
Просто сделай такое изменение - сделай 2 переменные:
- currentReadEmails - в ней будут все прочитанные из файла имейлы
- resultEmails - тут будут те, что вернешь

В resultEmails только добавляешь, из currentReadEmails - только читаешь

[Cod Rob]

Сергей Соловьев, Переписал класс таким образом:

internal class EmailReaderWriter : IEmailReaderWriter
{
    public List<string> AllReadEmails { get; private set; }
    public List<string> AllWrittenEmails { get; private set; }
    public string InputPath { get; }
    public string OutputPath { get; }

    private delegate void EmailReaderWriterHandler(string message);
    private event EmailReaderWriterHandler _successNotification, _errorNotification;
    private FileOptions _customFileOptions;
    private object _locker;

    public EmailReaderWriter(string inputpath = "..\\..\\..\\Input", string outputpath = "..\\..\\..\\Output")
    {
        InputPath = inputpath;
        OutputPath = outputpath;
        AllReadEmails = new List<string>();
        AllWrittenEmails = new List<string>();
        _successNotification += OperationSuccessNotifier.Success;
        _errorNotification += OperationSuccessNotifier.Error;
        _customFileOptions = FileOptions.Asynchronous | FileOptions.SequentialScan;
        _locker = new object();
    }

    public List<string>? Read(string filepath)
    {
        if (!File.Exists(filepath))
        {
            _errorNotification($"File {filepath} doesn't exist!");
            return null;
        }

        List<string> allReadEmailsFromFile;

        using (FileStream fstream = new FileStream(filepath, FileMode.Open,
            FileAccess.Read, FileShare.None, 0, _customFileOptions))
        {
            using (StreamReader reader = new StreamReader(fstream))
            {
                allReadEmailsFromFile = new List<string>(
                    (
                    reader.ReadToEnd()).
                    Split(new string[] { " ", ". ", ", " }, StringSplitOptions.RemoveEmptyEntries).
                    Where(x => x.Contains("@gmail.com"))
                    );
            }
        }

        lock (_locker)
        {
            if (allReadEmailsFromFile.Count() == 0)
            {
                _errorNotification($"0 emails were found!");
                return null;
            }

            List<string> readEmails = new List<string>();

            foreach (var email in allReadEmailsFromFile)
            {
                if (AllReadEmails.Contains(email))
                {
                    _errorNotification($"Email {email} is already read!");
                    continue;
                }
                else
                {
                    readEmails.Add(email);
                    AllReadEmails.Add(email);
                    _successNotification($"Email {email} is read!");
                }
            }

            if (readEmails.Count > 0)
                _successNotification($"Such quantity of emails was read: {readEmails.Count}!");
            return readEmails;
        }

    }

    public bool Write(string filepath, string email)
    {
        if (!Directory.Exists(OutputPath))
            Directory.CreateDirectory(OutputPath);

        lock (_locker)
        {
            if (AllWrittenEmails.Contains(email))
            {
                _errorNotification($"Email {email} is already written on file!");
                return false;
            }
        }

        using (FileStream fstream = new FileStream(filepath, FileMode.Append,
            FileAccess.Write, FileShare.None, 0, _customFileOptions |
            FileOptions.WriteThrough))
        {
            using (StreamWriter writer = new StreamWriter(fstream))
            {
                writer.WriteLineAsync(email);
                AllWrittenEmails.Add(email);
                _successNotification($"Email {email} is successfully written on file!");
            }
        }

        return true;
    }
}

Что скажете?

Answer 1

[Сергей Соловьев]

SynchronizedCollection<T> - это легаси из .NET фреймворка, в современном рантмайме ты его не найдешь.
Могу предложить использовать:
- BlockingCollection<T>
- Channel<T>

Если есть возможность, то конвертируй задачу на Scatter/Gather манеру и необходимость в этих структурах данных отпадет.