Re[11]: TPL или Go - .NET

Привет всем,

Небольшой тест на использование TPL и тому подобных библиотек и что будет если агентов будет мало или слишком много...
Идея простая: данные падают в очередь там разбираются на n воркеров, после обработки (5ms) каждым воркером сливаются обратно.

TPL

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

internal class Program
{
    const int processingTime = 5;

    private static IPropagatorBlock<int, int> Reduce(int seed)
    {
        var total = seed;
        var input = new ActionBlock<int>(async arg =>
        {
            await Task.Delay(processingTime);
            total += arg;
        }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, BoundedCapacity = 1 });

        var output = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1 });
        input.Completion.ContinueWith(done =>
        {
            output.SendAsync(total).ContinueWith(task => output.Complete());
        });

        return DataflowBlock.Encapsulate(input, output);
    }

    private static IReadOnlyCollection<ISourceBlock<int>> CreateWorkers(ISourceBlock<int> source, int totalWorkers, Func<int, IPropagatorBlock<int, int>> workerFactory)
    {
        var output = new List<ISourceBlock<int>>();
        for (var i = 0; i < totalWorkers; ++i)
        {
            var block = workerFactory(0);
            source.LinkTo(block, new DataflowLinkOptions { PropagateCompletion = true });
            output.Add(block);
        }

        return output;
    }

    private static ISourceBlock<int> MergeResults(IReadOnlyCollection<ISourceBlock<int>> workers, Func<int, IPropagatorBlock<int, int>> factory)
    {
        var localdata = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = workers.Count });
        var output = factory(0);
        var totalWorkers = workers.Count;

        foreach (var worker in workers)
        {
            worker.LinkTo(localdata);
            worker.Completion.ContinueWith(task =>
            {
                if (Interlocked.Decrement(ref totalWorkers) == 0)
                    localdata.Complete();
            });
        }
        localdata.LinkTo(output, new DataflowLinkOptions { PropagateCompletion = true });

        return output;
    }

    private static async Task<int> Test(int iterations, int totalWorkers)
    {
        var buffer = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1 });
        var workers = CreateWorkers(buffer, totalWorkers, Reduce);
        var results = MergeResults(workers, Reduce);
        for (var i = 0; i < iterations; ++i)
        {
            await buffer.SendAsync(1);
        }
        buffer.Complete();
        return await results.ReceiveAsync();
    }

    private static void Measure(TextWriter writer, int iterations, int workers)
    {
        var estimated = TimeSpan.FromMilliseconds(processingTime * iterations / workers).TotalSeconds;
        var timer = Stopwatch.StartNew();

        var total = Test(iterations, workers).GetAwaiter().GetResult();
        if (total != iterations)
            throw new ApplicationException($"{total} != {iterations}");

        var elapsed = timer.Elapsed.TotalSeconds;
        writer.WriteLine($"Iterations: {iterations}, Workers: {workers}, Elapsed: {elapsed}s, Estimated: {estimated}s, Loss: {elapsed - estimated}s");
    }

    static void Main(string[] args)
    {
        Measure(TextWriter.Null, 1, 1);

        Console.Out.WriteLine($"Processing time: {processingTime}ms");

        var data = new Dictionary<int, int[]>
        {
            [1000] = new[] { 1, 10, 100 },
            [10000] = new[] { 50, 100, 200 },
            [100000] = new[] { 50, 100, 500, 1000 },
            [1000000] = new[] { 100, 200, 500, 1000 },
        };

        var keys = data.OrderBy(item => item.Key);
        for (var i = 1; i <= 10; ++i)
        {
            var timer = Stopwatch.StartNew();
            foreach (var item in keys)
            {
                foreach (var workers in item.Value)
                {
                    Measure(Console.Out, item.Key, workers);
                }
            }
            var elapsed = timer.Elapsed.TotalSeconds;
            Console.Out.WriteLine($"Iteration: {i}, Elapsed: {elapsed}s");
        }
    }
}

GoLang

package main

import (
    "io/ioutil"
    "log"
    "os"
    "sort"
    "sync"
    "time"
)

const processingTime = time.Duration(time.Millisecond * 5)

type workerFactory func(int, chan int) chan int

func reduce(seed int, c chan int) chan int {
    output := make(chan int)
    process := func() {
        total := seed
        for i := range c {
            time.Sleep(processingTime)
            total = total + i
        }
        output <- total
        close(output)
    }

    go process()
    return output
}

func createWorkers(buffer chan int, totalWorkers int, factory workerFactory) []chan int {
    output := make([]chan int, totalWorkers)
    for i := 0; i < totalWorkers; i++ {
        output[i] = factory(0, buffer)
    }
    return output
}

func mergeResults(workers []chan int, factory workerFactory) chan int {
    localdata := make(chan int, len(workers))
    output := factory(0, localdata)
    wd := sync.WaitGroup{}

    for _, worker := range workers {
        wd.Add(1)
        go func(w chan int) {
            for data := range w {
                localdata <- data
            }

            wd.Done()
        }(worker)
    }

    go func() {
        wd.Wait()
        close(localdata)
    }()

    return output
}

