CancellationTokenSource-ով ալիքները, որոնց հիշողության ժամկետը լրանում է, հեռացվելուց հետո

Թարմացնել

Մեկնաբանություններից.

կա պրոցեսոր՝ հաղորդագրությունները խմբաքանակով մշակելու համար: այն սկսում է մշակվել, երբ բավականաչափ հաղորդագրություններ կան կամ ժամանակն ավարտվում է, այստեղ է հայտնվում ժամանակի չեղարկումը

Սա նշանակում է, որ այն, ինչ իրականում անհրաժեշտ է, հաղորդագրությունները խմբաքանակով խմբավորելու միջոց է և՛ ըստ քանակի, և՛ ժամանակաշրջանի: Երկուսն էլ անելը համեմատաբար հեշտ է:

Այս մեթոդի խմբաքանակները ըստ հաշվարկի: Մեթոդը հաղորդագրություններ է ավելացնում batch ցուցակին, մինչև սահմանը հասնի, ուղարկում է տվյալները ներքև և մաքրում ցուցակը.

static ChannelReader<Message[]> BatchByCount(this ChannelReader<Message> input, int count, CancellationToken token=default)
{
    var channel=Channel.CreateUnbounded();
    var writer=channel.Writer;   

    _ = Task.Run(async ()=>{
        var batch=new List<Message>(count);
        await foreach(var msg in input.ReadAllAsync(token))
        {
            batch.Add(msg);
            if(batch.Count==count)
            {
                await writer.WriteAsync(batch.ToArray());
                batch.Clear();
            }
        }
    },token)
   .ContinueWith(t=>writer.TryComplete(t.Exception));
   return channel;
}

Մի մեթոդ, որը խմբաքանակ է բաժանում ըստ ժամանակաշրջանի, ավելի բարդ է, քանի որ ժմչփը կարող է միևնույն ժամանակ հաղորդագրություն ստանալ: Interlocked.Exchange-ը փոխարինում է գոյություն ունեցող batch ցուցակը նորով և ուղարկում խմբաքանակի տվյալները ներքև: :

static ChannelReader<Message[]> BatchByPeriod(this ChannelReader<Message> input, TimeSpan period, CancellationToken token=default)
{
    var channel=Channel.CreateUnbounded();
    var writer=channel.Writer;   

    var batch=new List<Message>();
    Timer t=new Timer(async obj =>{
        var data=Interlocked.Exchange(ref batch,new List<Message>());
        writer.WriteAsync(data.ToArray());
    },null,TimeSpan.Zero,period);

    _ = Task.Run(async ()=>{
        
        await foreach(var msg in input.ReadAllAsync(token))
        {
            batch.Add(msg);
        }
    },token)
   .ContinueWith(t=>{
        timer.Dispose();
        writer.TryComplete(t.Exception);
   });
   return channel;
}

Երկուսն էլ անել, ես դեռ աշխատում եմ դրա վրա: Խնդիրն այն է, որ և՛ հաշվառման, և՛ ժամանակաչափի ժամկետի ավարտը կարող է տեղի ունենալ միաժամանակ: Վատագույն դեպքում, lock(batch)-ը կարող է օգտագործվել՝ ապահովելու համար, որ միայն շարանը կամ օղակը կարող են ուղարկել տվյալները ներքև

Բնօրինակ պատասխան

Ալիքները չեն արտահոսում, երբ ճիշտ օգտագործվում են, ինչպես ցանկացած այլ տարա: Ալիքը ասինխրոն հերթ չէ և հաստատ արգելափակող չէ: Դա շատ տարբեր կոնստրուկտ է, բոլորովին այլ արտահայտություններով: Դա ավելի բարձր մակարդակի կոնտեյներ է, որը օգտագործում է հերթեր: Շատ լավ պատճառ կա, որ կան ChannelReader և ChannelWriter առանձին դասեր:

Տիպիկ սցենարն այն է, որ հրատարակիչը ստեղծի և տիրանա ալիքին: Միայն հրատարակիչը կարող է գրել այդ ալիքին և զանգահարել Complete()ին: Channel-ը չի իրականացնում IDisposable-ը, ուստի այն չի կարող տնօրինվել: Հրատարակիչը բաժանորդներին տրամադրում է միայն ChannelReader:

