Do csp adapters support asyncio?

[kootsZhin]

I am trying to implement IPC for multiple CSP processes running on the same machine using NATS, which turns out to require using asyncio in python

I tried playing with it in CSP but it seems that it is not working with adapters?

If so, alternatively, is there any recommendation on IPC for CSP processes? (I saw csp.gateway on the roadmap but looking for some immediate solutions for now)

Thank you so much!

[AdamGlustein]

Can you provide a bit more context on the error you're seeing? If I understand correctly you're not trying to use any coroutines directly within an adapter, but there is some weird behavior due to an async call coming from the NATS library. It would be super helpful if you could provide a small code snippet which reproduces the issue.

[kootsZhin]

Thanks and sorry for not being clear enough. What I were trying to do is using coroutines directly within OutputAdapter, here is a snippet I did, the logger.info at the async start was not logging so I assumed something is wrong there

class NatsOutputAdapterImpl(OutputAdapter):
    def __init__(self):
        self.nc = None

    async def start(self):
        self.nc = await nats.connect("nats://localhost:4422")
        logger.info("Connected to NATS")

    async def stop(self):
        await self.nc.flush()
        await self.nc.close()
        logger.info("Disconnected from NATS")

    async def on_tick(self, time: datetime, value: object):
        # not yet implemented
        pass

NatsOutputAdapter = py_output_adapter_def(
    name="NatsOutputAdapter",
    adapterimpl=NatsOutputAdapterImpl,
    input=csp.ts[object],
)

[robambalu]

@kootsZhin as Adam said, some more context would help here. Without more context, I do think you should be able to use asyncio on your adapter thread. This would be for realtime data only, so your adapter would need to manage its own thread for the async data. That processing on that thread can do whatever you want, and can use asyncio as needed

[robambalu]

output adapters are not async aware. You cant really mix asyncio with the main csp engine thread since the thread is driven directly by the engine ( which doesnt manage any asyncio at the moment )
What you can do here is offload the data to a separate thread using a threading queue, and have that thread run asyncio to feed into NATS

[kootsZhin]

Thank you and wondering any better suggested solutions for IPC among multiple CSP processes?

[robambalu]

To be clear, are you looking for IPC for a realtime process or for parallelizing a historical run?
For the former you can use NATS if that is your preference, you can use anything you like really if you build out the adapter for it
For distributed simulated runs it would be non-trivial to do it outside of the engine itself. Its on our roadmap to support this though