func test(iterations int, totalWorkers int) int {
    buffer := make(chan int)
    workers := createWorkers(buffer, totalWorkers, reduce)
    results := mergeResults(workers, reduce)

    for i := 0; i < iterations; i++ {
        buffer <- 1
    }

    close(buffer)

    return <-results
}

func measure(writer *log.Logger, iterations int, totalWorkers int) {
    estimated := time.Duration(iterations * int(processingTime) / totalWorkers)
    startTime := time.Now()

    total := test(iterations, totalWorkers)
    if total != iterations {
        log.Fatalf("%d != %d", total, iterations)
    }

    elapsed := time.Since(startTime)
    writer.Printf("Iterations: %d, Workers: %d, Elapsed: %s, Estimated: %s, Loss: %s\n", iterations, totalWorkers, elapsed, estimated, elapsed-estimated)
}

func main() {
    null := log.New(ioutil.Discard, "", 0)
    std := log.New(os.Stdout, "", 0)
    measure(null, 1, 1)

    std.Printf("Processing time: %s\n", processingTime)
    data := map[int][]int{
        1000:    []int{1, 10, 100},
        10000:   []int{50, 100, 200},
        100000:  []int{50, 100, 200, 500, 1000},
        1000000: []int{100, 200, 500, 1000},
    }

    var keys []int
    for key := range data {
        keys = append(keys, key)
    }

    sort.Ints(keys)

    for i := 1; i <= 10; i++ {
        start := time.Now()
        for _, iterations := range keys {
            for _, workers := range data[iterations] {
                measure(std, iterations, workers)
            }
        }

        elapsed := time.Since(start)
        std.Printf("Iteration %d, Elapsed %s\n", i, elapsed)
    }
}

Что удивительно, по количеству строк кода совпало в пределах погрешности. Правда по числу букв больше почти на 40%

Ну и результаты

GoLang:
Processing time: 5ms
Iterations: 1000, Workers: 1, Elapsed: 5.3568382s, Estimated: 5s, Loss: 356.8382ms
Iterations: 1000, Workers: 10, Elapsed: 583.9921ms, Estimated: 500ms, Loss: 83.9921ms
Iterations: 1000, Workers: 100, Elapsed: 590.8429ms, Estimated: 50ms, Loss: 540.8429ms
Iterations: 10000, Workers: 50, Elapsed: 1.3435108s, Estimated: 1s, Loss: 343.5108ms
Iterations: 10000, Workers: 100, Elapsed: 1.063587s, Estimated: 500ms, Loss: 563.587ms
Iterations: 10000, Workers: 200, Elapsed: 1.3624499s, Estimated: 250ms, Loss: 1.1124499s
Iterations: 100000, Workers: 50, Elapsed: 11.0128041s, Estimated: 10s, Loss: 1.0128041s
Iterations: 100000, Workers: 100, Elapsed: 5.9047845s, Estimated: 5s, Loss: 904.7845ms
Iterations: 100000, Workers: 200, Elapsed: 3.7495542s, Estimated: 2.5s, Loss: 1.2495542s
Iterations: 100000, Workers: 500, Elapsed: 3.7843446s, Estimated: 1s, Loss: 2.7843446s
Iterations: 100000, Workers: 1000, Elapsed: 5.911478s, Estimated: 500ms, Loss: 5.411478s
Iterations: 1000000, Workers: 100, Elapsed: 54.2715363s, Estimated: 50s, Loss: 4.2715363s
Iterations: 1000000, Workers: 200, Elapsed: 27.8255456s, Estimated: 25s, Loss: 2.8255456s
Iterations: 1000000, Workers: 500, Elapsed: 13.5996723s, Estimated: 10s, Loss: 3.5996723s
Iterations: 1000000, Workers: 1000, Elapsed: 11.1325319s, Estimated: 5s, Loss: 6.1325319s
Iteration 1, Elapsed 2m27.4934724s

