import asyncio
import functools
import signal
counter = 0
async def run_bash_sleep():
  global counter
  counter += 1
  local_counter = counter
  try:
    print(f"Started - {local_counter}")
    proc = await asyncio.create_subprocess_exec(
      'bash', '-c', 'sleep .001',
      stdout = asyncio.subprocess.PIPE,
      stderr = asyncio.subprocess.PIPE,
      start_new_session = True
    )
    print(f"Waiting - {local_counter}")
    stdout, stderr = await proc.communicate()
    print(f"Done - {local_counter}!")
  except asyncio.CancelledError:
    print(f"Canceled - {local_counter}!")
async def run_loop(loop):
  max_jobs = 8
  active_tasks = []
  while True:
    try:
      # Add jobs to the list of active jobs
      while len(active_tasks) < max_jobs:
        active_tasks.append(loop.create_task(run_bash_sleep()))
      # All tasks have finished, end the loop
      if len(active_tasks) == 0:
        break
      # Wait for a test to finish (or a 1 second timeout)
      done, pending = await asyncio.wait(
        active_tasks,
        timeout = 1,
        return_when = asyncio.FIRST_COMPLETED
      )
      print(f"Running jobs: {len(active_tasks)}")
      # Update the active jobs
      active_tasks = list(pending)
    except asyncio.CancelledError:
      max_jobs = 0
def stop_asyncio_loop(signame, loop):
  for task in asyncio.all_tasks(loop):
    task.cancel()
def main():
  loop = asyncio.new_event_loop()
  asyncio.set_event_loop(loop)
  for signame in {'SIGINT', 'SIGTERM'}:
    loop.add_signal_handler(
      getattr(signal, signame),
      functools.partial(stop_asyncio_loop, signame, loop)
    )
  loop.run_until_complete(loop.create_task(run_loop(loop)))
main()