Skip to content

Use a known buffer size to reduce allocations #2306

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
allreadyexists wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Use a known buffer size to reduce allocations #2306

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
120 changes: 120 additions & 0 deletions Rx.NET/Source/src/System.Reactive/Linq/Observable/Buffer.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,9 @@
using System.Collections.Generic;
using System.Reactive.Concurrency;
using System.Reactive.Disposables;
#if NET8_0_OR_GREATER
using System.Runtime.InteropServices;
#endif

namespace System.Reactive.Linq.ObservableImpl
{
Expand Down Expand Up @@ -517,6 +520,122 @@ public Ferry(IObservable<TSource> source, TimeSpan timeSpan, int count, ISchedul

protected override void Run(_ sink) => sink.Run();

#if NET8_0_OR_GREATER

@idg10 idg10 Apr 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The conditional compilation here is at what seems like an unnecessarily coarse level. It makes it very hard to see what's actually different between the #if and #else arms.

As far as I can tell there are only 3 lines that are different: 1) you've added a new _count field, 2) you've used that to reserve capacity when initializing _s and 3) also when re-initializing it after a buffer fills.

If I have missed other changes, then that illustrates my point.

I would want the conditional compilation directives to wrap only those parts that are different, because otherwise, anyone maintaining this code is forced to read a lot of code to try to deduce why there are two different versions.

The unnecessary duplication also makes it more likely that if bugfixes or further performance enhancements are made in this code, that they are accidentally applied only to one of the two copies of the code.

internal sealed class _ : Sink<TSource, IList<TSource>>
{
private readonly Ferry _parent;
private readonly int _count;
private readonly object _gate = new();
private List<TSource> _s;

public _(Ferry parent, IObserver<IList<TSource>> observer)
: base(observer)
{
_parent = parent;
_count = _parent._count;
_s = new List<TSource>(_count);
}

private SerialDisposableValue _timerSerial;
private int _n;
private int _windowId;

public void Run()
{
_n = 0;
_windowId = 0;

CreateTimer(0);

SetUpstream(_parent._source.SubscribeSafe(this));
}

protected override void Dispose(bool disposing)
{
if (disposing)
{
_timerSerial.Dispose();
}

base.Dispose(disposing);
}

private void CreateTimer(int id)
{
var m = new SingleAssignmentDisposable();
_timerSerial.Disposable = m;

m.Disposable = _parent._scheduler.ScheduleAction((@this: this, id), _parent._timeSpan, static tuple => tuple.@this.Tick(tuple.id));
}

private void Tick(int id)
{
lock (_gate)
{
if (id != _windowId)
{
return;
}

var res = _s;
CollectionsMarshal.SetCount(res, _n);
_n = 0;
var newId = ++_windowId;

_s = new List<TSource>(_count);
ForwardOnNext(res);

CreateTimer(newId);
}
}

public override void OnNext(TSource value)
{
var newWindow = false;
var newId = 0;

lock (_gate)
{
_s.Add(value);

_n++;
if (_n == _parent._count)
{
newWindow = true;
_n = 0;
newId = ++_windowId;

var res = _s;
_s = new List<TSource>(_count);

@idg10 idg10 Apr 27, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If I understand correctly, this is basically the whole point of this change: it sets the capacity of the List<TSource> so that the list will never have to reallocate as the buffer contents grow.

If that's right then I don't understand why you made this change apply only to .NET 8.0 or greater. As far as I can remember that constructor has been available since .NET Framework 2.0 (i.e., for over 20 years). It's available in netstandard2.0, net472, and uap10.0.18362, i.e. all of the frameworks we target.

@allreadyexists allreadyexists Jun 4, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

CollectionsMarshal.SetCount available in .NET 8.0 or greater. This is necessary to specify the correct size of the list.

ForwardOnNext(res);
}

if (newWindow)
{
CreateTimer(newId);
}
}
}

public override void OnError(Exception error)
{
lock (_gate)
{
_s.Clear();
ForwardOnError(error);
}
}

public override void OnCompleted()
{
lock (_gate)
{
ForwardOnNext(_s);
ForwardOnCompleted();
}
}
}
#else
internal sealed class _ : Sink<TSource, IList<TSource>>
{
private readonly Ferry _parent;
Expand Down Expand Up @@ -627,6 +746,7 @@ public override void OnCompleted()
}
}
}
#endif
}
}

Expand Down