Running skaha in async main function error: RuntimeError: This event loop is already running

[axshen]

I am writing some async code to wrap the skaha sessions into a job function so that I can concurrently generate containers. Here is a simplified example of the code I am trying to run

# imports ...

async def job(name, session, params):
    session_id = session.create(**params)
    res = session.info(session_id)
    return res

async def main():
    session = Session()
    await job('test', session, {name: 'test', 'image': '...', ...})

if __name__ == '__main__':
    asyncio.run(main())

When I try and run my code, I get a RuntimeError: This event loop is already running. I read this stack overflow post which suggests you can't use loop.run_until_complete inside of another event loop. They also suggest a hack for getting it working with nest_asyncio, which I guess is fine and something I can try.

Do you have other ideas or potential fixes for this so that it can work inside of async code?

Here is the relevant snippet from the error logs:

  File "/Users/she393/Documents/canfar/./pipeline.py", line 84, in pipeline
    await job(params['name'], session, params)
  File "/Users/she393/Documents/canfar/./pipeline.py", line 55, in job
    session_id = session.create(**params)
  File "/Users/she393/Documents/canfar/venv/lib/python3.9/site-packages/skaha/session.py", line 261, in create
    results = loop.run_until_complete(scale(self.session.post, arguments))
  File "/Applications/Xcode.app/Contents/Developer/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 618, in run_until_complete
    self._check_running()
  File "/Applications/Xcode.app/Contents/Developer/Library/Frameworks/Python3.framework/Versions/3.9/lib/python3.9/asyncio/base_events.py", line 578, in _check_running
    raise RuntimeError('This event loop is already running')

Software/OS:

• macOS 15.1
• Python 3.9.6
• skaha==1.4.2

[shinybrar]

Skaha manages concurrency underneath the hood for you. If you call skaha.session.Session.create with argument replicas=64, it will launch 64 containers in parallel for you.

But you are correct, if you run skaha from within an asyncio loop, it fetches the existing loop to thread under, which causes the issue you see. I had not really considered the case when the user was already in a asyncio loop, e.g. JupyterLab. I will release a fix for this issue soon.

[shinybrar]

@axshen After thinking about this a bit more, what you actually want is a new class called AsyncSession rather than some a function that can support both sync and async environments.

from skaha.sessions import AsyncSession
session = AsyncSession()
jobs = await session.create(...)
print(await session.logs(jobs))

Then you are free use an async env, as needed. Would that be your ideal solution?