Aszinkron I/O

Az I/O műveletek, mint a fájlok írása/olvasása vagy a hálózati kommunikáció, jóval lassabb még ma is, mint a processzor adatfeldolgozási képessége. Ezért a lassabb, főleg hálózati I/O műveleteket megéri aszinkron módon elvégezni és nem blokkolni a végrehajtó szálat a lassú erőforrásra várakozással.

.NET esetén az I/O műveletek a Stream osztályhoz és leszármazottjaihoz köthetőek és nincs ez másként az aszinkron változatokkal sem. Ez gyakorlatban leegyszerűsítve számunkra azt jelenti, hogy a korábbi „Fájl és könyvtárkezelés” fejezetben tanult metódusok nevét elég kiegészítenünk az Async végződéssel és máris egy Task vagy ValueTask visszatérésű metódust kapunk, amire várakozhatunk az await kulcsszóval:

using System.IO;
using System.Threading.Tasks;

namespace ParallelExceptionHandling;

class Program
{
    static async Task Main(string[] args)
    {
        using (FileStream source = File.OpenRead("source.png"))
        {
            using (FileStream target = File.Create("output.png"))
            {
                byte[] buffer = new byte[1024 * 16];
                int bytesRead = 0;
                do
                {
                    // Aszinkron olvasás
                    bytesRead = await source.ReadAsync(buffer, 0, buffer.Length);
                    // Aszinkron írás
                    await target.WriteAsync(buffer, 0, bytesRead);
                }
                while (bytesRead > 0);
            }
        }
    }
}

Ha saját Stream osztályt írunk, akkor alap esetben elég csak a szinkron metódusokat felülírni, mivel az aszinkron műveletek ezeket hívják meg. Nézzünk is egy példát erre:

internal sealed class RandomStream : Stream
{
    public override bool CanRead => true;
    public override bool CanSeek => true;
    public override bool CanWrite => false;
    public override long Length => long.MaxValue;
    public override long Position { get; set; }

    private Random _randomGenerator;

    public RandomStream(int seed)
    {
        _randomGenerator = new Random(seed);
    }

    // Nem támogatott művelet
    public override void Flush()
        => throw new NotSupportedException();

    public override int Read(byte[] buffer, int offset, int count)
    {
        _randomGenerator.NextBytes(buffer.AsSpan(offset, count));
        return count;
    }

    public override long Seek(long offset, SeekOrigin origin)
    {
        // Üres implementáció, mivel a pozíció kezelése
        // nem szükséges a véletlenszerű adat generálásához
        return Position;
    }

    public override void SetLength(long value)
        => throw new NotSupportedException();

    // Jelen implementációban nem támogatott művelet
    public override void Write(byte[] buffer, int offset, int count)
        => throw new NotSupportedException();
}

A fenti példakód azért működik, mert a Stream osztály aszinkron metódusai a szinkron változatokat csomagolják be aszinkron környezetbe alapértelmezetten. Az Async végződésű metódusok virtuálisként vannak definiálva és külön is felülírhatóak. Általában annak nincs értelme, hogy külön-külön viselkedést implementáljunk le, de adódhat a szituáció, például ha egy olyan Stream-et szeretnénk implementálni, ami képes limitálni az írási és olvasási sebességét.

Ebben a specifikus példában a szinkron és aszinkron metódusokat is felül kell írnunk, mégpedig azért, mert ha az aszinkron változatok a szinkron változatokat hívnák meg, amiben Thread.Sleep megvalósítással várakozunk, akkor azzal kreálnánk egy deadlock-ot és végső soron nem úgy működne az alkalmazásunk, ahogy szeretnénk. Egy ilyen Stream osztály lehetséges implementációja:

using System;
using System.IO;
using System.Threading;
using System.Threading.Tasks;

public class RateLimitedStream : Stream
{
    private readonly Stream _innerStream;
    private readonly int _readLimitBytesPerSec;
    private readonly int _writeLimitBytesPerSec;

    private DateTime _nextReadAllowed = DateTime.UtcNow;
    private DateTime _nextWriteAllowed = DateTime.UtcNow;

    public RateLimitedStream(Stream innerStream, int readLimitBytesPerSec, int writeLimitBytesPerSec)
    {
        _innerStream = innerStream ?? throw new ArgumentNullException(nameof(innerStream));
        _readLimitBytesPerSec = readLimitBytesPerSec;
        _writeLimitBytesPerSec = writeLimitBytesPerSec;
    }

    public override bool CanRead => _innerStream.CanRead;
    public override bool CanSeek => _innerStream.CanSeek;
    public override bool CanWrite => _innerStream.CanWrite;
    public override long Length => _innerStream.Length;
    public override long Position { get => _innerStream.Position; set => _innerStream.Position = value; }

    private async ValueTask ThrottleAsync(int bytes, ref DateTime nextAllowed, int limit, CancellationToken ct)
    {
        if (limit <= 0 || bytes <= 0) return;

        double secondsRequired = (double)bytes / limit;
        nextAllowed = nextAllowed < DateTime.UtcNow ? DateTime.UtcNow : nextAllowed;
        var targetTime = nextAllowed.AddSeconds(secondsRequired);
        var delay = targetTime - DateTime.UtcNow;

        if (delay > TimeSpan.Zero)
            await Task.Delay(delay, ct).ConfigureAwait(false);

        nextAllowed = targetTime;
    }

    private void Throttle(int bytes, ref DateTime nextAllowed, int limit)
    {
        if (limit <= 0 || bytes <= 0) return;

        double secondsRequired = (double)bytes / limit;
        nextAllowed = nextAllowed < DateTime.UtcNow ? DateTime.UtcNow : nextAllowed;
        var targetTime = nextAllowed.AddSeconds(secondsRequired);
        var delay = targetTime - DateTime.UtcNow;

        if (delay > TimeSpan.Zero)
            Thread.Sleep(delay);

        nextAllowed = targetTime;
    }

    public override int Read(byte[] buffer, int offset, int count)
    {
        int bytesRead = _innerStream.Read(buffer, offset, count);
        Throttle(bytesRead, ref _nextReadAllowed, _readLimitBytesPerSec);
        return bytesRead;
    }

    public override void Write(byte[] buffer, int offset, int count)
    {
        _innerStream.Write(buffer, offset, count);
        Throttle(count, ref _nextWriteAllowed, _writeLimitBytesPerSec);
    }

    public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
        int bytesRead = await _innerStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
        await ThrottleAsync(bytesRead, ref _nextReadAllowed, _readLimitBytesPerSec, cancellationToken).ConfigureAwait(false);
        return bytesRead;
    }

    public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
        await _innerStream.WriteAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false);
        await ThrottleAsync(count, ref _nextWriteAllowed, _writeLimitBytesPerSec, cancellationToken).ConfigureAwait(false);
    }

    public override void Flush()
         => _innerStream.Flush();

    public override long Seek(long offset, SeekOrigin origin)
         => _innerStream.Seek(offset, origin);

    public override void SetLength(long value)
        => _innerStream.SetLength(value);
}