Processing time: 100ms
Iterations: 1000, Workers: 10, Elapsed: 11.0497247s, Estimated: 10s, Loss: 1.0497247s
Iterations: 1000, Workers: 100, Elapsed: 11.0446942s, Estimated: 1s, Loss: 10.0446942s
Iterations: 10000, Workers: 50, Elapsed: 25.1061292s, Estimated: 20s, Loss: 5.1061292s
Iterations: 10000, Workers: 100, Elapsed: 20.0803274s, Estimated: 10s, Loss: 10.0803274s
Iterations: 10000, Workers: 200, Elapsed: 25.1038125s, Estimated: 5s, Loss: 20.1038125s
Iterations: 100000, Workers: 50, Elapsed: 3m25.7957s, Estimated: 3m20s, Loss: 5.7957s
Iterations: 100000, Workers: 100, Elapsed: 1m50.4477704s, Estimated: 1m40s, Loss: 10.4477704s
Iterations: 100000, Workers: 200, Elapsed: 1m10.3049015s, Estimated: 50s, Loss: 20.3049015s
Iterations: 100000, Workers: 500, Elapsed: 1m10.3096703s, Estimated: 20s, Loss: 50.3096703s
TPL:
Processing time: 5ms
Iterations: 1000, Workers: 1, Elapsed: 15.6350619s, Estimated: 5s, Loss: 10.6350619s
Iterations: 1000, Workers: 10, Elapsed: 1.7193998s, Estimated: 0.5s, Loss: 1.2193998s
Iterations: 1000, Workers: 100, Elapsed: 1.7178575s, Estimated: 0.05s, Loss: 1.6678575s
Iterations: 10000, Workers: 50, Elapsed: 3.9055687s, Estimated: 1s, Loss: 2.9055687s
Iterations: 10000, Workers: 100, Elapsed: 3.1233519s, Estimated: 0.5s, Loss: 2.6233519s
Iterations: 10000, Workers: 200, Elapsed: 3.9079247s, Estimated: 0.25s, Loss: 3.6579247s
Iterations: 100000, Workers: 50, Elapsed: 32.0285062s, Estimated: 10s, Loss: 22.0285062s
Iterations: 100000, Workers: 100, Elapsed: 17.2048496s, Estimated: 5s, Loss: 12.2048496s
Iterations: 100000, Workers: 500, Elapsed: 15.1436388s, Estimated: 1s, Loss: 14.1436388s
Iterations: 100000, Workers: 1000, Elapsed: 28.7291761s, Estimated: 0.5s, Loss: 28.2291761s
Iterations: 1000000, Workers: 100, Elapsed: 157.9577418s, Estimated: 50s, Loss: 107.9577418s
Iterations: 1000000, Workers: 200, Elapsed: 81.5264742s, Estimated: 25s, Loss: 56.5264742s
Iterations: 1000000, Workers: 500, Elapsed: 80.9878838s, Estimated: 10s, Loss: 70.9878838s
Iterations: 1000000, Workers: 1000, Elapsed: 144.7310845s, Estimated: 5s, Loss: 139.7310845s
Iteration: 1, Elapsed: 588.3252783s
Processing time: 100ms
Iterations: 1000, Workers: 10, Elapsed: 12.0203535s, Estimated: 10s, Loss: 2.0203535s
Iterations: 1000, Workers: 100, Elapsed: 12.0300487s, Estimated: 1s, Loss: 11.0300487s
Iterations: 10000, Workers: 50, Elapsed: 27.3427199s, Estimated: 20s, Loss: 7.3427199s
Iterations: 10000, Workers: 100, Elapsed: 21.8628653s, Estimated: 10s, Loss: 11.8628653s
Iterations: 10000, Workers: 200, Elapsed: 27.3254114s, Estimated: 5s, Loss: 22.3254114s
Iterations: 100000, Workers: 50, Elapsed: 224.1880851s, Estimated: 200s, Loss: 24.1880851s
Iterations: 100000, Workers: 100, Elapsed: 119.6893727s, Estimated: 100s, Loss: 19.6893727s

Альтернативный вариант Thread.Sleep и TPL:
Processing time: 5ms
Iterations: 1000, Workers: 1, Elapsed: 5.6992209s, Estimated: 5s, Loss: 0.699220899999999s
Iterations: 1000, Workers: 10, Elapsed: 0.7892746s, Estimated: 0.5s, Loss: 0.2892746s
Iterations: 1000, Workers: 100, Elapsed: 1.2913144s, Estimated: 0.05s, Loss: 1.2413144s
Iterations: 10000, Workers: 50, Elapsed: 7.5742033s, Estimated: 1s, Loss: 6.5742033s
Iterations: 10000, Workers: 100, Elapsed: 7.9033241s, Estimated: 0.5s, Loss: 7.4033241s
Iterations: 10000, Workers: 200, Elapsed: 8.4574322s, Estimated: 0.25s, Loss: 8.2074322s
Iterations: 100000, Workers: 50, Elapsed: 41.7972214s, Estimated: 10s, Loss: 31.7972214s
Iterations: 100000, Workers: 100, Elapsed: 13.6094305s, Estimated: 5s, Loss: 8.6094305s
Iterations: 100000, Workers: 500, Elapsed: 12.7404464s, Estimated: 1s, Loss: 11.7404464s
Iterations: 100000, Workers: 1000, Elapsed: 13.8424686s, Estimated: 0.5s, Loss: 13.3424686s
Iterations: 1000000, Workers: 100, Elapsed: 77.8382217s, Estimated: 50s, Loss: 27.8382217s
Iterations: 1000000, Workers: 200, Elapsed: 66.6656733s, Estimated: 25s, Loss: 41.6656733s
Iterations: 1000000, Workers: 500, Elapsed: 57.0298951s, Estimated: 10s, Loss: 47.0298951s
Iterations: 1000000, Workers: 1000, Elapsed: 53.365894s, Estimated: 5s, Loss: 48.365894s
Iteration: 1, Elapsed: 368.6105906s
Processing time: 100ms
Iterations: 1000, Workers: 10, Elapsed: 11.4807278s, Estimated: 10s, Loss: 1.4807278s
Iterations: 1000, Workers: 100, Elapsed: 16.9978477s, Estimated: 1s, Loss: 15.9978477s
Iterations: 10000, Workers: 50, Elapsed: 36.6479652s, Estimated: 20s, Loss: 16.6479652s
Iterations: 10000, Workers: 100, Elapsed: 24.3926649s, Estimated: 10s, Loss: 14.3926649s
Iterations: 10000, Workers: 200, Elapsed: 31.9125948s, Estimated: 5s, Loss: 26.9125948s
Iterations: 100000, Workers: 50, Elapsed: 206.0061483s, Estimated: 200s, Loss: 6.00614829999998s
Iterations: 100000, Workers: 100, Elapsed: 111.0950787s, Estimated: 100s, Loss: 11.0950787s
Iterations: 100000, Workers: 500, Elapsed: 119.0079956s, Estimated: 20s, Loss: 99.0079956s

