Как передать данные в работающий поток?

Question

[Михаил Усоцкий]

У меня есть поток, который бесконечно работает в цикле. Но возникла одна проблема. Не могу передать потоку периодически загружаемые данные. Пробовал делегаты. Бесполезно.

public void Thread()
        {
            GetMessage msg = new GetMessage(Fail);
            GetCommands val = new GetCommands(Check);

            Ядро_хххх.BasicStructures.Command cmds = new Ядро_хххх.BasicStructures.Command();

            do
            {
                #region Стадия записи.

                if (cmd.Count > 0) cmds = cmd.ElementAt(0);

                cmd.TryDequeue(out cmds); // в пределах класса копится, но здесь остаётся пустым

                if (Device.CheckTxWaiting(ref numBytesWritten))
                    if (Commands.HasFlag(Ядро_хххх.BasicStructures.Command.NotOperation))
                    {
                        Device.Write(хххх.Upload(Outbox, Ядро_хххх.BasicStructures.Command.NotOperation), ref numBytesWritten);
                        IsButton = false;
                        RevisionTxFail = 0;
                    }
                    else
                    {
                        if (Commands.HasFlag(Ядро_хххх.BasicStructures.Command.LoadFPGAAll))
                        {
                            Device.Write(хххх.Upload(Outbox, Ядро_хххх.BasicStructures.Command.LoadFPGAAll), ref numBytesWritten);
                            //Commands = Ядро_хххх.BasicStructures.Command.NotOperation;
                        }
                        else
                        {
                            RevisionTxFail += 1;
                            if (RevisionTxFail > 1000) msg(5);
                        }
                    }
                #endregion

                #region Стадия чтения.
                if (Device.CheckRxQueue(ref numBytesAvailable))
                {
                    Device.Read(numBytesAvailable, ref ReadData, ref numBytesRead);
                    Inbox = хххх.Download(ReadData);
                    RevisionRxFail = 0;
                }
                else
                {
                    RevisionRxFail += 1;
                    if (RevisionRxFail > 1000) msg(5);
                }
                #endregion
            } while (IsWorking); // стоит всегда true, пока msg не изменит в значение false, в идеальном случае здесь всегда стоит true
            //логика для работы с приёмом и передачей
        }

Answer 1

[Александр Кузнецов]

А поле, через которое вы пихаете данные, оно volatile? Может это поможет?

Comments

[Михаил Усоцкий]

Полем на данный момент является cmd для отладки. А так - это Command. Просто я расставил точки останова, чтобы поймать данные. Изначально они без volatile. Раньше как-то работало без этого прекрасно. Только большие данные надо было пинком передать. Потом после вашего упоминания про volatile решил разобраться с этим. Да, volatile действительно может чем-то помочь. Сумел-таки поймать в точках останова. Правда, они немного странно действуют. Придётся проанализировать алгоритм. Сейчас с volatile у меня программа виснет. Просто у меня изначально потоком является Task. Так как он проще и производительней.

[Александр Кузнецов]

Михаил Усоцкий: Я бы посоветовал отдельно реализовать логику получения из вне данных без ее проработки. Чтобы протестировать. Потом, еще можно добавить Thread.Spleep(1) (после #region Стадия чтения.), чтобы обработка была "мягче".

[Михаил Усоцкий]

Я пробовал сделать подобному варианту: https://msdn.microsoft.com/ru-ru/library/x13ttww7.aspx Приложение игнорирует такой подход. Честно говоря, я удивлён, что это перестало работать. Этот же код нормально работал в предыдущем проекте. Даже сейчас используется тот же подход.

[Михаил Усоцкий]

Александр Кузнецов: такой способ работает (данные приходят), но программа зависает намертво. https://itvdn.com/ru/video/how-to-c-sharp-essentia...

[Александр Кузнецов]

Михаил Усоцкий: Вы добавляли Thread.Sleep(1)?
Еще, вы можете к примеру через AutoResetEvent/ManualResetEvent сигнализировать в поток о событии, что данные пошли. Но тут нужно подумать как подавать данные.

Еще вариант, попробовать вот такое:

void Main()
{
	System.Threading.ThreadPool.QueueUserWorkItem(ThreadWork); 
	
	
	for(int i = 0;i<10; i++)
	{
		data.Enqueue(new { data = i });
		Thread.Sleep(1000);
		mre.Set();
	}
	
	isFinish = true;
	mre.Set();
}

ManualResetEvent mre = new ManualResetEvent(false);
Queue<object> data = new Queue<object>();
bool isFinish = false;


void ThreadWork(object state)
{
	while(true)
	{
		mre.WaitOne();
		
		if (isFinish)
			break;
		
		if (data.Count()>0)
		{
			var item = data.Dequeue();
			new Worker(item).Process();
		}
	}
}

class Worker
{
	readonly object _data;

	public Worker(object data)
	{
		this._data = data;
	}
	
	public void Process()
	{
		//Вот тут ваша логика обработки
	}
}

[Александр Кузнецов]

Это простой пример, написанный на коленке в LinqPad-е. void Main() это запуск обработчика и точка входа данных. Как то нужно было это сделать..)

