esper - win:time のウィンドウを使用すると、Nesper の CPU/メモリ使用量が高くなる

Question

アプリケーションでイベントを処理するための CEP エンジンとしてNesperを使用しています。

次の EPL ステートメントをモデル化しようとしています。

• イベントのValueフィールドは、限られたウィンドウ時間にわたって平均化されます。
• これらの平均のいずれかが1目的の値と一致した場合、イベントが生成されます。

私はこれを次のようにモデル化しました：

SELECT (
(AVG(ParameterEvent1.Value) = 1) OR 
(AVG(ParameterEvent2.Value) = 1)
...
(AVG(ParameterEvent50.Value) = 1)
) AS BooleanValue 
FROM 
ParameterEvent(Id = 1).win:time(3 sec) AS ParameterEvent1, 
ParameterEvent(Id = 2).win:time(3 sec) AS ParameterEvent2 
...
ParameterEvent(Id = 50).win:time(3 sec) AS ParameterEvent50

別のスレッドで、 50 イベント/秒の安定した速度でパラメーター値をエンジンに供給しています。

com.espertech.esper.client.EPException: ReaderWriterLock timeout expiredこの設定により、例外が発生するまで CPU と RAM が大量に使用されます。

何が問題を引き起こしているのだろうか。win:time(3 sec)ウィンドウを からに変更すると、std:lastevent()この問題が解決するのは興味深いことです。ただし、イベントを生成する前にパラメーターの値が13 秒であることを確認するためのウィンドウが必要です。

完全なデモ コードは次のとおりです。Nesperパッケージをインストールして実行するだけです：

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using com.espertech.esper.client;

namespace NesperTester
{
    class Program
    {
        public class ParameterEvent
        {
            public int Id { get; set; }
            public int Value { get; set; }
        }

        static EPServiceProvider engine;

        const int NParameters = 50;

        static void Main(string[] args)
        {
            engine = EPServiceProviderManager.GetDefaultProvider();

            engine.EPAdministrator.Configuration.AddEventType("ParameterEvent", typeof(ParameterEvent));

            CreateEPLStatement();

            Task.Factory.StartNew(() => SimulateParameterChange(), TaskCreationOptions.LongRunning);

            System.Threading.Thread.Sleep(int.MaxValue);
        }

        static void CreateEPLStatement()
        {
            const int windowLength = 3;

            var sb = new StringBuilder();

            for (int i = 1; i <= NParameters; i++)
            {
                sb.AppendFormat("(AVG(ParameterEvent{0}.Value) = 1) OR ", i);
            }

            var rule = sb.ToString();
            rule = rule.Remove(rule.Length - 4, 4);
            sb.Clear();

            for (int paramId = 1; paramId <= NParameters; paramId++)
            {
                sb.AppendFormat("ParameterEvent(Id = {0}).win:time({1} sec) AS ParameterEvent{0}, ", paramId, windowLength);
                //sb.AppendFormat("ParameterEvent(Id = {0}).std:lastevent() AS ParameterEvent{0}, ", paramId, windowLength);
            }

            var selectSources = sb.ToString();
            selectSources = selectSources.Remove(selectSources.Length - 2, 2);

            var eplStr = string.Format("SELECT ({0}) AS BooleanValue FROM {1}", rule, selectSources);

            var statementName = string.Format("statement1");

            var statement = engine.EPAdministrator.CreateEPL(eplStr, statementName, null);
            statement.Start();
            statement.Events += Statement_Events;
        }


        static bool PrevState = false;
        static void Statement_Events(object sender, UpdateEventArgs e)
        {
            var underlying = e.NewEvents[0].Underlying as Dictionary<string, object>;

            if (underlying.Last().Value == null)
            {
                Console.WriteLine("Statement value is unknown");
            }
            else
            {
                var statementValue = (bool)underlying.First().Value;

                if (statementValue != PrevState)
                {
                    PrevState = statementValue;
                    Console.WriteLine("Statement is {0}", statementValue);
                }
            }
        }

        static void SimulateParameterChange()
        {
            int counter = 0;

            while (true)
            {
                for (int i = 1; i <= NParameters; i++)
                {
                    var evt = new ParameterEvent();

                    evt.Id = i;

                    if (i == 1 && (counter/4) % 2 == 0)
                    {
                        evt.Value = 1;
                    }
                    else
                    {
                        evt.Value = 0;
                    }

                    engine.EPRuntime.SendEvent(evt);

                    System.Threading.Thread.Sleep(1000 / NParameters);
                }

                counter++;
            }
        }
    }
}