Итого:
1. На мелкие асинхронные задачах TPL откровенно сливается.
2. На относительно крупных лучше но, тоже не фонтан.
3. ОС писали похоже не дураки и ей проще поднять кучу потоков, чем разбираться с Task.Delay

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

TK>1. На мелкие асинхронные задачах TPL откровенно сливается.
TK>2. На относительно крупных лучше но, тоже не фонтан.
TK>3. ОС писали похоже не дураки и ей проще поднять кучу потоков, чем разбираться с Task.Delay

1. В тесте мертворожденный TPL Dataflow (последнее обновление 2 года назад), а не просто TPL
2. new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, BoundedCapacity = 1 }); после выделенного результаты можно не читать
3. В .NET вовсе необязательно использовать модель акторов, есть гораздо более простые и быстрые способы. TPL тоже не дураки писали.

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

TK>Привет всем,

TK>Небольшой тест на использование TPL и тому подобных библиотек и что будет если агентов будет мало или слишком много...
TK>Идея простая: данные падают в очередь там разбираются на n воркеров, после обработки (5ms) каждым воркером сливаются обратно.

Правильно понял, что весь код можно на .net сделать так:

private static async Task<int> Test(int iterations)
{
    var results = await Task.WhenAll(
        Enumerable.Range(0, iterations)
        .Select(x =>
            Task.Run(async () => {
                await Task.Delay(TimeSpan.FromSeconds(5));
                return x;
            })));
    return results.Sum();
}

?

Количество реальных воркеров в любом случае система выбирает, поэтому нет смысла делать воркеры разными, если они все делают одно и то же.

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

G>1. В тесте мертворожденный TPL Dataflow (последнее обновление 2 года назад), а не просто TPL
G>2. new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 1, BoundedCapacity = 1 }); после выделенного результаты можно не читать
G>3. В .NET вовсе необязательно использовать модель акторов, есть гораздо более простые и быстрые способы. TPL тоже не дураки писали.

Код то есть что-бы показать?

Вижу

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

TK>Привет всем,

TK>Небольшой тест на использование TPL и тому подобных библиотек и что будет если агентов будет мало или слишком много...

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

Замеры — не для TPL, а TPL dataflow. Забавный такой зверёк — никто так толком и не может объяснить, зачем и откуда оно прибилось, а выбрасывать жалко.
Своеобразную производительность ещё можно вытерпеть, а вот необходимость кропотливо собирать элементарщину через вот такие вот приседания убивает всю затею на корню.

А главная хохма в следующем:
1. Task.Delay() действительно оччень неэффективно реализован, статей про это дело куча, навскидку:
http://stackoverflow.com/questions/32132544/how-is-the-performance-when-there-are-hundreds-of-task-delay
https://ayende.com/blog/174851/reducing-allocations-and-resource-usages-when-using-task-delay
но проблема внезапно не в нём.

2. Угу, текущая реализация async в принципе не предназначена для раскидывания миллионов конкурентных задач. По опыту использования в bing services оччень многое было допилено, в седьмом шарпе с ValueTask должно быть ещё немного получше, но это когда ещё будет?

3. К чему это я:
вся простыня наотлично заменяется на

    private static async Task MeasureAsync(TextWriter writer, int iterations, int workers)
    {
        var estimated = TimeSpan.FromMilliseconds(1.0 * processingTime * iterations / workers).TotalSeconds;
        var timer = Stopwatch.StartNew();

        var repeat = iterations/workers;

        var t = Enumerable.Range(0, workers).Select(async (i) =>
        {
            for (int j = 0; j < repeat; j++)
            {
                await Task.Delay(processingTime);
            }
            return repeat;
        });

        var total = (await Task.WhenAll(t)).Sum();
        if (total != iterations)
            throw new ApplicationException($"{total} != {iterations}");

        var elapsed = timer.Elapsed.TotalSeconds;
        writer.WriteLine($"Iterations    (async): {iterations}, Workers: {workers}, Elapsed: {elapsed}s, Estimated: {estimated}s, Loss: {elapsed - estimated}s");
    }

с следующими результатами:

простыня

Iterations (DataFlow): 1000, Workers: 1, Elapsed: 15,6414634s, Estimated: 5s, Loss: 10,6414634s
Iterations    (async): 1000, Workers: 1, Elapsed: 15,6232925s, Estimated: 5s, Loss: 10,6232925s

Iterations (DataFlow): 1000, Workers: 10, Elapsed: 1,7185944s, Estimated: 0,5s, Loss: 1,2185944s
Iterations    (async): 1000, Workers: 10, Elapsed: 1,5605506s, Estimated: 0,5s, Loss: 1,0605506s

