Здравствуйте, Cynic, Вы писали:

C>Я конечно извиняюсь, скорее всего я туплю, но как можно использовать например ReaderWriterLockSlim для моей задачи

RWLock вместо полной блокировки нужен при обращении к обычным дотнетовским коллекциям — они поддерживают режим с параллельным чтением без блокировки.

C>Мне нужно, чтобы например три рабочих потока взяли себе по задаче выполнили их и при этом каждый из них дождался пока закончат остальные, потом все снова взяли себе по задаче и т.д.

Типичная для TPL задача. Зачем велосипед городить неясно.
И, кстати, это не задача, а твое решение уже. Было бы неплохо исходную задачу описать, зачем тебе потоки понадобились и именно 3, и почему они должны ждать завершения. Координация то исходно, наверное, не по потокам, а по результатам вычислений, верно?

C> lock() (ну и Monitor) я ещё как то смог прикрутить к этой задаче, а вот про ReaderWriterLockSlim идей чего то нету. Можно примерчик

У тебя же в реальном коде какие то данные появятся, а не просто пустые потоки крутиться будут, верно?

Здравствуйте, nikov, Вы писали:

N>Советую забыть про потоки и учить таски и TPL.

А какую-нибудь толковую книгу (или серию статей) по ним посоветуете?

Здравствуйте, Sinix, Вы писали:

S>Шадулер может быть любой, по умолчанию используется крайне эффективный пул задач. Плюс с тасками и await очень удобно писать код с non-blocking wait. Вот тут

Автор: Sinix
Дата: 13.02.14

был пример, можете попробовать достичь того же без тасков.

В общем попробовал я таски и офигел. Ни каким другим из предложенных способов даже рядом не удавалось достигнуть той-же эффективности. Производительность других решений на 2-3 порядка ниже. Тестировал примерно так:

    class Program
    {
        static readonly int counterValue = 1000000;
        static readonly int taskCount = 30000;

        static void Main(string[] args)
        {
            List<Action> actions = new List<Action>();

            for (int i = 0; i < taskCount; i++)
            {
                actions.Add(new Action(Work));
            }

            Console.WriteLine("Task speed test");

            DateTime start = DateTime.Now;

            Parallel.Invoke(actions.ToArray());

            DateTime stop = DateTime.Now;

            Console.WriteLine("Time is: {0} secs", (stop - start).TotalSeconds);
            Console.ReadLine();
        }

        public static void Work()
        {
            int counter = counterValue;

            while (true)
            {
                counter--;
                if (counter <= 0)
                    return;
            }
        }
    }

Появилось несколько вопросов:

1. Каким образом удалось достигнуть такой производительности?
   
2. Стоит ли надеяться, что можно достигнуть большей производительности другим способом?
   
3. Интересно что если менять параметры counterValue и taskCount увеличивая в 10 раз один и в 10 раз уменьшая другой (или наоборот), то время выполнения задач практически не меняется. На моей машине всегда примерно 9 секунд. Складывается впечатление, что создается ограниченное количество потоков которым потом по очереди скармливаются задачи. В общем как это работает?

Здравствуйте, breee breee, Вы писали:

N>>Советую забыть про потоки и учить таски и TPL.
BB>А какую-нибудь толковую книгу (или серию статей) по ним посоветуете?

Присоединяюсь к вопросу.

Материалов мало, полтора года назад всё ограничивалось вот этим

Автор: Sinix
Дата: 17.09.13

. Из недавнего — только Concurrency in C# Cookbook попадалась. Так себе.

Здравствуйте, AndrewVK, Вы писали:

Как бы ноги растут отсюда — http://rsdn.ru/forum/design/5979819.1

Автор: Cynic
Дата: 12.03.15

Здравствуйте, Cynic, Вы писали:

C>По поводу TPL: Правильно ли я понимаю, что TPL реализует паттерн поставщик-потребитель, т.е. создаёт один или несколько рабочих потоков и скармливает им задачи.

Дефолтный шедулер внутри — примерно так, только очень грубо. Но это не должно тебя особо волновать. Тебе нужно разделить свою задачу на кусочки и продекларировать между ними зависимости. С остальным TPL разберется сам.

