How to check the remaining buffer of a channel

I am using the `OmmProvider` Class to implement a custom data provider. Essentially I am doing this as demonstrated by [this sample](https://github.com/Refinitiv/Real-Time-SDK/blob/master/CSharp/Ema/Examples/Training/IProvider/100_Series/100_MP_Streaming/IProvider.cs):

 public static void Main(string[] args)
    {
        OmmProvider? provider = null;
        AppClient appClient = new AppClient();
        FieldList fieldList = new FieldList();
        OmmIProviderConfig config = new OmmIProviderConfig();
        provider = new OmmProvider(config.Port("14002"), appClient);
        while (true)
        {
            fieldList.Clear();
            fieldList.AddReal(22, 3991 + i, OmmReal.MagnitudeTypes.EXPONENT_NEG_2);
            fieldList.AddReal(30, 10 + i, OmmReal.MagnitudeTypes.EXPONENT_0);
            provider.Submit(new UpdateMsg().Payload(fieldList.Complete()), appClient.ItemHandle);
            Thread.Sleep(1000);
        }
    }

The problem is, without `Thread.Sleep()`, the provider will produce update faster than my consumer can consume, casuing the following exception:

Exception Type='OmmInvalidUsageException',
Text='Failed to submit UpdateMsg. Reason: NO_BUFFERS. Error text: channel out of buffers chnl=Rssl Channel
scktChannel: <Undefined>
connected: True
state: ACTIVE
connectionType: SOCKET
clientIP: 127.0.1.1
clientHostname: <omitted>
pingTimeout: 60
majorVersion: 14
minorVersion: 1
protocolType: RWF
userSpecObject: Client handle = 1
, errorId=FAILURE, errorText=Channel out of buffers', Error Code='-3'

Using `Thread.Sleep()` does work, but it does not seem to be a performant solution. Therefore, I am wondering how I can query the remaining space of the buffer, so that I can pause the producer if it is almost full.

Thanks,

Find more posts tagged with

.net

ema-api

refinitiv-realtime-sdk

#technology

Accepted answers

wasin.w

Hello @vnaik01

About the issue.

I got the information below from the RTSDK team.

The EMA API does not have an interface for checking the remaining space of the buffer. The RTSDK team suggestions are as follows:

The Provider side:

An application can catch this issue and increase number of guaranteed output buffers by using the OmmProvider.ModifyIOCtl(IOCtlCode code, int val, long handle) method and then submitting the message again

Alternatively, you can increase the guaranteed output buffers via the GuaranteedOutputBuffers configuration of the EmaConfig.xml file.

The Consumer side:

Consumer slowness should be addressed either by taking the time to identify bottleneck in code or by running more instances of it with a split watchlist.

A consumer application should minimize the time spent processing each data message in the callback methods (onRefreshMsg(), onUpdateMsg(), onStatusMsg() , …). If an application spends too much time on those callback methods, it impacts overall data dispatcher process which may cause an application got next data delay from the API.

About the support model:

There are two support models from the Real-Time APIs:

The free forum-based support on this Q&A forum website. The questions will be answered by the forum admin (LSEG employees) in the "best effort" manner or other developers who have experience on the issue. There is no guarantee SLA on this forum.
The paid service Real-Time APIs Support Team (aka RDC). The issue will be traced with a ticket number and being investigated by the Real-Time APIs Support Team with SLA. The communication channel is email-based.

Alternatively, you can submit questions to the RTSDK team directly via the GitHub issues page.

I hope this information helps.

All comments

wasin.w

Hello @vnaik01

By default, the API uses "OperationModel.API_DISPATCH" mode that the API run a second thread internally for an application to dispatch incoming message/data to IOmmProviderClient class. That is way an application needs the Thread.sleep() method statement to lets a second thread works.

There is a "OperationModelMode.USER_DISPATCH" mode that the API will not run a second thread. An application is responsible for calling the OmmProvider.Dispatch() method to dispatch all received messages by itself.

Example Code:

provider = new OmmProvider(config.OperationModel(OmmIProviderConfig.OperationModelMode.USER_DISPATCH).Port("14002"), appClient);


while (appClient.ItemHandle == 0)
{
    provider.Dispatch(1000);
    Thread.Sleep(1000); //Just for delay data
}


int count = 0;
var endTime = System.DateTime.Now + TimeSpan.FromMilliseconds(60_000);


while(DateTime.Now < endTime)
{
    provider.Dispatch(1000);


    fieldList.Clear();
    fieldList.AddReal(22, 3991 + count, OmmReal.MagnitudeTypes.EXPONENT_NEG_2);
    fieldList.AddReal(30, 10 + count, OmmReal.MagnitudeTypes.EXPONENT_0);


    provider.Submit(new UpdateMsg().Payload(fieldList.Complete()), appClient.ItemHandle);
    count++;


    Thread.Sleep(1000); //Just for delay data
}

You can find the full example code of the USER_DISPATCH mode on 130_MP_UserDispatch example.

About the "how I can query the remaining space of the buffer" question, I cannot find it on the API document. Let me re-check and get back to you.

vnaik01

Hi @wasin.w ,

thanks for the answer. Please let me know once you have some new discoveries.

Regarding the API document, currently I mainly use [ENTERPRISE MESSAGE API DEVELOPERS GUIDE](https://developers.lseg.com/content/dam/devportal/refinitivrealtimeapi_pdfs/ema_csharp_dev_guide.pdf). Do you have any other pertinent documents that I should be aware of?

Thanks

wasin.w

Hello @vnaik01

There is the API reference manual in the SDK package (download here), but I cannot find any information about this feature too. I am asking the RTSDK team, and I will get back to you once I have any updates from the team.

umer.nalla

Hi @vnaik01

Is this a theoritical excercise you are performing or are you planning to have your consumer connect directly to the provide in real-life?

Also, what is your consumer doing with the data it receives from your Provider? Based on your code snippet above, you are publishing one update per second and your consumer is unable to keep up with this rate?

In a real-life scenario, your provider will normally publish data at a rate determined by external actions/criteria - e.g. an upate for every market activity event or at a pre-determined rate perhaps such once every 10ms or 1s etc.

And the majority of (iif not all) publishers publish to an intemediary enterprise level real-time distribution component such as our Refinitiv Real-time Advance Distribution HUB (ADH) or Real-Time Connector(RTC). These components are designed to accept tens/hundreds of thousands of updates a seconds from connected publishers.

The Consumer would normally be connected to an Advanced Distribution Server (ADS) or an RTC which would provide some additional level of buffering. However, the consumer would still be expected to keep up with the published rate - otherwise, over time even the largest buffer would eventually overflow and/or the real-time data could become somewhat outdated as the time lag between when the publisher submitted the update and the consumer processed the update would continue to increase...

vnaik01

Hi @umer.nalla ,

This is a part of our infrastructure. Our data handler will be able to connect to two sources, one is from Refinitiv, like Refinitiv Real-Time Optimized/Refinitiv Real-Time Direct, this source for real-time market data consumption; the other source from our own provider, our own provider will read from a data file and stream to data handler, it could be used for backtesting or unit testing purposes. The snippet is just an example, as I said in the original post, without

Thread.Sleep(1000);

the provider program crashes after a short while, complaining:

Exception Type='OmmInvalidUsageException',
Text='Failed to submit UpdateMsg. Reason: NO_BUFFERS. Error text: channel out of buffers chnl=Rssl Channel

Currently we are taking a trial and error approach to decide a smaller Thread.Sleep() that is just enough to not crashing the program. But we have to leave some margin of safety so perhaps we are still only at 20% to 50% of the upper limit of throughput.

In this post, we are asking exactly how we can get rid of the `Thread.Sleep()` call so that we can utilize resources (i.e., I/O, bandwidth, CPU, or whatever bottleneck) to its maximum.

vnaik01

Hi @wasin.w ,

I just took a closer look at your user dispatch 130_MP_UserDispatch example and I tried to incorporate this into my program. It turns out that UserDispatch alone it will not solve my problem. The problem is that I will need to find a way to call

provider.Dispatch(1000);

from time to time so that my buffer will not fill up, but in the meantime, I need to try not calling `provider.Dispatch();` too frequently to avoid sending small data packages. How can I decide the interval dynamically? Still I will need to query the current buffer status...

wasin.w

Hello @vnaik01

I have contacted the RTSDK team, but I did not receive any updates yet. I will ping the team.

In the meantime, please be informed that the Q&A forum is the "best effort" forum-based support model with no SLA. If you need the dedicated Real-Time APIs Support team (RDC) with the SLA, please ask your LSEG representative/Account Manager to contact the rdc.administrator@lseg.com to join the Premium Support Program. Then you can submit a support ticket to the Real-Time APIs Support team via a "Contact premium support" button on the https://developers.lseg.com/en/api-catalog/refinitiv-real-time-opnsrc/refinitiv-real-time-csharp-sdk page as follows:

vnaik01

Hi @wasin.w ,

this is also one confusion I have--what is the role of this forum? Are you an employee of LSEG or a hobbyist that somehow is interested in/good at the LSEG/Refinitiv stuff?

wasin.w

Hello @vnaik01

About the issue.

I got the information below from the RTSDK team.

The EMA API does not have an interface for checking the remaining space of the buffer. The RTSDK team suggestions are as follows:

The Provider side:

Alternatively, you can increase the guaranteed output buffers via the GuaranteedOutputBuffers configuration of the EmaConfig.xml file.

The Consumer side:

Consumer slowness should be addressed either by taking the time to identify bottleneck in code or by running more instances of it with a split watchlist.

About the support model:

There are two support models from the Real-Time APIs:

The free forum-based support on this Q&A forum website. The questions will be answered by the forum admin (LSEG employees) in the "best effort" manner or other developers who have experience on the issue. There is no guarantee SLA on this forum.
The paid service Real-Time APIs Support Team (aka RDC). The issue will be traced with a ticket number and being investigated by the Real-Time APIs Support Team with SLA. The communication channel is email-based.

Alternatively, you can submit questions to the RTSDK team directly via the GitHub issues page.

I hope this information helps.

EXPLORE OUR SITES