Если интересно, прошу под кат.
В данный момент синхронизация потоков в c# вызывает некоторые затруднения, в частности передача примитивов синхронизации между объектами Вашего приложения и поддержка этого всего в дальнейшем.
Текущая модель с Task и IAsyncResult а так же TPL в целом решают все проблемы при должном проектировании но хотелось создать простой класс через который можно будет отправлять и принимать сигналы с блокировкой потока.
В общем в голове созрел некий интерфейс:
public interface ISignal<T> : IDisposable
{
void Send(T signal);
T Receive();
T Receive(int timeOut);
}
, где T — сущность которую необходимо передать получателю.
Пример вызова:
[TestMethod]
public void ExampleTest()
{
var signal = SignalFactory.GetInstanse<string>();
var task1 = Task.Factory.StartNew(() => // старт потока
{
Thread.Sleep(1000);
signal.Send("Some message");
});
// блокировка текущего потока
string message = signal.Receive();
Debug.WriteLine(message);
}
Для получения объекта сигнала создадим фабрику.
public static class SignalFactory
{
public static ISignal<T> GetInstanse<T>()
{
return new Signal<T>();
}
public static ISignal<T> GetInstanse<T>(string name)
{
return new CrossProcessSignal<T>(name);
}
}
Signal — internal класс для синхронизации внутри одного процесса. Для синхронизации необходима ссылка на объект.
CrossProcessSignal — internal класс который может синхронизировать потоки в отдельных процессах(но об этом чуть позже).
Теперь о реализации Signal
Первое, что приходит на ум, в Receive блокировать выполнение потока с помощью Semaphore а в методе Send вызывать Release() этого семафора с количеством блокированных потоков.
После разблокировки потоков возвращать результат из поля класса T buffer. Но мы не знаем какое количество потоков будет висеть в Receive и нет гарантии что к вызову Release не подбежит еще пара потоков.
В качестве примитива синхронизации был выбран AutoResetEvent. Для каждого нового потока будет создаваться свой AutoResetEvent, хранить все это добро мы будем в словаре Dictionary<int,AutoResetEvent> где ключ это id потока.
Собственно поля класса выглядят так:
private T buffer;
Dictionary<int,AutoResetEvent> events = new Dictionary<int, AutoResetEvent>();
private volatile object sync = new object();
private bool isDisposabled = false;
Объект sync будет нам необходим при вызове Send, дабы несколько потоков не начали перетирать буфер.
isDisposabled флаг указывающий был ли вызван Dispose(), если не вызван то вызываем его в деструкторе.
public void Dispose()
{
foreach(var resetEvent in events.Values)
{
resetEvent.Dispose();
}
isDisposabled = true;
}
~Signal()
{
if (!isDisposabled)
{
Dispose();
}
}
Теперь о методе Receive.
public T Receive()
{
var waiter = GetEvents();
waiter.WaitOne();
waiter.Reset();
return buffer;
}
GetEvents() достает из словаря AutoResetEvent если есть, если нет то создает новый и кладет его в словарь.
waiter.WaitOne() блокировка потока до ожидания сигнала.
waiter.Reset() сброс текущего состояния AutoResetEvent. Следующий вызов WaitOne приведет к блокировке потока.
Осталось только вызвать метод Set для каждого AutoResetEvent.
public void Send(T signal)
{
lock (sync)
{
buffer = signal;
foreach(var autoResetEvent in events.Values)
{
autoResetEvent.Set();
}
}
}
Проверить данную модель можно тестом:
private void SendTest(string name = "")
{
ISignal<string> signal;
if (string.IsNullOrEmpty(name))
{
signal = SignalFactory.GetInstanse<string>(); // создаем локальный сигнал
}
else
{
signal = SignalFactory.GetInstanse<string>(name);
}
var task1 = Task.Factory.StartNew(() => // старт потока
{
for (int i = 0; i < 10; i++)
{
// блокировка потока, ожидание сигнала
var message = signal.Receive();
Debug.WriteLine($"Thread 1 {message}");
}
});
var task2 = Task.Factory.StartNew(() => // старт потока
{
for (int i = 0; i < 10; i++)
{
// блокировка потока, ожидание сигнала
var message = signal.Receive();
Debug.WriteLine($"Thread 2 {message}");
}
});
for (int i = 0; i < 10; i++)
{
// отправка сигнала ожидающим потокам.
signal.Send($"Ping {i}");
Thread.Sleep(50);
}
}
using System.Collections.Generic;
using System.Threading;
namespace Signal
{
internal class Signal<T> : ISignal<T>
{
private T buffer;
Dictionary<int,AutoResetEvent> events = new Dictionary<int, AutoResetEvent>();
private volatile object sync = new object();
private bool isDisposabled = false;
~Signal()
{
if (!isDisposabled)
{
Dispose();
}
}
public T Receive()
{
var waiter = GetEvents();
waiter.WaitOne();
waiter.Reset();
return buffer;
}
public T Receive(int timeOut)
{
var waiter = GetEvents();
waiter.WaitOne(timeOut);
waiter.Reset();
return buffer;
}
public void Send(T signal)
{
lock (sync)
{
buffer = signal;
foreach(var autoResetEvent in events.Values)
{
autoResetEvent.Set();
}
}
}
private AutoResetEvent GetEvents()
{
var threadId = Thread.CurrentThread.ManagedThreadId;
AutoResetEvent autoResetEvent;
if (!events.ContainsKey(threadId))
{
autoResetEvent = new AutoResetEvent(false);
events.Add(threadId, autoResetEvent);
}
else
{
autoResetEvent = events[threadId];
}
return autoResetEvent;
}
public void Dispose()
{
foreach(var resetEvent in events.Values)
{
resetEvent.Dispose();
}
isDisposabled = true;
}
}
}
Данной реализации есть куда расти в плане надежности. В исходниках есть межпроцессорная реализация этой идеи с передачей сигнала через shared memory, если будет интересно могу написать об этом отдельную статью.
Исходники на Гитхабе
Комментарии (42)
horse315
30.08.2017 13:51+1Я так понял, что задача уведомить все ожидающие потоки, которые запросили AutoResetEvent из пула? Просто не вижу причин строить все это на set-reset event'ах, когда есть Monitor, который работает гораздо быстрее.
Какого поведения ожидаете в случае, если источник эмитит собщение до того, как поток-потребитель обработал предыдущее? И еще вижу в коде привязку к экземпляра AutoResetEvent к ThreadId, это опасно, она как-то используется?Kalatyn11 Автор
30.08.2017 14:25AutoResetEvent привязан к ThreadId в
Dictionary<int,AutoResetEvent> events = new Dictionary<int, AutoResetEvent>();horse315
30.08.2017 14:28+1А зачем?
Kalatyn11 Автор
30.08.2017 14:59Для идентификации потока, Вы не можете идентифицировать кто пришел в метод Receive кроме как по ThreadId.
Какого поведения ожидаете в случае, если источник эмитит собщение до того, как поток-потребитель обработал предыдущее?
Это проблемы потока потребителя. Можно в дальнейшем реализовать очередь если необходимо, но если между методами Receive() прошло какое то сообщение то поток соответственно его пропустит.
Использования Monitor вполне интересно.horse315
30.08.2017 15:32Один await в потребителе в консольном приложении уже не гарантирует, что вы получите тот же Thread Id.
horse315
30.08.2017 15:47+1Могу посоветовать книгу «Concurrency in C# Coockbook» by Stephen Cleary. Есть на известном сайте (только что проверил). У меня издание 2014 и все перечисленное там до сих пор актуально.
Gentlee
30.08.2017 14:25Поражаюсь, насколько много программистов C# не понимают даже базовых вещей работы с потоками. Блокировка потоков это зло, нужно этого всячески избегать. «через который можно будет отправлять и принимать сигналы с блокировкой потока» — не вижу ни одной причины, когда бы понадобилась блокировка потока в таком случае, не подскажете для чего писать такие костыли?
Kalatyn11 Автор
30.08.2017 14:29Если Вам нужно выполнить какую то задачу то действительно лучше использовать Task. Если Вы ждете выполнения какого то события и обработку не желательно осуществлять в том потоке которое спровоцировало событие то сигналы это вариант.
lair
30.08.2017 14:33+2Если Вы ждете выполнения какого то события и обработку не желательно осуществлять в том потоке которое спровоцировало событие то сигналы это вариант.
Нет, "вариант" — это не сигналы, а асинхронные события в любой их ипостаси.
d3nver777
30.08.2017 15:311) Для варианта wpf/mvvm — Messenger. Входит в состав большинства mvvm библиотек.
2) Вообще для всех вариантов — Observable и ReactiveExtension. www.introtorx.com/content/v1.0.10621.0/00_Foreword.html
Основная бизнес логика может выполнять в отдельном потоке. Уведомления от текущем состоянии, о завершении — могут выполняться в UI-потоке.
Пример:
GetChildren() .ToObservable() .Select(ReCreateElement) .SubscribeOn(NewThreadScheduler.Default) .ObserveOn(SynchronizationContext.Current) .Subscribe(shaft => { }, OnCompleted, _tokenSource.Token);
TIgorA
30.08.2017 15:50ConcurrentQueue принимает объекты на обработку, один поток крутит внутри себя
while (TryTake(T item)) и обрабатывает.
Angelina_Joulie
30.08.2017 16:53+1Муть.
Вы изобрели Enterprise Service Bus только в рамках одного процесса.
А так же на практике, не без изъянов, реализовали паттерн Producer-Consumer.
Посмотрите в сторону TPL (DataFlow Blocks)
Я всем своим коллегам говорю — вы прикладные программисты, и всё что вы изобрели сегодня, до вас уже изобрели.
Kalatyn11 Автор
30.08.2017 17:24DataFlow Blocks очень годная штука.
В частности если взять BufferBlock при вызове Receive только один поток получит сигнал который отправили в Post. Пример:
var bufferBlock = new BufferBlock<int>(); Task.Factory.StartNew(() => { for (int i = 0; i < 3; i++) { Thread.Sleep(3000); bufferBlock.Post(i); } }); Task.Factory.StartNew(() => { for (int i = 0; i < 3; i++) { Console.WriteLine($"{bufferBlock.Receive()} threadId={Thread.CurrentThread.ManagedThreadId}"); } }); for (int i = 0; i < 3; i++) { Console.WriteLine($"{bufferBlock.Receive()} threadId={Thread.CurrentThread.ManagedThreadId}"); }
Основная моя идея это не создание очереди сообщений а синхронизация потоков в том числе в разных процессах.lair
30.08.2017 17:38+1Вопрос как раз в том, зачем вам нужна синхронизация потоков (тем более в разных процессах), когда "современный тренд" — это как раз максимально избегать синхронизации.
Kalatyn11 Автор
30.08.2017 18:18В статье есть мысль.
Текущая модель с Task и IAsyncResult а так же TPL в целом решают все проблемы при должном проектировании
Я не утверждаю что область применения такого подхода широка, нет она даже очень специфична.
В качестве примера можно привести asp.net приложение где может быть несколько процессов (сколько выставили в IIS), в каждом процессе несколько потоков.
Вот к Вам пришел клиент и Вам нужно сделать дорогой запрос к базе, как синхронизировать потоки так что бы был только один запрос?lair
30.08.2017 18:24Если бы у меня была такая задача, у меня был бы отдельный сервис (единственный, конечно же) отвечающий за такие дорогие запросы, и в нем была бы принудительная сериализация.
(я, правда, не вижу практического смысла делать несколько процессов для одного приложения asp.net, но не суть)
IL_Agent
30.08.2017 20:03+2Зачем ваш ISignal, когда есть TaskCompletionSource?
Kalatyn11 Автор
30.08.2017 21:21TaskCompletionSource про результат выполнения задачи
ISignal про получение сигналов о событиях в другом потоке.
Что бы было наглядно вот тест:
[TestMethod] public void PingPong() { var signal = SignalFactory.GetInstanse<string>(); var task = Task.Factory.StartNew(() => { for(int i = 0;i<10;i++) { Debug.WriteLine(signal.Receive()); Thread.Sleep(30); signal.Send("pong"); } }); Task.Factory.StartNew(() => { Thread.Sleep(10); for (int i = 0; i < 10; i++) { signal.Send("ping"); Debug.WriteLine(signal.Receive()); Thread.Sleep(30); } }); task.Wait(); }
Результping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
ping
pong
pinglair
30.08.2017 21:33Ну или
ISubject<string>, если вам вот прямо надо, чтобы оно было двухсторонним, хотя я бы делил на observable и observer, это упрощает зависимости.
IL_Agent
30.08.2017 22:57Тогда, как указал коллега выше, это IObservable и rx.
Kalatyn11 Автор
31.08.2017 00:22Не очень понимаю причем тут IObservable и IObserver ведь это интерфейсы по сути представляют callBack.
lair
31.08.2017 01:03-1В том-то и дело, что они предоставляют не колбэки, а поток событий, над которым определены полезные операции. В частности, на этом потоке можно выбрать первый элемент (в будущем!), сконвертировать этот элемент в таск, и сделать тому await или Wait — вот и ваш Receive, только, на выбор потребителя, блокирующий или асинхронный.
Kalatyn11 Автор
01.09.2017 15:47Смысл не в том что бы вернуть задачу, а заблокировать выполнение текущего потока до получения сигнала.
Но Вы подкинули пищу как можно оптимизировать данный механизм.
Создать одну задачу внутри класса с блокировкой, остальные потоки в Receive обращаются к Result этой задачи.
Как итог:
- Нужно блокировать только один поток(внутренний таск)
- У обычных потоков будет такая же блокировка как и сейчас
- Поток из пула освободится для выполнения других задач.
Нужно только прикинуть что будет дороже, примитивы синхронизации или вся эта пляска с тасками.lair
01.09.2017 16:08Смысл не в том что бы вернуть задачу, а заблокировать выполнение текущего потока до получения сигнала.
Говорят же вам: блокировать потоки — плохо. Но, что важнее, таск легко позволяет заблокировать поток до своего выполнения, причем с конфигурируемым таймаутом и кооперативной отменой.
Создать одну задачу внутри класса с блокировкой, остальные потоки в Receive обращаются к Result этой задачи.
Как итог:
- Нужно блокировать только один поток(внутренний таск)
- У обычных потоков будет такая же блокировка как и сейчас
- Поток из пула освободится для выполнения других задач.
Вообще-то, ничего не изменится. Просто раньше у вас потоки блокировались на
WaitOne, а станут блокироваться наResult.
Пока вы таск не вытащите в публичный интерфейс, ничего не изменится.
Kalatyn11 Автор
01.09.2017 16:38Вообще-то, ничего не изменится. Просто раньше у вас потоки блокировались на WaitOne, а станут блокироваться на Result.
Соль то реализации в блокировке исполнения до получения сигнала.
Вообще-то, ничего не изменится. Просто раньше у вас потоки блокировались на WaitOne, а станут блокироваться на Result
По поводу блокировке на Result, немного расскажу в чем Task отличается от обычного потока на примере.
Пример 1:
ThreadPool.SetMaxThreads(1000, 1000); var task1 = Task.Factory.StartNew(() => { Console.WriteLine($"Task ThreadId {Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(10000); return -1; }); int mainThreadId = Thread.CurrentThread.ManagedThreadId; Console.WriteLine($"mainThreadId {mainThreadId}"); var task2 = Task.Factory.StartNew(() => { for (int i = 0; i < 1000; i++) { Task.Factory.StartNew(() => { if (Thread.CurrentThread.ManagedThreadId == mainThreadId) { Console.WriteLine($"other thread with id {Thread.CurrentThread.ManagedThreadId }"); } Console.WriteLine($"thread create {Thread.CurrentThread.ManagedThreadId }"); Thread.Sleep(2000); }); } }); int result = task1.Result; Console.ReadKey();
Пример 2:
ThreadPool.SetMaxThreads(1000, 1000); Task.Factory.StartNew(() => { var task1 = Task.Factory.StartNew(() => { Console.WriteLine($"Task ThreadId {Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(10000); return -1; }); int mainThreadId = Thread.CurrentThread.ManagedThreadId; Console.WriteLine($"mainThreadId {mainThreadId}"); var task2 = Task.Factory.StartNew(() => { for (int i = 0; i < 1000; i++) { Task.Factory.StartNew(() => { if (Thread.CurrentThread.ManagedThreadId == mainThreadId) { Console.WriteLine($"other thread with id {Thread.CurrentThread.ManagedThreadId }"); } Console.WriteLine($"thread create {Thread.CurrentThread.ManagedThreadId }"); Thread.Sleep(2000); }); } }); int result = task1.Result; Console.ReadKey(); });
Второй пример отличается от первого только в том что обернут Task.Factory.StartNew.
То есть вся логика второго примера будет в потоке из пула потоков, первого примера в обычной потоке.
При выполнении первого примера Вы на строке int result = task1.Result; заблокируете поток, во втором случае Вы его не заблокируете а просто переназначите его на другую задачу.
В первом случае Вы не увидете строку other thread with id никогда, во втором же случае Вы увидите её несколько раз так как поток исполнения переназначен на другие задачи а продолжение после int result = task1.Result; по сути является callBackом.
Является ли текущий поток Taskом или обычным потоком можно узнать свойством
Task.CurrentIdlair
01.09.2017 16:50Соль то реализации в блокировке исполнения до получения сигнала.
Практическое отличие-то в чем?
При выполнении первого примера Вы на строке int result = task1.Result; заблокируете поток, во втором случае Вы его не заблокируете а просто переназначите его на другую задачу.
Resultвсегда блокирует поток, в котором находится (т.е. после него вы всегда окажетесь в том же потоке). Другое дело, что диспетчер — пользуясь своей собственной внутренней логикой, никак вам не подконтрольной — может решить выполнять таск синхронно (т.е., в том же потоке), или же асинхронно (в другом потоке):
If the current task has not started execution, the
Wait[Resultвызывает Wait] method attempts to remove the task from the scheduler and execute it inline on the current thread.Но это валидно тогда и только тогда, когда таск еще не начал выполнение — и когда его можно перенести в текущий тред (что верно не для всех тасков).
Проще говоря, если у вас есть один таск, который где-то как-то уже выполняется, и десять потоков, в которые вы этот таск передали, то если вы в этих потоках скажете
task.Result(илиtask.Wait()), то вы получите десять заблокированных потоков (и это не считая того места, где таск выполняется). И это будет как раз ваш сценарий — все потоки, в которых будет вызванReceive, будут заблокированы.
А вот если вы скажете
await task, то система вполне может эти потоки переиспользовать для чего-нибудь полезного (а continuation вызвать там, где сочтет нужным).Kalatyn11 Автор
01.09.2017 17:32Вызовите пожалуйста
примерThreadPool.SetMaxThreads(1000, 1000); Task.Factory.StartNew(() => { var task1 = Task.Factory.StartNew(() => { Console.WriteLine($"Task ThreadId {Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(10000); return -1; }); int mainThreadId = Thread.CurrentThread.ManagedThreadId; Console.WriteLine($"mainThreadId {mainThreadId}"); var task2 = Task.Factory.StartNew(() => { for (int i = 0; i < 1000; i++) { Task.Factory.StartNew(() => { if (Thread.CurrentThread.ManagedThreadId == mainThreadId) { Console.WriteLine($"other thread with id {Thread.CurrentThread.ManagedThreadId }"); } Console.WriteLine($"thread create {Thread.CurrentThread.ManagedThreadId }"); Thread.Sleep(2000); }); } }); int result = task1.Result; }).Wait(); Console.ReadKey();lair
01.09.2017 17:50Цитирую еще раз:
If the current task has not started execution, the Wait method attempts to remove the task from the scheduler and execute it inline on the current thread.
Простой способ проверить, что это не обязательно так:
Console.WriteLine($"Main ThreadId {Thread.CurrentThread.ManagedThreadId}"); Task.Factory.StartNew(() => { Console.WriteLine($"Processing ThreadId {Thread.CurrentThread.ManagedThreadId}"); var task = Task.Factory.StartNew(() => { Console.WriteLine($"Task ThreadId {Thread.CurrentThread.ManagedThreadId}"); Thread.Sleep(1000); Console.WriteLine("Finishing task"); }); var cts1 = new TaskCompletionSource<int>(); var cts2 = new TaskCompletionSource<int>(); ThreadPool.QueueUserWorkItem(_ => { Console.WriteLine($"Waiting for task on ThreadId {Thread.CurrentThread.ManagedThreadId}"); task.Wait(); Console.WriteLine($"Finished waiting for task on ThreadId {Thread.CurrentThread.ManagedThreadId}"); cts1.SetResult(0); } ); ThreadPool.QueueUserWorkItem(_ => { Console.WriteLine($"Waiting for task on ThreadId {Thread.CurrentThread.ManagedThreadId}"); task.Wait(); Console.WriteLine($"Finished waiting for task on ThreadId {Thread.CurrentThread.ManagedThreadId}"); cts2.SetResult(0); } ); Task.WaitAll(cts1.Task, cts2.Task); }).Wait();
lair
01.09.2017 18:23… эээ, а где вы ожидаете результатов
task2? Иными словами, откуда вы знаете, что таск не получает свой десятый поток после того, как закончилось ожиданиеtask1.Result, и, как следствие, внешнийWait?
kretuk
31.08.2017 14:27~Signal()
{
if (!isDisposabled)
{
Dispose();
}
}
вроде известная вещь — декструктор/финализатор должен реализовывать только класс, который сам аллоцирует/выделает неуправляемые ресурсы. нет?
lair
А можно увидеть реальный сценарий, когда это нужно?
Kalatyn11 Автор
На ум приходит централизованная обработка каких то событий в приложении простой отправкой сигнала.Синхронизировать потоки так же вполне удобно.
Что же касается межпроцессорных сигналов то можно применить такой сценарий -> запускаем процесс который что то делает -> ждем от него сигнал -> обрабатываем.
lair
А чем обычный event broker не подходит?
Ну так акторы же, с location transparency. Или очереди. Или вообще обычные сервисы.
Блокировать потоки — плохо, потоки надо отпускать, а потом поднимать по приходу "сигнала".
lair
… опять-таки ж, при межпроцессном взаимодействии немедленно возникают всякие вопросы про то "а какой там код с той стороны", что тоже намекает нам, что лучше использовать типовые сервисные решения.
gimbarr_dpr
на ум приходит обработка сетевых запросов в многопоточных приложениях. допустим, несколько потоков выполняют запрос к сети, а другие ожидают (через какой-нибудь SemaphoreSlim), пока какой-нибудь httpClient освободится от уже занятых потоков. И вот если в потоке произойдёт ошибка авторизации, то можно остановить доступ к httpClien'у, пока не произойдёт перелогин, чтобы у остальных не посыпались исключения / не открылось множество окон авторизации.
lair
… и как это сделать с помощью блокирующего
Receive?(особенно учитывая, что HttpClient — он весь на тасках)
lair
Я хочу заметить, что ваша задача в общем случае решения не имеет, потому что если у нас есть больше одного параллельно работающего http-клиента, то нет никакой гарантии, что в момент получения одним из них ошибки авторизации сколько-то других не находятся в in-flight — останавливать к ним доступ уже поздно, а ошибку авторизации они все равно вернут. Поэтому все равно придется писать код, который обрабатывает ошибки авторизации, а тогда можно уже и не делать блокировку доступа.