Iterations (DataFlow): 1000, Workers: 100, Elapsed: 1,7212329s, Estimated: 0,05s, Loss: 1,6712329s
Iterations    (async): 1000, Workers: 100, Elapsed: 0,1527294s, Estimated: 0,05s, Loss: 0,1027294s

Iterations (DataFlow): 10000, Workers: 50, Elapsed: 3,906125s, Estimated: 1s, Loss: 2,906125s
Iterations    (async): 10000, Workers: 50, Elapsed: 3,1275499s, Estimated: 1s, Loss: 2,1275499s

Iterations (DataFlow): 10000, Workers: 100, Elapsed: 3,1269456s, Estimated: 0,5s, Loss: 2,6269456s
Iterations    (async): 10000, Workers: 100, Elapsed: 1,5618327s, Estimated: 0,5s, Loss: 1,0618327s

Iterations (DataFlow): 10000, Workers: 200, Elapsed: 3,9020185s, Estimated: 0,25s, Loss: 3,6520185s
Iterations    (async): 10000, Workers: 200, Elapsed: 0,7807316s, Estimated: 0,25s, Loss: 0,5307316s

Iterations (DataFlow): 100000, Workers: 50, Elapsed: 32,0800174s, Estimated: 10s, Loss: 22,0800174s
Iterations    (async): 100000, Workers: 50, Elapsed: 31,2482231s, Estimated: 10s, Loss: 21,2482231s

Iterations (DataFlow): 100000, Workers: 100, Elapsed: 17,1876135s, Estimated: 5s, Loss: 12,1876135s
Iterations    (async): 100000, Workers: 100, Elapsed: 15,7520876s, Estimated: 5s, Loss: 10,7520876s

Iterations (DataFlow): 100000, Workers: 500, Elapsed: 17,8931075s, Estimated: 1s, Loss: 16,8931075s
Iterations    (async): 100000, Workers: 500, Elapsed: 3,1235903s, Estimated: 1s, Loss: 2,1235903s

Iterations (DataFlow): 100000, Workers: 1000, Elapsed: 31,5777071s, Estimated: 0,5s, Loss: 31,0777071s
Iterations    (async): 100000, Workers: 1000, Elapsed: 1,5621821s, Estimated: 0,5s, Loss: 1,0621821s

Iterations (DataFlow): 1000000, Workers: 100, Elapsed: 161,0784262s, Estimated: 50s, Loss: 111,0784262s
Iterations    (async): 1000000, Workers: 100, Elapsed: 157,1561241s, Estimated: 50s, Loss: 107,1561241s

Iterations (DataFlow): 1000000, Workers: 200, Elapsed: 81,7184672s, Estimated: 25s, Loss: 56,7184672s
Iterations    (async): 1000000, Workers: 200, Elapsed: 78,1205434s, Estimated: 25s, Loss: 53,1205434s

Iterations (DataFlow): 1000000, Workers: 500, Elapsed: 96,485345s, Estimated: 10s, Loss: 86,485345s
Iterations    (async): 1000000, Workers: 500, Elapsed: 31,2533724s, Estimated: 10s, Loss: 21,2533724s

Iterations (DataFlow): 1000000, Workers: 1000, Elapsed: 183,5426835s, Estimated: 5s, Loss: 178,5426835s
Iterations    (async): 1000000, Workers: 1000, Elapsed: 15,6299033s, Estimated: 5s, Loss: 10,6299033s

С учётом того, что для реально нагруженной обработки пакетов есть собственно TPL + Parallel.For с partitioners, тему можно закрывать целиком и полностью.

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

S>2. Угу, текущая реализация async в принципе не предназначена для раскидывания миллионов конкурентных задач. По опыту использования в bing services оччень многое было допилено, в седьмом шарпе с ValueTask должно быть ещё немного получше, но это когда ещё будет?

S>3. К чему это я:
S>вся простыня наотлично заменяется на
S>

S>    private static async Task MeasureAsync(TextWriter writer, int iterations, int workers)
S>    {
S>        var estimated = TimeSpan.FromMilliseconds(1.0 * processingTime * iterations / workers).TotalSeconds;
S>        var timer = Stopwatch.StartNew();

S>        var repeat = iterations/workers;

S>        var t = Enumerable.Range(0, workers).Select(async (i) =>
S>        {
S>            for (int j = 0; j < repeat; j++)
S>            {
S>                await Task.Delay(processingTime);
S>            }
S>            return repeat;
S>        });

S>        var total = (await Task.WhenAll(t)).Sum();
S>        if (total != iterations)
S>            throw new ApplicationException($"{total} != {iterations}");

S>        var elapsed = timer.Elapsed.TotalSeconds;
S>        writer.WriteLine($"Iterations    (async): {iterations}, Workers: {workers}, Elapsed: {elapsed}s, Estimated: {estimated}s, Loss: {elapsed - estimated}s");
S>    }

С таким подходом менять надо сразу на return iterations — это будет победитель данного конкурса

А так, даже рядом не похоже

но, в целом лучше тут хотя-бы Task.WhenAll по числу воркеров. gandjustas туда вообще массив размером с iterations вставил

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

TK>С таким подходом менять надо сразу на return iterations — это будет победитель данного конкурса

Ну а разница? типа counter++ после Delay() на что-то повлияет в плане производительности?