Բաժանորդները տեսնում են միայն ChannelReader և կարդում են դրանից մինչև այն ավարտվի: Օգտագործելով ReadAllAsync բաժանորդը կարող է շարունակել կարդալ ChannelReader-ից մինչև այն ավարտվի:

Սա բնորոշ օրինակ է.

ChannelReader<Message> Producer(CancellationToken token=default)
{
    var channel=Channel.CreateUnbounded<Message>();
    var writer=channel.Writer;

    //Create the actual "publisher" worker
    _ = Task.Run(async ()=>{
        for(int i=0;i<100;i++)
        {
            //Check for cancellation
            if(token.IsCancellationRequested)
            {
                return;
            }
            //Simulate some work
            await Task.Delay(100);
            await writer.WriteAsync(new Message(...));          
        }
    }  ,token)
    //Complete and propagate any exceptions
    .ContinueWith(t=>writer.TryComplete(t.Exception));

    //This casts to a ChannelReader
    return channel;
}

Բաժանորդին աշխատելու համար անհրաժեշտ է միայն ChannelReader: Օգտագործելով ChannelReader.ReadAllAsync բաժանորդին անհրաժեշտ է միայն await foreach հաղորդագրությունները մշակելու համար.

async Task Subscriber(ChannelReader<Message> input,CancellationToken token=default)
{
    await foreach(var msg in input.ReadAllAsync(token))
    {
        //Use the message
    }
}

Բաժանորդը կարող է արտադրել իր սեփական հաղորդագրությունները՝ վերադարձնելով ChannelReader: Եվ այստեղ ամեն ինչ շատ հետաքրքիր է դառնում, քանի որ Subscriber մեթոդը դառնում է քայլ շղթայված քայլերի խողովակաշարում: Եթե ​​մեթոդները փոխարկենք ընդլայնման մեթոդների ChannelReader-ում, մենք հեշտությամբ կարող ենք ստեղծել մի ամբողջ խողովակաշար:

Եկեք ստեղծենք որոշ թվեր.

ChannelReader<int> Generate(int nums,CancellationToken token=default)
{
    var channel=Channel.CreateBounded<int>(10);
    var writer=channel.Writer;

    //Create the actual "publisher" worker
    _ = Task.Run(async ()=>{
        for(int i=0;i<nums;i++)
        {
            //Check for cancellation
            if(token.IsCancellationRequested)
            {
                return;
            }

            await writer.WriteAsync(i*7);  
            await Task.Delay(100);        
        }
    }  ,token)
    //Complete and propagate any exceptions
    .ContinueWith(t=>writer.TryComplete(t.Exception));

    //This casts to a ChannelReader
    return channel;
}

Այնուհետև կրկնապատկեք և քառակուսիացրեք դրանք.

ChannelReader<double> Double(this ChannelReader<int> input,CancellationToken token=default)
{
    var channel=Channel.CreateBounded<double>(10);
    var writer=channel.Writer;

    //Create the actual "publisher" worker
    _ = Task.Run(async ()=>{
        await foreach(var msg in input.ReadAllAsync(token))
        {
            await writer.WriteAsync(2.0*msg);          
        }
    }  ,token)
    //Complete and propagate any exceptions
    .ContinueWith(t=>writer.TryComplete(t.Exception));

    return channel;
}

ChannelReader<double> Root(this ChannelReader<double> input,CancellationToken token=default)
{
    var channel=Channel.CreateBounded<double>(10);
    var writer=channel.Writer;

    //Create the actual "publisher" worker
    _ = Task.Run(async ()=>{
        await foreach(var msg in input.ReadAllAsync(token))
        {
            await writer.WriteAsync(Math.Sqrt(msg));          
        }
    }  ,token)
    //Complete and propagate any exceptions
    .ContinueWith(t=>writer.TryComplete(t.Exception));

    return channel;
}

Եվ վերջապես տպեք դրանք

async Task Print(this ChannelReader<double> input,CancellationToken token=default)
{
    await foreach(var msg in input.ReadAllAsync(token))
    {
        Console.WriteLine(msg);
    }
}

Այժմ մենք կարող ենք խողովակաշար կառուցել

await Generate(100)
          .Double()
          .Square()
          .Print();

