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");

to11mtm

using System;
using System.Collections.Generic;
using System.Net.Http;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Http.Logging;
using Microsoft.Extensions.Logging;

namespace httplogging
{

to11mtm

    public class ByteStringReadOnlySequenceSegment : ReadOnlySequenceSegment<byte>
    {
        public ByteStringReadOnlySequenceSegment(ReadOnlyMemory<byte> memory, long runningIndex)
        {
            Memory = memory;
            RunningIndex = runningIndex;
        }
        public static (ByteStringReadOnlySequenceSegment first, ByteStringReadOnlySequenceSegment last) Create(ByteString bs)
        {
            var bArr = bs.Buffers;