Can't make asynchronous read requests

[Aditya23456]

Hi,

I want to write an asynchronous readmultipleproperty application. Im new to multithreading programming so maybe some of my understanding is wrong. I have 4 devices in my network which I want to read asynchronously. I created my own custom IOCB class for each of the device and added them into the request_queue. I see each of my request in request_queue gets read by the process_task callback and is sent to app.request. This seems fine but i receive confirmation or response only for the first IOCB and remaining 3 iocb's doesn't get any response. But when I do time.sleep(1) after adding the IOCB in request_queue at the main thread, I get responses for all the IOCBs. I don't want my main thread to be blocked with time.sleep(1) because I want asynchronous communication(if I have time.sleep(1) after each addition to request_queue i can read only 60 devices per minute), so I tried with time.sleep(2) at the process_task callback(since this is running in other thread) after each iocb is popped out of request queue but that doesn't work. I still get response for only 1st IOCB. I really don't understand what is the problem. This is my code---

AT main thread--------- for device in devicelist: #prepared a point list and collect address of device iocblist.append(readproperties(address,point_list)) for iocb in iocblist: result=iocb.ioResult.get(True, 0.5) #gets response only for first iocb

def readproperties(target_Address,read_access_spec_list): #simplifing my pointlist into this variable request = ReadPropertyMultipleRequest(listOfReadAccessSpecs=read_access_spec_list) request.pduDestination = Address(target_address)

                iocb = IOCB(request, self.async_call)
                self.this_application.submit_request(iocb)
                #time.sleep(1) this line fetches data. if I comment this I get result for only 1st IOCB
                return iocb

#submit request puts my iocb in self.request_queue from which process_task reads iocb

Please let me know if I am missing something. Do you suggest me to use IOCB module in bacpypes library? I am using volttron IOCB module.

Thanks for your time. Regards, Aditya

[JoelBender]

You are probably getting tripped up by the spin parameter of the bacpypes.core.run() function being set too high so your threads are being starved, thanks to the GIL. Try calling enable_sleeping() in the same thread that you call the run() function, or pass a very small value to run(). This will change the nature of the while running loop from a "wait a long time until something happens" to "wait very briefly and give up, allowing other threads to get work done." You can also trigger a context switch by calling time.sleep(0.000001).

I'm not familiar enough with volttron to understand how their AsyncCall functionality works with their IOCB module in the BACnetProxyAgent.

[JoelBender]

The volttron requirements reference BACpypes 0.13.8, a lot has changed since then. If you need a different sample let me know.

[Aditya23456]

Respected Sir,

I still face thread starving issue. I tried with spin=0.01 and 0.005. I even started with enable_sleeping before calling core.run(). Still I get results for all devices only when I have time.sleep(1) in my main thread and response is only for 1st device if I don't have time.sleep(1). Also, if I have time.sleep(0.5) I get result for 2 out of 4 devices, so looks like I need to have minimum amount of sleep time at main thread. I want to explain the way Im working out so that you may suggest if I can change the architecture somehow to make async requests possible.

Although I am using volttron based IOCB, I am using bacpypes 0.16.2. I have a multiprocessor application with modbus, restful API and bacnet as 3 processes(other than server process) My bacnet process looks as follows: During the initialisation of the process I have, def setup_device(self, this_device,address): def start_server(): bacpypes.core.enable_sleeping(0.1) bacpypes.core.run(0.005) self.this_application = BACnet_application(this_device, address) #interfaced with bacpypes server_thread = threading.Thread(target=start_server) server_thread.daemon = True server_thread.start()

Where self class is the process running along with modbus, restfulapi processes. So I think core.run is not initialised on the main thread. Then I have other methods in my self class which can be called by rpc or message passing. Here I have my read property class, which builds all iocb's based on devices connected to my network. Each iocb is iniatialised with an asyncresult class of gevent which has a queue where collected data is put. I don't think there's a problem with iocb variables because, I get server confirmation called for only my first iocb message and confirmation is not called for all remaining iocbs. So I feel IOCB is not a problem here. Do you feel I can do anything so that I can make requests to devices and get back response without blocking the main thread?

Also, to make async requests in the other restfulapi and modbus process, I used python inbuilt async library which had run in executer function. This created a threadpool executor for each device and made async/concurrent requests. I tried similar way for bacpypes but it didn't work. So is there any way I can make async requests by utilizing asyncio library?

Also, As you suggested different sample, it would be great if you suggested some asynchronous read example file which I can modify as per my scenario.Thank you very much for your time.

Regards, Aditya

[JoelBender]

So I think core.run is not initialised on the main thread.

The asyncore.loop() function that sits in the middle of core.run() has to be given some time to process socket activity, so unless you can somehow tell your main thread to give up more time I don't think this is going to improve.

Each iocb is iniatialised with an asyncresult class of gevent which has a queue where collected data is put.

Ah, I see now that volttron is very intricately wired into gevent and Greenlets, and BACpypes is not. You could very easily be lost in the shuffle back and forth between the thread models.

Also, to make async requests in the other restfulapi and modbus process, I used python inbuilt async library which had run in executer function.

Isn't that a Python 3.4 feature, and the rest of volttron is Python 2.7? I have not done anything with concurrent futures, or async/await. I suspect that fully embracing the asyncio library will require replacing the core and task modules, which isn't necessarily a bad thing by itself, but it will cause some headaches. Similarly, supporting gevent might be quite a bit if work.

As long as the RPC protocol that you are using (which is the VOLTTRON Interconnect Protocol if I'm following your project correctly) remains consistent, then at least this possible. I see that as of version 15.0 of pyzmq that it supports asyncio, so it probably also falls on volttron to support it as well.

[JoelBender]

I wrote another sample application which might help, ThreadedReadProperty.py. This was added to stage so it will be included in the next release. I haven't tested it on a big network with lots of devices and points yet.

[JoelBender]

Has this been resolved? Do you have any more thoughts or comments?

[Aditya23456]

Respected Sir,

Sorry for late reply. I did test it like a few months ago when I found that with this implementation it would take about 5-6 seconds for 5 devices to respond which is similar to do time.sleep(1) in the main thread. I guess thread starving is inevitable. Thank you very much for your time and assistance.

Regards, Aditya