Does pyzmq support InterpreterPoolExecutor?

[Jzhenli]

Hi, py3.14 add InterpreterPoolExecutor, just want to know if pyzmq can run with the InterpreterPoolExecutor?

import zmq
import logging
from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, InterpreterPoolExecutor
import asyncio
import zmq.asyncio

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

def run_background_service():
    asyncio.run(background_service())

async def background_service():
    logging.info("............")
    try:
        context = zmq.asyncio.Context()
        socket = context.socket(zmq.REP)
        socket.bind("tcp://*:5555")
        
        logging.info("Background service started, waiting for messages...")
        
        while True:
            message = await socket.recv_string()
            logging.info(f"Background service received message: {message}")
            
  
            processed = message[::-1]
            await asyncio.sleep(0.5) 
            
            await socket.send_string(processed)
            logging.info(f"Background service sent response: {processed}")
            
    except asyncio.CancelledError:
        logging.info("Background service cancelled")
    except Exception as e:
        logging.error(f"Background service error: {e}", exc_info=True)
        raise

async def foreground_client():
    context = zmq.asyncio.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://localhost:5555")
    
    messages = ["Hello", "World", "PyZMQ", "Demo"]
    
    for msg in messages:
        try:
            logging.info(f"Foreground client sending message: {msg}")
            await socket.send_string(msg)
            
            response = await socket.recv_string()
            logging.info(f"Foreground client received response: {response}")
            
            await asyncio.sleep(0.2)
            
        except Exception as e:
            logging.error(f"Client error: {e}", exc_info=True)

async def main():
    logging.info("Starting demo - background service in separate process")
    
    with ThreadPoolExecutor() as executor:
        loop = asyncio.get_running_loop()
        service_future = loop.run_in_executor(executor, run_background_service)
        
        await asyncio.sleep(3) 
        
        try:
            await foreground_client()
        finally:
            service_future.cancel()
            try:
                await service_future
            except asyncio.CancelledError:
                logging.info("Background service successfully cancelled")
            except Exception as e:
                logging.error(f"Error cancelling service: {e}")
    
    logging.info("Demo completed")

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

it can not works, but if change it to
with ProcessPoolExecutor() as executor:
or
with ThreadPoolExecutor() as executor:
it will works well, what's the problem?

[minrk]

pyzmq doesn't support subinterpreters yet, but that's not what's happening here.

I don't know anything about InterpreterPoolExecutor or why it wouldn't work. But make sure you check service_future for errors before waiting for foreground_client.

Try this instead of await foreground_client():

foreground_task = asyncio.create_task(foreground_client())
await asyncio.gather(service_future, foreground_task)

That way, if service_future exits with an error, you'll see it immediately, rather than waiting for the foreground task to finish (which it won't). Then I think you'll see:

concurrent.interpreters.NotShareableError: args not shareable

I would guess it's because these functions are defined in __main__ and not serializable to the other interpreter. It might work with imported module functions. You'll see the same error entirely without pyzmq with e.g.:

def run_background_service():
    asyncio.run(whatever())
    
async def whatever():
    pass

I'm not really sure what is required for InterpreterPoolExecutor to be able to run a function.