10

我正在编写一个 C# (.NET 4.5) 应用程序，用于聚合基于时间的事件以进行报告。为了使我的查询逻辑可重用于实时和历史数据，我使用了 Reactive Extensions (2.0) 及其IScheduler基础设施（HistoricalScheduler和朋友）。

例如，假设我们创建了一个事件列表（按时间顺序排序，但它们可能重合！），其唯一的有效负载是它们的时间戳，并且想知道它们在固定持续时间的缓冲区中的分布：

const int num = 100000;
const int dist = 10;

var events = new List<DateTimeOffset>();
var curr = DateTimeOffset.Now;
var gap = new Random();

var time = new HistoricalScheduler(curr);

for (int i = 0; i < num; i++)
{
    events.Add(curr);
    curr += TimeSpan.FromMilliseconds(gap.Next(dist));
}

var stream = Observable.Generate<int, DateTimeOffset>(
    0,
    s => s < events.Count,
    s => s + 1,
    s => events[s],
    s => events[s],
    time);

stream.Buffer(TimeSpan.FromMilliseconds(num), time)
    .Subscribe(l => Console.WriteLine(time.Now + ": " + l.Count));

time.AdvanceBy(TimeSpan.FromMilliseconds(num * dist));

运行此代码会产生System.StackOverflowException以下堆栈跟踪（一直是最后 3 行）：

mscorlib.dll!System.Threading.Interlocked.Exchange<System.IDisposable>(ref System.IDisposable location1, System.IDisposable value) + 0x3d bytes    
System.Reactive.Core.dll!System.Reactive.Disposables.SingleAssignmentDisposable.Dispose() + 0x37 bytes    
System.Reactive.Core.dll!System.Reactive.Concurrency.ScheduledItem<System.DateTimeOffset>.Cancel() + 0x23 bytes    
...
System.Reactive.Core.dll!System.Reactive.Disposables.AnonymousDisposable.Dispose() + 0x4d bytes    
System.Reactive.Core.dll!System.Reactive.Disposables.SingleAssignmentDisposable.Dispose() + 0x4f bytes    
System.Reactive.Core.dll!System.Reactive.Concurrency.ScheduledItem<System.DateTimeOffset>.Cancel() + 0x23 bytes    
...

好的，问题似乎来自我Observable.Generate()对 的使用，具体取决于列表大小（num）并且无论调度程序的选择如何。

我究竟做错了什么？IObservable或者更一般地说，从IEnumerable提供自己时间戳的事件中创建事件的首选方法是什么？

4

2 回答 2

4

（更新 - 意识到我没有提供替代方案：见答案底部）

问题在于如何Observable.Generate工作 - 它用于根据参数展开核心递归（认为递归由内而外）生成器；如果这些参数最终生成了一个非常嵌套的 corecursive 生成​​器，那么你会炸毁你的堆栈。

从这一点开始，我推测了很多（我面前没有 Rx 源）（见下文），但我敢打赌你的定义最终会扩展为：

initial_state =>
generate_next(initial_state) => 
generate_next(generate_next(initial_state)) => 
generate_next(generate_next(generate_next(initial_state))) =>
generate_next(generate_next(generate_next(generate_next(initial_state)))) => ...

一直持续到你的调用堆栈变得足够大而溢出。比如说，一个方法签名 + 你的 int 计数器，每次递归调用大约需要 8-16 个字节（更多取决于状态机生成器的实现方式），所以 60,000 听起来是正确的（最大 1M / 16 ~ 62500深度）

编辑：提取源代码-确认：生成的“运行”方法如下所示-注意嵌套调用Generate：

protected override IDisposable Run(
    IObserver<TResult> observer, 
    IDisposable cancel, 
    Action<IDisposable> setSink)
{
    if (this._timeSelectorA != null)
    {
        Generate<TState, TResult>.α α = 
                new Generate<TState, TResult>.α(
                     (Generate<TState, TResult>) this, 
                     observer, 
                     cancel);
        setSink(α);
        return α.Run();
    }
    if (this._timeSelectorR != null)
    {
        Generate<TState, TResult>.δ δ = 
               new Generate<TState, TResult>.δ(
                   (Generate<TState, TResult>) this, 
                   observer, 
                   cancel);
        setSink(δ);
        return δ.Run();
    }
    Generate<TState, TResult>._ _ = 
             new Generate<TState, TResult>._(
                  (Generate<TState, TResult>) this, 
                  observer, 
                  cancel);
    setSink(_);
    return _.Run();
}

编辑： Derp，没有提供任何替代方案......这是一个可能有效的方案：

（编辑：固定Enumerable.Range，所以流大小不会乘以chunkSize）

const int num = 160000;
const int dist = 10;

var events = new List<DateTimeOffset>();
var curr = DateTimeOffset.Now;
var gap = new Random();
var time = new HistoricalScheduler(curr);

for (int i = 0; i < num; i++)
{
    events.Add(curr);
    curr += TimeSpan.FromMilliseconds(gap.Next(dist));
}

    // Size too big? Fine, we'll chunk it up!
const int chunkSize = 10000;
var numberOfChunks = events.Count / chunkSize;

    // Generate a whole mess of streams based on start/end indices
var streams = 
    from chunkIndex in Enumerable.Range(0, (int)Math.Ceiling((double)events.Count / chunkSize) - 1)
    let startIdx = chunkIndex * chunkSize
    let endIdx = Math.Min(events.Count, startIdx + chunkSize)
    select Observable.Generate<int, DateTimeOffset>(
        startIdx,
        s => s < endIdx,
        s => s + 1,
        s => events[s],
        s => events[s],
        time);

    // E pluribus streamum
var stream = Observable.Concat(streams);

stream.Buffer(TimeSpan.FromMilliseconds(num), time)
    .Subscribe(l => Console.WriteLine(time.Now + ": " + l.Count));

time.AdvanceBy(TimeSpan.FromMilliseconds(num * dist));
于 2012-11-20T21:54:50.493 回答
3

好的，我采用了一种不同的工厂方法，它不需要 lamdba 表达式作为状态转换，现在我不再看到任何堆栈溢出。我还不确定这是否可以作为我问题的正确答案，但它有效，我想我会在这里分享：

var stream = Observable.Create<DateTimeOffset>(o =>
    {
        foreach (var e in events)
        {
            time.Schedule(e, () => o.OnNext(e));
        }

        time.Schedule(events[events.Count - 1], () => o.OnCompleted());

        return Disposable.Empty;
    });

在（！）返回订阅之前手动安排事件对我来说似乎很尴尬，但在这种情况下，它可以在 lambda 表达式中完成。

如果这种方法有任何问题，请纠正我。此外，我仍然很高兴听到System.Reactive我的原始代码违反了哪些隐含假设。

（哦，天哪，我应该早点检查一下：使用 RX v1.0，原来的Observable.Generate()确实似乎工作！）

于 2012-11-20T21:13:17.310 回答