6

Как известно, TaskScheduler в TPL раскидывает таски по ядрам (хоть и не гарантирует это).

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

Вопрос в том, если этот поток переделать на task-модель, то будут ли эти таски перемалываться на разных ядрах или же будут тяготеть к одному и тому же ядру? Хоть это и таски, но по факту один поток разбивается на таски, чтобы избежать sleep и TaskScheduler может тяготеть переиспользовать этот же поток (других то в пуле нет)

9
  • По идее, функция, в которой происходит await Task.Delay, вообще не принадлежит никакому потоку. Так что возобновится она в случайном потоке (если вы скажете ConfigureAwait(false)).
    – VladD
    16 мар 2016 в 22:52
  • @VladD только вот случайных потоков нет. На таски ведь разбивается однопоточный код и пул потоков не будет рождать новых потоков без надобности, а ее не возникнет. Поэтому полагаю, что шедулер будет все время выдавать один и тот же поток из пула...будет ли он рождать новые потоки если увидит, что какое то ядро более свободно?- вот этого я не знаю
    – vitidev
    16 мар 2016 в 22:56
  • В моих тестах потоки из thread pool'а тасовались, возвращение шло каждый раз к разному. Но это было на какой-то конкретной версии .NET, у вас может быть и по-другому. Кстати, почему бы не свести все эти процессы в один?
    – VladD
    16 мар 2016 в 23:01
  • @VladD Особенности решения. Разбития на процессы не избежать. Возвращение идет обычно к доступному потоку. Более того в Parallel.For основной поток, который вызвал этот метод переиспользуется внутри метода, а не засыпает. В моей же задаче всегда будет доступен один поток - основной (((. Получается что в системе куча ядер, а выполняться будет на том, на котором процесс запустился невзирая на загруженность других ядер.
    – vitidev
    16 мар 2016 в 23:07
  • Момент, сейчас проэкспериментируем :)
    – VladD
    16 мар 2016 в 23:16

1 ответ 1

7

Окей, вам нужно перебрасывать выполнение между ядрами. Это можно сделать вот как.

  1. Подсчитываем количество ядер. Это легко: Environment.ProcessorCount.
  2. Запускаем столько UI-потоков, сколько у нас ядер. Для этого берём код отсюда, и заимствуем из него класс DispatcherThread. Каждый из них представляет собой поток, в который можно переключиться при помощи await AsyncHelper.RedirectTo(t.Dispatcher); (оттуда же).
  3. Нам нужно разбросать эти потоки по ядрам. Это можно сделать как описано здесь.
  4. Теперь в нашей async-функции, если мы хотим поменять ядро, просто пишем

    currentCore = (currentCore + 1) % Environment.ProcessorCount;
    await AsyncHelper.RedirectTo(threads[currentCore].Dispatcher);
    

Полный код:

using System;
using System.Collections;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using System.Windows.Threading;

namespace SO5
{
    class Program
    {
        static List<CoreAffineDispatcherThread> threads;
        static int currentCoreNo = 0;

        static void Main(string[] args)
        {
            threads = Enumerable.Range(0, Environment.ProcessorCount)
                                .Select(coreNo => new CoreAffineDispatcherThread(coreNo))
                                .ToList();
            Run().Wait();

            foreach (var t in threads)
                t.Dispose();
        }

        static async Task Run()
        {
            for (int i = 0; i < 10; i++)
            {
                await Task.Delay(100);

                currentCoreNo = (currentCoreNo + 1) % Environment.ProcessorCount;
                await AsyncHelper.RedirectTo(threads[currentCoreNo].Dispatcher);

                var t = Thread.CurrentThread;
                Console.WriteLine($"Task reporting from thread {t.ManagedThreadId}," +
                                  $" thread pool: {t.IsThreadPoolThread}");
            }
        }

        public class CoreAffineDispatcherThread : IDisposable
        {
            public Dispatcher Dispatcher { get; private set; }

            Thread thread;

            public CoreAffineDispatcherThread(int coreNumber)
            {
                using (var barrier = new AutoResetEvent(false))
                {
                    thread = new Thread(() =>
                    {
                        Dispatcher = Dispatcher.CurrentDispatcher;
                        barrier.Set();
                        Thread.BeginThreadAffinity();

#pragma warning disable 618
                        // The call to BeginThreadAffinity guarantees stable results
                        // for GetCurrentThreadId, so we ignore the obsolete warning
                        int osThreadId = AppDomain.GetCurrentThreadId();
#pragma warning restore 618

                        // Find the ProcessThread for this thread.
                        ProcessThread thread =
                            Process.GetCurrentProcess()
                                   .Threads.Cast<ProcessThread>()
                                   .Where(t => t.Id == osThreadId)
                                   .Single();
                        // Set the thread's processor affinity
                        var cpuMask = 1 << coreNumber;
                        thread.ProcessorAffinity = new IntPtr(cpuMask);

                        Dispatcher.Run();

                        Thread.EndThreadAffinity();
                    });

                    thread.SetApartmentState(ApartmentState.STA);
                    thread.Start();
                    barrier.WaitOne();
                }
            }

            public void Dispose()
            {
                Dispatcher.InvokeShutdown();
                if (thread != Thread.CurrentThread)
                    thread.Join();
            }
        }
    }

    static class AsyncHelper
    {
        public static DispatcherRedirector RedirectTo(Dispatcher d)
        {
            return new DispatcherRedirector(d);
        }
    }

    public struct DispatcherRedirector : INotifyCompletion
    {
        public DispatcherRedirector(Dispatcher dispatcher)
        {
            this.dispatcher = dispatcher;
        }

        #region awaiter
        public DispatcherRedirector GetAwaiter()
        {
            // combined awaiter and awaitable
            return this;
        }
        #endregion

        #region awaitable
        public bool IsCompleted
        {
            get
            {
                // true means execute continuation inline
                return dispatcher.CheckAccess();
            }
        }

        public void OnCompleted(Action continuation)
        {
            dispatcher.BeginInvoke(continuation);
        }

        public void GetResult() { }
        #endregion

        Dispatcher dispatcher;
    }
}

При тестовом пробеге выдаёт:

Task reporting from thread 10, thread pool: False
Task reporting from thread 11, thread pool: False
Task reporting from thread 12, thread pool: False
Task reporting from thread 13, thread pool: False
Task reporting from thread 14, thread pool: False
Task reporting from thread 15, thread pool: False
Task reporting from thread 16, thread pool: False
Task reporting from thread 9, thread pool: False
Task reporting from thread 10, thread pool: False
Task reporting from thread 11, thread pool: False

9
  • Ну в общем понятно все. Автоматический баланс невозможен, tpl не поможет в плане автоматического распределения нагрузки в зависимости от загруженности ядер. А самому делать такой балансер по счетчикам производительности неудобно.
    – vitidev
    17 мар 2016 в 0:02
  • @vitidev: Ну, не факт, что автоматический баланс невозможен. Тестовое приложение не создавало никакой нагрузки, вот автоматический балансировщик может быть и вовсе не включался (если балансировка не нужна, выгоднее оставаться на одном потоке, чтобы не сбрасывать кеш процессора). Как простой пример поведёт себя под нагрузкой — надо тестировать. Возможность ручной балансировки и круговой загрузки ядер в примере из ответа.
    – VladD
    17 мар 2016 в 0:05
  • currentCore = (currentCore + 1) % 8; очевидно в расчете на 8 ядер, а если их больше или меньше? для меня ответ очевиден, но копипастер будет удивлен =)
    – rdorn
    17 мар 2016 в 0:58
  • @rdorn: Точно! Исправил, спасибо!
    – VladD
    17 мар 2016 в 10:40
  • @VladD А привязаны ли вообще потоки из пула к конкретным ядрам? Я знаю, что однопоточные программы любят занимать одно ядро, из чего и сделал вывод, что поток прибит к ядру...но относится ли это к пулу. Ведь сложно выполнить очередную таску на другом ядре если потоки прибиты к ядрам...ну разве что заранее создавать по пачке потоков на каждое ядро и стараться выбирать поток из соседнего ядра.
    – vitidev
    17 мар 2016 в 10:43

Ваш ответ

By clicking “Отправить ответ”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Всё ещё ищете ответ? Посмотрите другие вопросы с метками или задайте свой вопрос.