C> При это сами потоки не создаются и уничтожаются каждый раз для новой задачи?

Дефолтный шедулер берет потоки из стандартного пула. Не переживай за такие вещи — TPL в плане перформанса оттюнен очень качественно.

Здравствуйте, Cynic, Вы писали:

C>В общем попробовал я таски и офигел. Ни каким другим из предложенных способов даже рядом не удавалось достигнуть той-же эффективности. Производительность других решений на 2-3 порядка ниже.

2-3 порядка, это потому что вместо спинлоков ты примитивы ОС используешь, а тела методов у тебя игрушечные. Кстати, вместо рукопашных циклом можно использовать Thread.SpinWait().

C>

Каким образом удалось достигнуть такой производительности?

В данном случае спинлоки прежде всего.

C>

Стоит ли надеяться, что можно достигнуть большей производительности другим способом?

Вопрос непонятный.

... << RSDN@Home 1.0.0 alpha 5 rev. 0 on Windows 8 6.2.9200.0>>

Здравствуйте, Sinix, Вы писали:

S>Из недавнего — только Concurrency in C# Cookbook попадалась. Так себе.

Так есть же православный аналог — https://www.packtpub.com/application-development/multithreading-c-50-cookbook

Здравствуйте, Cynic, Вы писали:

C>Каким образом удалось достигнуть такой производительности?
Переписыванием ThreadPool (шедулер по умолчанию работает поверх него), затем — перепиливанием внутренностей самих тасков (второе помогло куда больше, но и поломало пару мелочей).

Если коротко, то
1. Задачи не требуют переключения контекста.
2. Используются неблокирующие ожидания (например, на IOCP). Если задача ждёт чего-то, то она не блокирует поток. Пул может запустить следующую задачу — время CPU используется гораздо эффективней.
3. Начиная с 4.5 таски здорово оптимизированы под код в CPS-style (читай, с использованием await).

Если интересует реализация с точки зрения архитектуры — см раздел Decentralized Scheduling Techniques

Разумеется, таски НЕ улучшат производительность, если весь ваш код утыкается в любое другое горлышко: IO, CPU, сеть, свои взаимоблокировки и тд и тп. Кроме того, добавление долгих задач может испортить общую картину из-за исчерпания пула (чинится указанием TaskCreationOption.LongRunning). Потому что

aParllel prgoamrmnig s i ahdr

(с) Gastón C. Hillar

Но с самым популярным — блокировками на ожидание и неправильным использованием потоков и примитивов синхронизации — таски борятся довольно эффективно.


C>Стоит ли надеяться, что можно достигнуть большей производительности другим способом?
Разумеется. Делаете свой пул нативных потоков, поверх него изобретаете зелёные потоки или таски, добавляете load balancing + work stealing и в довершение переводите всю эту радость на lock-free, чтобы избежать лишних блокировок

Вполне возможно, есть чуть более быстрые или более удобные (для какой-то частной задачи) реализации того же самого, но общая идея не меняется.

Здравствуйте, AndrewVK, Вы писали:

AVK> Кстати, вместо рукопашных циклом можно использовать Thread.SpinWait().

Не понимаю где мене тут это может помочь.

C>>

Стоит ли надеяться, что можно достигнуть большей производительности другим способом?

AVK>Вопрос непонятный.

Можно ли добиться большей производительности другим способом? Или выигрыш будет не значительным и не стоит заморачиваться?

Здравствуйте, Cynic, Вы писали:

AVK>> Кстати, вместо рукопашных циклом можно использовать Thread.SpinWait().
C>Не понимаю где мене тут это может помочь.

int counter = counterValue;
while (true)
{
        counter--;
        if (counter <= 0)
                return;
}

Вместо этого можно просто Thread.SpinWait(counterValue).

C>>>

Стоит ли надеяться, что можно достигнуть большей производительности другим способом?
AVK>>Вопрос непонятный.
C>Можно ли добиться большей производительности другим способом? Или выигрыш будет не значительным и не стоит заморачиваться?

Не стоит заморачиваться.

... << RSDN@Home 1.0.0 alpha 5 rev. 0 on Windows 8 6.2.9200.0>>