void ThreadWork(object state) это то что вертится в треде и вызывает дальше логику обработки. Думаю, ход мыслей понятен.

[Михаил Усоцкий]

Thread.Sleep(1) добавлял - без толку. Для отладки решил просто тупо отключить чтение и запись и проверить механизм цикла в том методе, где данные приходили. И заметил, что поток продолжает работать, но почему-то становится потоком UI-интерфейса, отсюда и зависания приложения (тот же эффект вызывает при нажатии на кнопку с бесконечным в методе без потока).

Ну у меня нет точки входа данных. Данные располагаются прямо в классе. Данные передаются двунаправленно: ReadData и WriteData, а также дополнительные данные управления потоком. Судя по вашему примеру вы предлагаете мне вынести механизмы чтения и записи в отдельный класс? В общем, подумаю над вашим вариантом.

[Михаил Усоцкий]

Александр Кузнецов: мда... внезапно вернулся в норму... Теперь данные пошли по старому. Правда, пока без чтения и записи. Придётся мне перепроверить свой код.

[Александр Кузнецов]

Михаил Усоцкий: Ну не только это. Я предлагаю сигнализировать обработчику о том, что есть что обработать и да, вынести обработку в отдельный класс. Тогда вы сможете ее запускать асинхронно к примеру.

На мой взгляд, вам нужно сделать сейчас "заглушки" обработки чтобы понять, что вызывало торможения процесса.

[Михаил Усоцкий]

Александр Кузнецов: ваша мысль ясна. Ну у меня изначально является отдельным классом специально для потока. Так как у меня не один поток. Поэтому и взял только Task, который быстрее работает в асинхронном режиме. Но ваше предложение возьму на заметку, если всё-таки пойдёт не так хотелось бы. Просто с помощью заглушки я выяснил, что данные всё-таки передаются, но с интересной спецификой принятия условия. Например, сейчас у меня первым читается условие Commands.HasFlag(Ядро_хххх.BasicStructures.Command.NotOperation), значит, данные Commands будут пропускаться. Но если поставить первым вот такой вариант: Commands.HasFlag(Ядро_хххх.BasicStructures.LoadFPGAAll) - , то работает нормально, данные принимаются сразу. И никаких зависаний. Значит, у меня всё-таки есть ошибка в алгоритме, которую надо исправлять.

Answer 2

[Петр]

А чем ConcurrentQueue не угодил?
Enqueue(T) - добавили, где данные загрузили
TryDequeue(T) - получили в потоке для обработки

Comments

[Михаил Усоцкий]

Всё также, как и предыдущие. В этом случае очередь просто копится. Но в цикле видит как пустой.

[Петр]

Михаил Усоцкий: Упрощенный код покажите то

[Михаил Усоцкий]

Петр: только что показал.

[Рифат]

Используйте ConcurrentQueue в связке с SemaphoreSlim, например. Что то типа такого

while (_started)
{
await _semaphore.WaitAsync();
//---
ITask nextTask;
if (_tasksQueue.TryDequeue(out nextTask))
{
await _taskHandler.HandleAsync(nextTask);
}
}

[Петр]

Альфред: А зачем тут семафор, если все в одном отдельном потоке выполняется? А ConcurrentQueue уже потокобезопасный.

Answer 3

[Роман]

Вы создаёте cmds в потоке и никуда не передаёте? Как тогда там может что-то появиться?

Comments

[Михаил Усоцкий]

cmds - он исключительно внутренний для отладки, когда обнаружил, что данные не передаются в Commands. В реальности как это выглядит. Я нажимаю на кнопку. В методе присваивается некоторое значение Commands. А Commands объявлен в классе, где и находится метод потока. А поток, по идее, уже должен попутно подхватить новый параметр в классе.