Если серьёзно, то перед тем, как говорить о сферической производительности надо с задачей и с контекстом определиться. В шарпе для асинхронщины только сходу вспоминается
TPL, TPL DataFlow, PLINQ, await, Rx, Akka.Net и, наконец, Orleans, у каждого — своя область применения. Ну а сколько всякой мелочёвки и своих шедулеров под TPL вообще никто не считал.

Задача какая? Не в смысле "нужна очередь + n worker-ов", это уже решение. Я про исходную задачу — чего сделать-то надо? : )

TK>А так, даже рядом не похоже

Но делает то же самое, внезапно.

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

TK>>Небольшой тест на использование TPL и тому подобных библиотек и что будет если агентов будет мало или слишком много...
TK>>Идея простая: данные падают в очередь там разбираются на n воркеров, после обработки (5ms) каждым воркером сливаются обратно.

G>Правильно понял, что весь код можно на .net сделать так:

Что тут сказать... Фазу map освоили. Reduce где???

G>Количество реальных воркеров в любом случае система выбирает, поэтому нет смысла делать воркеры разными, если они все делают одно и то же.

В исходном коде Reduce возвращает интерфейс для общения с нодой в кластере (пушим в неё данные, потом забираем ответ). Каждая нода стоит $$$ — что значит система выберет сколько ей нужно???

Где вообще все это в вашем коде? по коду же понятно, что на выходе будет исходное число и итераций — его и надо было хардкодить.

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

TK>>С таким подходом менять надо сразу на return iterations — это будет победитель данного конкурса

S>Ну а разница? типа counter++ после Delay() на что-то повлияет в плане производительности?

Конечно, никакой разницы. Тесты ведь проходят?

S>Задача какая? Не в смысле "нужна очередь + n worker-ов", это уже решение. Я про исходную задачу — чего сделать-то надо? : )

Ну уж явное не завернуть iterations / workers в task и посчитать сумму...

TK>>А так, даже рядом не похоже

S>Но делает то же самое, внезапно.

Раньше суммировал поток входных данных (произвольной длины), а теперь возвращает константу

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

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

TK>>>Небольшой тест на использование TPL и тому подобных библиотек и что будет если агентов будет мало или слишком много...
TK>>>Идея простая: данные падают в очередь там разбираются на n воркеров, после обработки (5ms) каждым воркером сливаются обратно.

G>>Правильно понял, что весь код можно на .net сделать так:
TK>Что тут сказать... Фазу map освоили. Reduce где???
Ты о чем? Reduce — тупо сумма.

G>>Количество реальных воркеров в любом случае система выбирает, поэтому нет смысла делать воркеры разными, если они все делают одно и то же.
TK>В исходном коде Reduce возвращает интерфейс для общения с нодой в кластере (пушим в неё данные, потом забираем ответ). Каждая нода стоит $$$ — что значит система выберет сколько ей нужно???
Прости, а что ты пытаешься сравнить? Если у тебя ноды в кластере, то на передачу данных времени уйдет на три порядка больше, чем на работу кода по этой самой передаче.

TK>Где вообще все это в вашем коде? по коду же понятно, что на выходе будет исходное число и итераций — его и надо было хардкодить.
Нигде, это и не нужно вообще. В твоем коде тоже нету отдельных нод.

Если бы это были отдельные физические ноды, то код и библиотеки были бы совсем другие на обоих языках.
Например для .NET это будет https://dotnet.github.io/orleans/ или (возможно) Akka.NET, а вовсе не Dataflow.

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

G>1. В тесте мертворожденный TPL Dataflow (последнее обновление 2 года назад), а не просто TPL

Походу вы в 2014году так и остались

https://www.nuget.org/packages/System.Threading.Tasks.Dataflow/4.6.0

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

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

G>>1. В тесте мертворожденный TPL Dataflow (последнее обновление 2 года назад), а не просто TPL

TK>Походу вы в 2014году так и остались

TK>https://www.nuget.org/packages/System.Threading.Tasks.Dataflow/4.6.0

Такие разговоры мне всегда напоминают сцену из фильма С***ли (Snatch в переводе гоблина)

(*действие происходит в Лондоне*)
— Я очень не люблю выезжать из своего города, тем более не туда где есть песчаные пляжи и девушки в бикини приносят коктейли
— Но у нас тут есть песчаные пляжи!!!
— Ну и кому нафиг они нужны?

Сидит в редмонде команда (индусов) и пилит никому не нужный фреймворк\либу, который имеет по одному релизу в ГОД. При этом команда не общается с пользователями от слова вообще. Релизы походу пилит под релизы .NET FW.

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

G>>>Правильно понял, что весь код можно на .net сделать так:
TK>>Что тут сказать... Фазу map освоили. Reduce где???
G>Ты о чем? Reduce — тупо сумма.

Откуда такие познания? Скорее уж Reduce это тупо оператор.
Например, можно минимальное значение возвра

G>Прости, а что ты пытаешься сравнить? Если у тебя ноды в кластере, то на передачу данных времени уйдет на три порядка больше, чем на работу кода по этой самой передаче.

Так вы код нормальный пишите и порядков никаких не будет. 5мсек на отправку целого более чем достаточно.