Здравствуйте, Sinix, Вы писали:

S>Присоединяюсь к вопросу.

S>Материалов мало, полтора года назад всё ограничивалось вот этим

Автор: Sinix
Дата: 17.09.13

. Из недавнего — только Concurrency in C# Cookbook попадалась. Так себе.

Вот хороший блог -- http://blogs.msdn.com/b/pfxteam/

Здравствуйте, Sharov, Вы писали:

S>Вот хороший блог -- http://blogs.msdn.com/b/pfxteam/
Тынц по ссылке "вот этим

Автор: Sinix
Дата: 17.09.13

" из предыдущего поста :P

Здравствуйте

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

  [TestClass]
  public class ParallelRunUnitTest {
    public TestContext TestContext { get; set; }

    const int _actionsCount = 100;
    const int _threadsCount = 3;
    int _doneCount;
    readonly Queue<Action> _actions = new Queue<Action>(_actionsCount);
    bool _active;

    [TestMethod]
    public void UsingThreadsAndMonitorTest() {
      _doneCount = 0;
      _active = true;
      var threadList = new List<Thread>(_threadsCount);
      for (int i = 0; i < _threadsCount; i++) {
        var thread = new Thread(ThreadMethod) { IsBackground = true };
        threadList.Add(thread);
        thread.Start();
      }

      var sw = Stopwatch.StartNew();

      lock (_actions)
        for (int i = _actionsCount; i > 0;) {
          Monitor.Wait(_actions, 5);
          if (_actionsCount - i == _doneCount) {
            for (int k = Math.Min(_threadsCount, i); k > 0; k--, i--)
              _actions.Enqueue(TaskMethod);
            //Debug.WriteLine("    Add new tasks. Count: " + _actions.Count);
          }
          Monitor.Pulse(_actions);
        }

      // Подождать окончания расчета
      lock (_actions)
        while (_doneCount < _actionsCount)
          Monitor.Wait(_actions, 5);

      sw.Stop();

      // Подождать остановки потоков
      _active = false;
      foreach (var thread in threadList)
        thread.Join();

      Debug.WriteLine("  RunCount: {0}   Elapsed: {1}", _doneCount, sw.Elapsed);
    }

    void ThreadMethod() {
      while (_active) {
        Action action;
        lock (_actions) {
          while (_actions.Count == 0)
            if (_active)
              Monitor.Wait(_actions, 200);
            else
              return;
          action = _actions.Dequeue();
          Monitor.Pulse(_actions);
        }
        action();
      }
    }

    void TaskMethod() {
      const int count = 500000;
      for (int j = 0; j < 2; j++) 
      for (int i = 0; i < count; i++) {
        var a = Math.Sin(2*Math.PI*i/count);
      }
      Interlocked.Increment(ref _doneCount);
      //Debug.WriteLine("    Task done");
    }

    [TestMethod]
    public void UsingTasksTest() {
      _doneCount = 0;
      var tasks = new List<Task>(_threadsCount);
      var sw = Stopwatch.StartNew();

      for (int i = _actionsCount; i > 0;) {
        for (int k = Math.Min(_threadsCount, i); k > 0; k--, i--) {
          var task = new Task(TaskMethod);
          tasks.Add( task );
          task.Start();
        }
        //Debug.WriteLine("    Add new tasks. Count: " + tasks.Count);
        Task.WaitAll(tasks.ToArray());
        tasks.Clear();
      }

      sw.Stop();
      Debug.WriteLine("  RunCount: {0}   Elapsed: {1}", _doneCount, sw.Elapsed);
    }
  }

Время выполнения тестов:
UsingThreadsAndMonitorTest() : 3,23 секунды
UsingTasksTest() : 3,17 секунды

Здравствуйте, G_a_r_r_y, Вы писали:

G__>Ничего особенного, но может кому-то будет интересно.

Не нравится мне эта синхронизация из костылей и спичек (аля "while (_active)" и "_actionsCount — i == _doneCount"). Одна опечатка и упс. Собственно, она у вас уже есть, поля надо бы volatile пометить

