Bnaya Eshet : Queue

New version of Rx Contrib

bnaya — Fri, 14 May 2010 17:09:53 GMT

Rx Contrib has new version (1.3)

Release Notes

- Bug Fix
- BufferWithTimeOrCount with flexible time period enabled
when ever the time period elapsed it will use the lamda for getting the next period

this feature is a join effort of Amir Shitrit and Bnaya Eshet

Code Snippet
 IObservable<long> obs = Observable.Interval(TimeSpan.FromMilliseconds(10));
var bufferQueue = obs.BufferWithTimeOrCount(() => TimeSpan.FromMilliseconds(500), 10);
 

you can download the source or binaries from here.

you can install the Rx (the Rx framework is a prerequisite) from here.

Performance tips

bnaya — Tue, 04 May 2010 23:53:57 GMT

Performance tips

recently I was working on the Reactive Queue (which is part of the Rx Contrib).

the requirement for the Reactive Queue was to to achieve the highest throughput possible

for each queue provider (2,000,000 messages of 500 byte per second using Concurrent Queue provider

running on simple quad server).

while working on this project I was encounter the the following performance hits:

1. ManualResetEventSlim

there is a big difference between using the old ManualResetEvent Vs. ManualResetEventSlim the hit

factor can go up to 2 or 3 times slower if you having a very short synchronization period

(the ManualResetEventSlim does not escalate to kernel mode on short period waiting).

2. ManualResetEventSlim.WaitHandle

do not expose the ManualResetEventSlim instance using the WaitHandle, as shown in the following code snippet:

Code Snippet
 private readonly ManualResetEventSlim _syncWhileWorkingHandle;
public WaitHandle WaitWhileWorkingHandle
{
    get { return _syncWhileWorkingHandle.WaitHandle; }
}
 

consuming ManualResetEventSlim though its WaitHandle property does not gain

performance improvements benefits.

if you do want to expose synchronization API, use the following code:

Code Snippet
 private readonly ManualResetEventSlim _syncWhileWorkingHandle;
public void WaitWhileWorking()
{
    _syncWhileWorkingHandle.Wait();
}
 

3. CancellationToken

when ever using CancellationToken:

Code Snippet
 var cancel = new CancellationTokenSource();
Task.Factory.StartNew(Dequeuing, cancel.Token);
 

do monitor GC collect counter, especially the 3rd generation collection.

trying to use this technique I got a huge hit (3.5 times slower), because of

extra (multiple) GC 3rd generation collection.

instead of using the CancellationToken the Reactive Queue is using old good

dispose pattern.

4. Thread.Yield()

do consider to use Thread.Yield().

putting it on the right places may improve overall performance,

but you have to be careful and measure the actual performance, because

doing it on the wrong places may result as performance hit.

Summary

performance improving required continuously measurement which should

be compared with historical measurement.

be careful with synchronization and do use the new .net synchronization types like ManualResetEventSlim.

Rx Contrib

bnaya — Sat, 01 May 2010 22:57:52 GMT

Rx Contrib release V1

Rx Contrib is ongoing effort of community extensions for Rx.

you can can find the project at http://rxcontrib.codeplex.com/

Current features are:

ReactiveQueue: is ISubject that does not loose values if there are no subscribers.

The reactive queue is using the IQueueAdapter for injecting concrete

queue adapters (which mean that it can be extend to support other queues).

the out of the box implementation is using ConcurrentQueue which have extreme throughput.

(up to 2,000,000 messages of 500 byte per second on basic quad server).

It is also support Weak Subscribe option which holds the subscriber using week reference,

(those subscription will be removed after it been collected by the GC).

Add contribution

If you want to contribute to this efforts, you can contact bnayae@sela.co.il

Credits

The Reactive Queue, Weak Subscription, Unsubscribe Disposable are join effort of Bnaya Eshet and Yuval Mazor.

Yuval Mazor blog: http://blogs.microsoft.co.il/blogs/yuvmaz/

Bnaya Eshet : Queue

New version of Rx Contrib

Rx Contrib has new version (1.3)

Release Notes

Performance tips

Performance tips

1. ManualResetEventSlim

2. ManualResetEventSlim.WaitHandle

3. CancellationToken

4. Thread.Yield()

Summary

Rx Contrib

Rx Contrib release V1

Add contribution

Credits

Read more about Rx