TK>>Где вообще все это в вашем коде? по коду же понятно, что на выходе будет исходное число и итераций — его и надо было хардкодить.
G>Нигде, это и не нужно вообще. В твоем коде тоже нету отдельных нод.

В моем коде за это отвечает — метод CreateWorkers. Работа с нодами скрыта за самодостаточным интерфейсом. Можно локально обрабатывать данные, можно удаленно. Это уже не принципиально.

G>Если бы это были отдельные физические ноды, то код и библиотеки были бы совсем другие на обоих языках.
G>Например для .NET это будет https://dotnet.github.io/orleans/ или (возможно) Akka.NET, а вовсе не Dataflow.

Думаете, что orleans все за вас волшебным образом сделает?

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

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

G>>>>Правильно понял, что весь код можно на .net сделать так:
TK>>>Что тут сказать... Фазу map освоили. Reduce где???
G>>Ты о чем? Reduce — тупо сумма.
TK>Откуда такие познания? Скорее уж Reduce это тупо оператор.
TK>Например, можно минимальное значение возвра
И какая разница в принципе?

G>>Прости, а что ты пытаешься сравнить? Если у тебя ноды в кластере, то на передачу данных времени уйдет на три порядка больше, чем на работу кода по этой самой передаче.
TK>Так вы код нормальный пишите и порядков никаких не будет. 5мсек на отправку целого более чем достаточно.
А при чем тут код? Банальная передача инта по сети займет условно 1мс. Это на каждый iteration. У тебя итераций 1 000 000, то есть передача по сети отнимет около 16 минут.
130 сек оверхеда от dataflow на этом фоне потеряются. А в реальности будет еще хуже, потому что передать надо в две стороны.

TK>>>Где вообще все это в вашем коде? по коду же понятно, что на выходе будет исходное число и итераций — его и надо было хардкодить.
G>>Нигде, это и не нужно вообще. В твоем коде тоже нету отдельных нод.
TK>В моем коде за это отвечает — метод CreateWorkers. Работа с нодами скрыта за самодостаточным интерфейсом. Можно локально обрабатывать данные, можно удаленно. Это уже не принципиально.

То есть не принципиально какой код ты замеряешь? Тогда какой смысл твоего поста?
Ты еще не понял, что в случае удаленных узлов код будет кардинально другой?

G>>Если бы это были отдельные физические ноды, то код и библиотеки были бы совсем другие на обоих языках.
G>>Например для .NET это будет https://dotnet.github.io/orleans/ или (возможно) Akka.NET, а вовсе не Dataflow.
TK>Думаете, что orleans все за вас волшебным образом сделает?

А ты думаешь что go волшебным образом каналы прокинет на другой сервер?
Или что ты хочешь этим сказать?

Твои замеры не имеют никакого смысла для задачи, про которую ты говоришь.
А для задачи, которую ты реально решаешь в своем примере, код на .NET глубоко ошибочен.
Понятно?

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

TK>>Откуда такие познания? Скорее уж Reduce это тупо оператор.
TK>>Например, можно минимальное значение возвра
G>И какая разница в принципе?

Грубо говоря вы путаете сложение и умножение. А так, в принципе да, разницы никакой

TK>>Так вы код нормальный пишите и порядков никаких не будет. 5мсек на отправку целого более чем достаточно.
G>А при чем тут код? Банальная передача инта по сети займет условно 1мс. Это на каждый iteration. У тебя итераций 1 000 000, то есть передача по сети отнимет около 16 минут.
G>130 сек оверхеда от dataflow на этом фоне потеряются. А в реальности будет еще хуже, потому что передать надо в две стороны.

В исходном сообщении было 5мс на итерацию. На 1000000 итераций по 5мсек требовалось от ~15сек до минуты.
16 минут это если их слать в одно рыло одно за другим

понятно, что и оверхед тут будет сем-восем-двенадцать. и стороны не две, а четыре...

G>Понятно?
Понятно

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

TK>Грубо говоря вы путаете сложение и умножение. А так, в принципе да, разницы никакой

В данном обсуждении вообще никакой.

G>>А при чем тут код? Банальная передача инта по сети займет условно 1мс. Это на каждый iteration. У тебя итераций 1 000 000, то есть передача по сети отнимет около 16 минут.
G>>130 сек оверхеда от dataflow на этом фоне потеряются. А в реальности будет еще хуже, потому что передать надо в две стороны.
TK>В исходном сообщении было 5мс на итерацию. На 1000000 итераций по 5мсек требовалось от ~15сек до минуты.
TK>16 минут это если их слать в одно рыло одно за другим

понятно, что и оверхед тут будет сем-восем-двенадцать. и стороны не две, а четыре...
Ага, и не выйграл, а проиграл.

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

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

TK>>Грубо говоря вы путаете сложение и умножение. А так, в принципе да, разницы никакой

G>В данном обсуждении вообще никакой.

Посмотри на свой код — в нем никакого reduce нету — там есть только сумма по всему массиву элементов. Что-бы понять проблему — поставь количество итераций в int.MaxValue и запусти на соответствующей платформе. Очевидно же, что Task.WhenAll с его массивом никуда не годится.
Смысл reduce в том, что мы не работаем со всем массивом данных целиком, а основная работа разбивается на n потоков и по завершению по n результатам делается ещё один раз reduce. В данном случае это принципиальное отличие от простого Sum по массиву (количество входных данных может быть сильно больше int)