Вариант с тасками гораздо проще. Только Task.WaitAll() лучше заменить на await, ждущая блокировка тут не нужна. И "new Task() ... Start()" — на "Task.Run()";
Более-менее правильный пример выше

Автор: Sinix
Дата: 13.03.15

.

Здравствуйте, Sinix, Вы писали:

S>Не нравится мне эта синхронизация из костылей и спичек (аля "while (_active)" и "_actionsCount — i == _doneCount"). Одна опечатка и упс.
Никаких костылей и спичек, хотя выглядит сложновато, но можно и упростить, сразу не сообразил.

S>Вариант с тасками гораздо проще. Только Task.WaitAll() лучше заменить на await, ждущая блокировка тут не нужна. И "new Task() ... Start()" — на "Task.Run()";
У меня пример для 4 фреймворка (посыпаю голову пеплом, забыл предупредить), там этих "красивостей" еще нет.
А ждущая блокировка нужна по условиям задачи.

S>Более-менее правильный пример выше

Автор: Sinix
Дата: 13.03.15

.
Мой пример лучше.

Здравствуйте, G_a_r_r_y, Вы писали:

G__>У меня пример для 4 фреймворка (посыпаю голову пеплом, забыл предупредить), там этих "красивостей" еще нет.

Там есть Task.WhenAll, красивости ничего принципиально не меняют.

Здравствуйте, AndrewVK, Вы писали:

AVK>Там есть Task.WhenAll, красивости ничего принципиально не меняют.
Одно из исходных требований: "Управляющий поток должен дожидаться пока все вычислительные потоки не встанут на паузу и только после этого раздавать им задачи". Поэтому в данном примере нельзя использовать Task.WhenAll, а нужна именно блокировка Task.WaitAll управляющего потока.

Здравствуйте, G_a_r_r_y, Вы писали:

AVK>>Там есть Task.WhenAll, красивости ничего принципиально не меняют.
G__>Одно из исходных требований: "Управляющий поток должен дожидаться пока все вычислительные потоки не встанут на паузу и только после этого раздавать им задачи". Поэтому в данном примере нельзя использовать Task.WhenAll

Можно. Потому что там именно логически нужно дождаться, для симуляции, а не физически поток CPU лочить.

Здравствуйте, Sinix, Вы писали:
S> ...

Попробовал сделать как у Вас:

    interface IWorker
    {
        Task GetWorker();
    }

    class Worker: IWorker
    {
        readonly string _name;
        readonly int _delay;

        public Worker(string name, int delay)
        {
            _name = name;
            _delay = delay;

            TimeEngine.Add(this);
        }

        public Task GetWorker()
        {
            Task worker = Task.Factory.StartNew(() =>
            {
                Console.WriteLine("Worker {0} start", _name);

                Task.Delay(_delay);

                Console.WriteLine("Worker {0} end", _name);
            });

            return worker;
        }
    }

    class TimeEngine
    {
        static List<IWorker> _workers = new List<IWorker>();
        static CancellationTokenSource _cts = new CancellationTokenSource();

        public static void Add(IWorker worker)
        {
            _workers.Add(worker);
        }

        private static Task Print(int x)
        {
            Console.WriteLine(x);
            return new Task(() => { });
        }

        private static async Task Worker(CancellationToken ct)
        {
            while (!ct.IsCancellationRequested)
            {
                var tasks = Enumerable.Range(0, _workers.Count).Select(x => _workers[x].GetWorker());
                await Task.WhenAll(tasks);
                await Task.Delay(TimeSpan.FromSeconds(1));
            }
        }

        public static void Start()
        {
            Worker(_cts.Token);
        }

        public static void Stop()
        {
            _cts.Cancel();
        }
    }

    class Program
    {
        static void Main(string[] args)
        {
            Worker worker1 = new Worker("1", 1000);
            Worker worker2 = new Worker("2", 500);
            Worker worker3 = new Worker("3", 1200);

            Console.WriteLine("TimeEngine start (Press any key to Stop)");
            TimeEngine.Start();

            Console.ReadLine();

            TimeEngine.Stop();
            Console.WriteLine("Time Engine Stop");

            Console.ReadLine();
        }
    }

Посмотрите плз. ни где не накосячил