to11mtm

Transport	Serializer	Payload	Best Msgs/sec	At Clients
Pipe/TCP + Protobuf + ZeroCopy	JSON (default)	Primitive long	1,257,862	15
Pipe/TCP + Protobuf	JSON (default)	Primitive long	1,255,493	10
Pipe/TCP + Protobuf + ZeroCopy	MessagePack (typeless)	Custom object (serializer path)	1,026,168	10
Pipe/TCP + Protobuf	MessagePack (typeless)	Custom object (serializer path)	1,019,368	5
DotNetty/TCP + Protobuf	JSON (default)	Primitive long	971,110	20
DotNetty/TCP + Protobuf	MessagePack (typeless)	Custom object (serializer path)	948,767	20
Pipe/TCP + Protobuf + ZeroCopy	Hyperion	Custom object (serializer path)	926,785	15
Pipe/TCP + Protobuf	Hyperion	Custom object (serializer path)	921,376	15

to11mtm

#nullable enable
using System;
using System.Buffers;
using System.Buffers.Binary;

namespace Akka.Remote.Transport.Pipelines;

public interface IPrependableBufferWriter : IBufferWriter<byte>, IDisposable
{
    Memory<byte> WrittenMemoryRange { get; }

to11mtm

Now I can see the full picture from the new benchmark data. Let me analyze carefully:

• DrainOnly:

100 items: New wins (14.8µs vs 19.4µs) ✅

1000 items, MaxWeight=1: New wins (110µs vs 136µs) ✅

10000 items, MaxWeight=1: New LOSES (651µs vs 342µs) ❌ — 2x

to11mtm

// -----------------------------------------------------------------------
//  <copyright file="ChannelQueueWithBatch.cs" company="Akka.NET Project">
//      Copyright (C) 2013-2023 .NET Foundation <https://github.com/akkadotnet/akka.net>
//  </copyright>
// -----------------------------------------------------------------------

using System;
using System.Threading;
using System.Threading.Channels;
using System.Threading.Tasks;

to11mtm

using Akka.Annotations;
using Akka.Streams.Dsl;
using Akka.Streams.Implementation.Stages;
using Akka.Streams.Stage;
using Akka.Streams.Supervision;
using Akka.Util;
using Akka.Util.Internal;

namespace Akka.Streams.NATS.KeyValueStore.AkkaStreamsExtensions;

to11mtm

using System.Runtime.CompilerServices;
using System.Runtime.Serialization;

namespace derp;

class Program
{
    static void Main(string[] args)
    {
        var obj = new object();

to11mtm

using System.Text;
using System.Threading.Tasks.Dataflow;
using Akka;
using Akka.Streams;
using Akka.Streams.Dsl;
using Akka.Streams.Implementation.Fusing;
using Akka.Util;
using JetBrains.Annotations;
using NATS.Client.Core;
using NATS.Client.JetStream;

to11mtm

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Linq2Db.EfCore.Samples.Tests.Microsoft.EntityFrameworkCore.Infrastructure;
using LinqToDB.Data;
using LinqToDB.EntityFrameworkCore;
using LinqToDB.Mapping;

to11mtm

//// THIS IS PROBABLY NOT QUITE WORKING IN A FEW WAYS, JUST A CONVERSATION STARTER AT THIS POINT    

    /// <summary>
    /// INTERNAL API
    /// </summary>
    /// <typeparam name="TIn">TBD</typeparam>
    /// <typeparam name="TOut">TBD</typeparam>
    [InternalApi]
    public sealed class UnfoldAsync<TIn, TUnfoldState, TOut> : GraphStage<FlowShape<TIn, TOut>>
    {

to11mtm

public static class WatCache<T> where T : class
{
    private static ConcurrentDictionary<object, Action<T,Guid>> _cache = new();

    public static Action<T, Guid> Get(Expression<Func<T, object>> selector) => _cache.GetOrAdd(selector, sel =>
    {
        var expr = (Expression<Func<T, object>>)sel;
        var member = expr.Body as MemberExpression;
        var prop = member!.Member as PropertyInfo;
        var ep = Expression.Parameter(typeof(T), "entity");