Caching of OMMStreamCache objects using a lot of memory

cory.schmidt.1

We have a use case where we allow users to scan market data for instruments that meet certain criteria. These scans can request historical data snapshots across 1000-2000 instruments or more. Memory profiling shows that after performing a scan of almost 2000 instruments the OMMConnection object was using 158MB of memory (see attached screenshot from dotMemory).

Is there a way to not cache OMMStreamCache objects if it is a snapshot request, or could the amount of data held in memory be reduced? Here is our code when making a historical snapshot request...

        public IObservable<WorkspaceSnapShotStreamingResponse> GetTimeSeriesDataSnapShot(
            ISession workspaceSession,
            WorkspaceSnapShotStreamingRequest request,
            WorkspaceMetadataResponse metadataResponse)
        {
            return Observable.Create<WorkspaceSnapShotStreamingResponse>(observer =>
            {
                SerialDisposable activeSubscriptionToken = new SerialDisposable();
                if (workspaceSession.OpenState == Session.State.Opened)
                {
                    // Build the request
                    var summariesDefinition = BuildRequest(request);

                    // Make the request
                    activeSubscriptionToken.Disposable = summariesDefinition
                        .GetDataAsync(workspaceSession)
                        .ToObservable()
                        .Subscribe(OnSubscribeAction, observer.OnError);

                    // Deal with the response
                    void OnSubscribeAction(IDataSetResponse workspaceResponse)
                    {
                        if (workspaceResponse.Data != null)
                        {
                            if (workspaceResponse.Data.Table != null)
                            {
                                observer.OnNext(ConvertDataTableToWorkspacePriceRecords(
                                    workspaceResponse.Data.Table,
                                    request.Interval,
                                    request.View,
                                    TimeSeriesUpdateType.Update));
                                observer.OnCompleted();
                            }
                            else
                            {
                                Logger.Warn(workspaceResponse.Data.Errors);
                                observer.OnNext(new WorkspaceSnapShotStreamingResponse());
                            }
                        }
                    }
                }

                return activeSubscriptionToken;
            });
        }

        private ISummariesDefinition BuildRequest(WorkspaceSnapShotStreamingRequest request)
        {
            var summariesDefinition = Summaries
                .Definition(request.Symbol)
                .Interval(ResponseHelper.ConvertInterval(request.Interval, Logger))
                .Fields(request.View.MapTo.ToArray());

            if (request.Count > 0)
            {
                summariesDefinition.Count(request.Count);
            }
            else if (request.RequestRange.StartDateTime.HasValue)
            {
                summariesDefinition.Start((DateTime)request.RequestRange.StartDateTime);
            }
            else if (request.RequestRange.EndDateTime.HasValue)
            {
                summariesDefinition.End((DateTime)request.RequestRange.EndDateTime);
            }

            return summariesDefinition;
        }

Find more posts tagged with

historical

time-series

refinitiv-data-platform-libraries

#technology

Accepted answers

nick.zincone

Hi @cory.schmidt.1

If by snapshot you mean a historical pricing snapshot using .GetData(), then my suspicions is that there may be other requests going out within your application the uses .GetStream(), via historical or Pricing. For example, if I were to write a small, standalone example that only uses .GetData() for the historical summaries interface, there would be no streaming connection nor any requests for streaming data in this scenario. If you are seeing the opposite, can you put together a small test example our team can try to replicate? Or, maybe you can try one of the github examples to confirm.

thanks.

All comments

nick.zincone

Hi @cory.schmidt.1

The OMMStreamCache is a container managing open stream IDs and references as opposed to holding a cache of the data objects. If you are asking for snapshots, the OMMStreamCache will clean up the references once a refresh or status has been sent. In theory, the OMMStreamCache should be clean after fulfilling a request. There may be something else dangling within this container.

I can bring it to the attention of the development team to observe.

thanks.

nick.zincone

Hi @cory.schmidt.1

If you want to perform a streaming snapshot, then you should make a call something like this:

var snap = Pricing.Definition("CAD=").Fields("BID", "ASK", "DSPLY_NAME")
                                     .GetStream().Streaming(false);

When you specify .Streaming(false) this will automatically clean up the cache with the details I mentioned above. The default is: .Streaming(true) and this will keep the streams open for every item requested.

Now, if you're mixing this with historical streaming bars, by definition, the historical streaming bars keep the stream open. I don't fully understand what you are trying to retrieve given that a historical snapshot using .getData() already gives you the latest values within the bars. But if your requirement is to retrieve latest realtime values, then you can perform the snapshot based on the code segment I provided above.

cory.schmidt.1

Thanks Nick, we'll give that a try.

cory.schmidt.1

For our use case we need to retrieve historical data (~1250 records) for each one of the ~2000 instruments. It's the historical data request using the Summaries API that is holding open the OMMStreamCache object so setting Streaming to false won't help us much. Is there anything else that might help?

nick.zincone

Hi @cory.schmidt.1

thanks.

cory.schmidt.1

I'm still working on getting a sample put together to illustrate the problem. Stay tuned...

cory.schmidt.1

@nick.zincone You were correct about having requests that were creating subscriptions. We had a section of code checking QoS for each requested instrument creating a stream. Once we added Streaming(false) to get a snapshot instead then the memory consumption was greatly reduced. Thanks for the feedback.

EXPLORE OUR SITES