controller.py

import uuid
import execnet
import remote
import time
import asyncio


loop = None


def received(id):
    asyncio.ensure_future(_received(id), loop=loop)


async def _received(id):
    print(f"\t<{id}")


async def send(*msg):
    asyncio.get_running_loop().run_in_executor(None, ch.send, *msg)


async def main():
    global loop
    loop = asyncio.get_running_loop()
    id = 0
    while True:
        id += 1
        print(f">{id}")
        await send((id, 4))
        await asyncio.sleep(1)

print ("Connecting")
gw = execnet.makegateway()
ch = gw.remote_exec(remote)
ch.setcallback(received)
print ("Starting up")
asyncio.run(main())

remote.py

import time
import asyncio
import sys
import concurrent.futures

loop = None


work_pool = concurrent.futures.ThreadPoolExecutor(max_workers=20)
submit_pool = concurrent.futures.ThreadPoolExecutor(max_workers=20)


def do(duration):
    time.sleep(duration)


async def task(id, duration):
    print(f"Working on {id}", flush=True)
    await loop.run_in_executor(work_pool, do, duration)

    print(f"Done with {id}", flush=True)
    await loop.run_in_executor(submit_pool, send, id)

    print(f"Sent response for {id}", flush=True)


def send(result):
    print(f"Sending response for {result}", flush=True)
    channel.send(result)


def receive(channel):
    id, duration = channel.receive()
    print(f"Received {id}", flush=True)
    return id, duration


async def main(channel):
    global loop
    loop = asyncio.get_running_loop()
    while not channel.isclosed():
        print("Waiting for message")
        msg = await loop.run_in_executor(None, receive, channel)
        print("Got msg {msg[0]}, sending future")
        asyncio.ensure_future(task(*msg))


if __name__ == '__channelexec__':
    sys.stdout = open('remote.log', 'w')
    asyncio.run(main(channel))