Եվ բոլոր քայլերին ավելացրեք չեղարկման նշան.

using var cts=new CancellationTokenSource();
await Generate(100,cts.Token)
          .Double(cts.Token)
          .Square(cts.Token)
          .Print(cts.Token);

Հիշողության օգտագործումը կարող է աճել, եթե մեկ քայլով հաղորդագրություններ ստացվեն ավելի արագ, քան դրանք սպառվում են երկար ժամանակ: Սա հեշտությամբ կարգավորվում է՝ օգտագործելով սահմանափակ ալիքի փոխարեն: Այս կերպ, եթե մեթոդը չափազանց դանդաղ է, բոլոր նախորդ մեթոդները պետք է սպասեն մինչև նոր տվյալներ հրապարակելը:

Ես կարողացա վերարտադրել ձեր դիտարկած խնդիրը։ Դա ակնհայտորեն թերություն է Ալիքների գրադարանում: IMHO. Ահա իմ ռեպո.

using System;
using System.Diagnostics;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

public static class Program
{
    public static async Task Main()
    {
        var channel = Channel.CreateUnbounded<int>();
        var bufferBlock = new BufferBlock<int>();
        var asyncCollection = new Nito.AsyncEx.AsyncCollection<int>();
        var mem0 = GC.GetTotalMemory(true);
        int timeouts = 0;
        for (int i = 0; i < 10; i++)
        {
            var stopwatch = Stopwatch.StartNew();
            while (stopwatch.ElapsedMilliseconds < 500)
            {
                using var cts = new CancellationTokenSource(1);
                try
                {
                    await channel.Reader.ReadAsync(cts.Token);
                    //await bufferBlock.ReceiveAsync(cts.Token);
                    //await asyncCollection.TakeAsync(cts.Token);
                }
                catch (OperationCanceledException) { timeouts++; }
            }
            var mem1 = GC.GetTotalMemory(true);
            Console.WriteLine($"{i + 1,2}) Timeouts: {timeouts,5:#,0},"
                + $" Allocated: {mem1 - mem0:#,0} bytes");
        }
    }
}

Արդյունք:

 1) Timeouts:   124, Allocated: 175,664 bytes
 2) Timeouts:   250, Allocated: 269,720 bytes
 3) Timeouts:   376, Allocated: 362,544 bytes
 4) Timeouts:   502, Allocated: 453,264 bytes
 5) Timeouts:   628, Allocated: 548,080 bytes
 6) Timeouts:   754, Allocated: 638,800 bytes
 7) Timeouts:   880, Allocated: 729,584 bytes
 8) Timeouts: 1,006, Allocated: 820,304 bytes
 9) Timeouts: 1,132, Allocated: 919,216 bytes
10) Timeouts: 1,258, Allocated: 1,011,928 bytes

Փորձեք այն Fiddle-ում:

Մեկ գործողության ընթացքում արտահոսում է մոտ 800 բայթ, ինչը բավականին տհաճ է: Հիշողությունը վերականգնվում է ամեն անգամ, երբ ալիքում նոր արժեք է գրվում, ուստի զբաղված ալիքի համար այս դիզայնի թերությունը չպետք է խնդիր լինի: Բայց մի ալիքի համար, որը ժամանակ առ ժամանակ արժեքներ է ստանում, սա կարող է լինել ցուցադրական:

Կան այլ ասինխրոն հերթերի իրականացումներ, որոնք չեն տառապում նույն խնդրից: Կարող եք փորձել մեկնաբանել await channel.Reader.ReadAsync(cts.Token); տողը և չմեկնաբանել ստորև նշված երկու տողերից որևէ մեկը: Դուք կտեսնեք, որ և՛ BufferBlock<T>< /a> TPL Dataflow գրադարանը և AsyncCollection<T> Nito.AsyncEx.Coordination փաթեթից, թույլ տվեք ասինխրոն որոնում հերթը՝ ընդմիջումով, առանց հիշողության արտահոսքի։

Ես այնքան զբաղված էի բուն խնդրի տեխնիկական մանրամասներով, ես մոռացել էի, որ խնդիրն արդեն լուծված է գրեթե առանց ներբեռնման:

Մեկնաբանություններից թվում է, թե իրական հարցը հետևյալն է.

