2
\$\begingroup\$

Imagine I have connection in which I have to write numbers, represented by this contract:

interface ISocket
{
    Task WriteAsync(Byte[], Int32 offset, Int32 count, CancellationToken cancel);
}

There is one single thread calling this WriteAsync method.

Till now each caller to ISocket was buffering independently and knows when it has to flush, however I would like to centralize that logic. Also I want to make sure that when such buffer is being transmitted, I can take advantage of async/await and release the thread pool thread.

My idea would be to create something like this:

class SocketWriter
{
    private ISocket _socket;
    private Byte[] _buffer; // assume 8K
    private Int32 _cursor;

    [ ... ]

    private async Task FlushCacheIfNotEnoughSpace(Int32 length, CancellationToken cancel)
    {
        if(!FitsInBuffer(length))
        {
            await _socket.WriteAsync(_buffer, 0, _cursor, cancel).ConfigureAwait(false);
            _cursor = 0;
        }
    }

    [ ... ]

    public async Task WriteAsync(UInt64 value, CancellationToken cancel)
    {
        await FlushCacheIfNotEnoughSpace(8, cancel).ConfigureAwait(false);
        CopyToByteArray(value);
    }

    public Task Flush(CancellationToken cancel)
    {
        return FlushCacheIfAny(cancel);
    }
}

There are some memory constraints, so I cannot create more than one of those Byte[] buffers. CopyToByteArray requires enough space in the _buffer and increments _cursor accordingly.

Note that FlushCacheIfNotEnoughSpace only awaits if the given length does not fit the current buffer.

My intention is that when the buffer is not complete and there is space for another number this method call executes synchronously, or properly awaits using IO completion ports otherwise:

await _writer.WriteAsync(1UL, cancel).ConfigureAwait(false);

I wonder if there is a better or more elegant way of doing this, because it seems like a lot of "noise":

await _writer.WriteAsync(obj.Member1, cancel).ConfigureAwait(false);
await _writer.WriteAsync(obj.Member2, cancel).ConfigureAwait(false);
await _writer.WriteAsync(obj.Member3, cancel).ConfigureAwait(false);
await _writer.WriteAsync(obj.Member4, cancel).ConfigureAwait(false);
\$\endgroup\$
  • \$\begingroup\$ every time you write you immediately await. That's not a good design. \$\endgroup\$ – ratchet freak Feb 2 '16 at 16:54
  • \$\begingroup\$ It only awaits when the buffer is full, rest of times it executes synchronously. \$\endgroup\$ – vtortola Feb 2 '16 at 16:59
  • \$\begingroup\$ but the only time you ever call WriteAsync is when doing any kind of flush. \$\endgroup\$ – ratchet freak Feb 2 '16 at 17:02
  • \$\begingroup\$ It is flushing only if the buffer is full. Considering a buffer of 8K, I should write 1001 UInt64 in order to flush the first 1000. \$\endgroup\$ – vtortola Feb 2 '16 at 17:08
1
\$\begingroup\$

OK, now it is SOLID :)

Let's do not mix infrastructure and application code. Unfortunatly a lot of stuff is missing in .NET, so infrastructure first.

AsyncStream:

public interface IAsyncStream
{
    Task WriteAsync(
        byte[] buffer, 
        int offset, 
        int count, 
        CancellationToken cancellationToken);        

    Task FlushAsync(CancellationToken cancellationToken);
}

BufferedAsyncStream:

public class BufferedAsyncStream : IAsyncStream
{
    public BufferedAsyncStream(IAsyncStream inner, long bufferSize = 8192)
    {
        Inner = inner;
        Buffer = new byte[bufferSize];
        BufferStream = new MemoryStream(Buffer);
    }

    public async Task WriteAsync(
         byte[] buffer, 
         int offset, 
         int count, 
         CancellationToken cancellationToken)
    {
        if (FreeSpace < count)
            await FlushAsync(cancellationToken)
                .ConfigureAwait(false);

        BufferStream.Write(buffer, offset, count);
    }

    public async Task FlushAsync(CancellationToken cancellationToken)
    {
        await Inner.WriteAsync(Buffer, 0, Allocated, cancellationToken)
            .ConfigureAwait(false);

        await Inner.FlushAsync(cancellationToken)
            .ConfigureAwait(false);

        BufferStream.Seek(0, SeekOrigin.Begin);
    }

    int FreeSpace => (int)BufferStream.Length - (int)BufferStream.Position;
    int Allocated => (int)BufferStream.Position;
    IAsyncStream Inner { get; }
    byte[] Buffer { get; }
    MemoryStream BufferStream { get; }
}

AsyncWriter:

public class AsyncWriter
{
    public AsyncWriter(IAsyncStream stream)
    {
        Stream = stream;
    }

    public Task WriteAsync(ulong value, CancellationToken cancellationToken)
    {
        var buffer = BitConverter.GetBytes(value);
        return Stream.WriteAsync(buffer, 0, buffer.Length, cancellationToken);
    }

    public Task FlushAsync(CancellationToken cancellationToken)
    {
        return Stream.FlushAsync(cancellationToken);
    }

    IAsyncStream Stream { get; }
}

Now application code. ISocket being adapted to IAsyncStream:

class SocketStream : IAsyncStream
{
    public SocketStream(ISocket socket)
    {
        Socket = socket;
    }

    public Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
    {
        return Socket.WriteAsync(buffer, offset, count, cancellationToken);
    }

    public Task FlushAsync(CancellationToken cancellationToken)
    {
        return Task.CompletedTask;
    }

    ISocket Socket { get; }        
}

So, we can construct writer in the following way:

var writer = new AsyncWriter(
    new BufferedAsyncStream(
        new SocketStream(socket)));

UPDATE You could use this overload to remove "noise":

public class AsyncWriter
{
     ...
    public async Task WriteAsync(
        CancellationToken cancellationToken, params ulong[] values)
    {
        foreach (var value in values)
            await WriteAsync(value, cancellationToken)
                .ConfigureAwait(false);
    }

Now it is this way:

  await writer
     .WriteAsync(
         cancellationToken, 
         obj.Member1, obj.Member2, obj.Member3, obj.Member4)
     .ConfigureAwait(false);
\$\endgroup\$
  • \$\begingroup\$ Lovely :) However the question is about how to avoid the "noise" at the end of my post. Your AsyncWriter has to be used exactly in the same way, and that is what I was wondering if could be avoided somehow. \$\endgroup\$ – vtortola Feb 3 '16 at 10:48
  • \$\begingroup\$ See an update above. \$\endgroup\$ – Dmitry Nogin Feb 3 '16 at 20:54
  • 1
    \$\begingroup\$ I like the idea. Cheers. \$\endgroup\$ – vtortola Feb 5 '16 at 16:04

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.