G>Все равно реальный код, работающий на множестве серверов, будет бесконечно отличатся от представленного. О чем разговор вообще?

Меня вот больше пугает, что есть люди которым дают потенциально не ограниченную IEnumerable последовательность и они ей фактически делают ToArray. О каких серверах вообще речь может идти?

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

TK>>А так, даже рядом не похоже

S>Но делает то же самое, внезапно.

Это пока число итераций делится нацело на число воркеров

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

TK>Меня вот больше пугает, что есть люди которым дают потенциально не ограниченную IEnumerable последовательность и они ей фактически делают ToArray. О каких серверах вообще речь может идти?
Угу, любой падаван сходу должен распознавать великий замысел и из одного "метод обозван Reduce" делать глубокомысленный вывод "где-то тут должен быть кластер".

Если коротко, по пунктам:
0. Для распределённой обработки данных есть orleans, но он очень малопопулярен и на нём крутится всякая мелочёвка типа Halo и каких-то сервисов Visa, народ своё лепит.
1. Для бысрых онлайн-числодробилок можно глянуть статьи и видео от Age Of Ascent. Ben Adams и ко развлекаются с шарпом, Azure services, вебом и MMORG на 50k одновременных игроков на одной карте.
2. Представленные замеры к обсуждаемой теме относятся как обсуждение преимуществ шиповки применительно к шинам формулы 1.

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

TK>>Меня вот больше пугает, что есть люди которым дают потенциально не ограниченную IEnumerable последовательность и они ей фактически делают ToArray. О каких серверах вообще речь может идти?
S>Угу, любой падаван сходу должен распознавать великий замысел и из одного "метод обозван Reduce" делать глубокомысленный вывод "где-то тут должен быть кластер".

т.е даже то, что код совершенно не эквивалентный получается тоже не видно? ну

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

TK>т.е даже то, что код совершенно не эквивалентный получается тоже не видно? ну

Да видно, конечно, но стартовое сообщение про что было?

Про запустить n конкурентных воркеров и посчитать время, так? Ну вот код именно это и делает. Без обработки частных случаев, да — так они на картину в общем никак не влияют.

С TPL DataFlow, согласен, всё своеобразно. Хорошая новость: его никто и не использует. И альтернативы тоже, нынче в моде готовые решения типа akka.net. Локальное раскидывание задач наотлично с await решается. Ну, или Rx, если нужно быстро сделать и особых требований к производительности нет.

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

S>Да видно, конечно, но стартовое сообщение про что было?

S>Про запустить n конкурентных воркеров и посчитать время, так? Ну вот код именно это и делает. Без обработки частных случаев, да — так они на картину в общем никак не влияют.

В итоге то что вышло?
Один сказал, что воркеры тут не нужны (хотя, это было в условиях), распараллелил Task.Delay/Thread.Sleep, а всю остальную работу (ради чего все и затевалось) сказал, что проще сделать последовательно.
Другой решил, что очередь воркерам не нужна — все проще, когда работа делится поровну и планируется заранее, а что-бы еще проще было вместо того, что-бы посчитать сумму элементов просто вернул их количество.

Все так?

PS
В исходном коде был метод Test:

    private static async Task<int> Test(int iterations, int totalWorkers)
    {
        var buffer = new BufferBlock<int>(new DataflowBlockOptions { BoundedCapacity = 1 });
        var workers = CreateWorkers(buffer, totalWorkers, Reduce);
        var results = MergeResults(workers, Reduce);
        for (var i = 0; i < iterations; ++i)
        {
            await buffer.SendAsync(1);
        }
        buffer.Complete();
        return await results.ReceiveAsync();
    }

Который можно переписать примерно так (просто меняем foreach на IObservable или IEnumerable — не принципиально):

    private static async Task<TResult> Test<TResult>(IObservable<TResult> source, int totalWorkers, Func<IPropagatorBlock<TResult, TResult>> reduce)
    {
        var buffer = new BufferBlock<TResult>(new DataflowBlockOptions { BoundedCapacity = 1 });
        var workers = CreateWorkers(buffer, totalWorkers, reduce);
        var results = MergeResults(workers, reduce);
    source.Subscrive(buffer.AsObserver());
        buffer.Complete();
        return await results.ReceiveAsync();
    }

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

Enumerable.Range, число итераций, вызов функции с константой N раз — это просто пример для теста. В реальности такого кода не будет.

Итого, чуть сменили сигнатуру на более осмысленную — что надо поменять в ваших реализациях что-бы хоть как-то соотноситься с реальностью?

От:	TK	кывт.рф
Дата:	07.11.16 18:43
Оценка:	3 (1) -1

От:	gandjustas	http://blog.gandjustas.ru/
Дата:	07.11.16 19:24
Оценка:

От:	gandjustas	http://blog.gandjustas.ru/
Дата:	07.11.16 19:45
Оценка:

От:	TK	кывт.рф
Дата:	07.11.16 19:48
Оценка:

	От:	Sinix
	Дата:	07.11.16 19:57
	Оценка:	70 (4) +1 -1