կա պրոցեսոր՝ հաղորդագրությունները խմբաքանակով մշակելու համար: այն սկսում է մշակվել, երբ բավականաչափ հաղորդագրություններ կան կամ ժամանակն ավարտվում է, այստեղ է հայտնվում ժամանակի չեղարկումը

Սա տրամադրված է Բուֆեր ReactiveX.NET, որը ստեղծվել է նույն թիմի կողմից, որը ստեղծում է System.Linq.Async :

ChannelReader<Message> reader=_channel;

IAsyncEnumerable<IList<Message>> batchItems = reader.ReadAllAsync(token)
                                              .ToObservable()
                                              .Buffer(TimeSpan.FromSeconds(30), 5)
                                              .ToAsyncEnumerable();

await foreach(var batch in batchItems.WithCancellation(token))
{
 ....
}

Այս զանգերը կարող են վերածվել ընդլայնման մեթոդի, ուստի DequeueAsync-ի փոխարեն հարցի դասը կարող է ունենալ BufferAsync կամ GetWorkItemsAsync մեթոդ.

public IAsyncEnumerable<T[]> BufferAsync(
            TimeSpan timeSpan,
            int count,
            CancellationToken cancellationToken = default)
{
    return _channel.Reader.BufferAsync(timeSpan,count,cancellationToken);
}

ToObservable-ը և ToAsyncEnumerable-ը տրամադրվում են System.Linq.Async-ի կողմից և փոխակերպվում են IAsyncEnumerable-ի և IObservable, ինտերֆեյսը, որն օգտագործվում է ReactiveX.NET-ի կողմից:

Բուֆեր տրամադրվում է System.Reactive և բուֆերացնում է տարրը ըստ քանակի կամ կետի, նույնիսկ թույլ տալով համընկնող հաջորդականությունները:

Մինչ LINQ-ը և LINQ to Async-ը տրամադրում են հարցումների օպերատորներ օբյեկտների վրա, Rx.NET-ն անում է նույնը իրադարձությունների ժամանակի վրա հիմնված հոսքերի նկատմամբ: Հնարավոր է ժամանակի ընթացքում համախմբել իրադարձությունները, բուֆերացնել իրադարձություններն ըստ ժամանակի, դրանք կարգավորել և այլն: Օրինակները փաստաթղթի (ոչ պաշտոնական) էջում Բուֆեր ցույց է տալիս, թե ինչպես ստեղծել համընկնող հաջորդականություններ (օրինակ՝ լոգարիթմական պատուհաններ): Նույն էջը ցույց է տալիս, թե ինչպես Sample-ը կամ Throttle-ը կարող են օգտագործվել արագ իրադարձությունների հոսքերը ճնշելու համար՝ տարածելով միայն վերջին իրադարձությունը տվյալ ժամանակահատվածում:

Rx-ն օգտագործում է push մոդելը (նոր իրադարձությունները հրվում են բաժանորդներին), մինչդեռ IAsyncEnumerable-ը, ինչպես IEnumerable-ը, օգտագործում է ձգողական մոդել: ToAsyncEnumerable()-ը կպահի տարրերը, քանի դեռ դրանք չեն պահանջվում, ինչը կարող է հանգեցնել խնդիրների, եթե ոչ ոք չլսի:

Այս մեթոդներով կարելի է նույնիսկ ընդլայնման մեթոդներ ստեղծել հրատարակիչներին բուֆերացնելու կամ շնչափողելու համար.

    //Returns all items in a period
    public static IAsyncEnumerable<IList<T>> BufferAsync<T>(
        this ChannelReader<T> reader, 
        TimeSpan timeSpan, 
        int count,
        CancellationToken token = default)
    {
        return reader.ReadAllAsync(token)
            .ToObservable()
            .Buffer(timeSpan, count)
            .ToAsyncEnumerable();
    }
        
        
    //Return the latest item in a period
    public static IAsyncEnumerable<T> SampleAsync<T>(
        this ChannelReader<T> reader, 
        TimeSpan timeSpan,
        CancellationToken token = default)
    {
        return reader.ReadAllAsync(token)
            .ToObservable()
            .Sample(timeSpan)
            .ToAsyncEnumerable();
    }