Cannot pickle local function when sending callable Filter objects via multiprocessing Queue

A few points:

1. Function function should be at global scope.
2. There is no reason why method ask cannot be an instance method given the rest of the code.
3. Trying to pickle a queue as is done in your current ask implementation by putting responsequeue to a queue will not work. But given that the FilterBroker.run method is processing requests serially in a loop anyway, there is no reason why the response queue should not be pre-allocated by the init method. Then we just need to use a lock to ensure that ask calls are serialized so that the response always matches up with the request.
4. We can use the multiprocessing.Queue class instead of a managed queue thereby avoiding having to start an additional process and making for more efficient get and put method calls.

from threading import Thread
from multiprocessing import Lock, Queue
from dataclasses import dataclass
from typing import Optional
import logging
import inspect
@dataclass
class Service:
    id: Optional[int] = None
    name: str = ""
    port: int = 0
class Filter:
    def init(self, filterfunction: callable):
        self.filterfunction: callable = filterfunction
        self.subscribers: list[Service] = []
    def call(self, args, kwds):
        return self.filter_function(args, kwds)
class FilterBroker(Thread):
    def init(self) -> None:
        super().init()
        self.requestqueue = Queue()
        self.responsequeue = Queue()
        self.lock = Lock()
        self.filters: dict[str, Filter] = {}
    def addfilter(self, name: str, filter: Filter):
        if len(inspect.signature(filter).parameters) != 2:
            raise TypeError("Invalid Filter: must have exactly two parameters")
        self.filters[name] = filter
    def run(self):
        classname = self.class.name
        logging.info(f"[{classname}]: Process started")
        while True:
            try:
                task = self.requestqueue.get()
                logging.debug(f"[{classname}]: Task received: {task}")
                if task is None:
                    response = None  # We must set this due to the finally block
                    break
                if not isinstance(task, tuple) or not callable(task[0]):
                    raise ValueError('bad arguments')
                method, *args = task
                response = method(self, *args)
            except Exception as e:
                print(e)
                response = None
            finally:
                self.responsequeue.putnowait(response)
    def shutdownthread(self):
        with self.lock:
            self.requestqueue.put(None)
            self.responsequeue.get()
        self.join()
    def ask(self, task):
        with self._lock:
            self._request_queue.put(task)
            print("I put in queue")
            result = self._response_queue.get()
            print("I got result", result)
            return result
def function(x):
    return x > 0
def task(broker):
    f = Filter(function)
    print(f(2))
    broker.ask(FilterBroker.add_filter, 'test', f)
    logging.debug(f"Filter added")
if name == 'main':
    broker = FilterBroker()
    broker.start()
    thread = Thread(target=task, args=(broker,))
    thread.start()
    thread.join()
    print("Thread finished")
    # Kill broker:
    broker.shutdown_thread()

from dataclasses import dataclass
from typing import Optional
import logging
import inspect
from multiprocessing import Process
from multiprocessing.managers import BaseManager
import cloudpickle
@dataclass
class Service:
    id: Optional[int] = None
    name: str = ""
    port: int = 0
class Filter:
    def init(self, filter_function: callable):
        self.filter_function: callable = filter_function
        self.subscribers: list[Service] = []
    def call(self, args, kwds):
        return self.filterfunction(*args, kwds)
class FilterBrokerImpl:
    def init(self) -> None:
        self.filters: dict[str, Filter] = {}
    def addfilter(self, serializedargs: bytes):
        """Add a filter."""
        name, filter = cloudpickle.loads(serializedargs)
        if len(inspect.signature(filter).parameters) != 2:
            raise TypeError("Invalid Filter: must have exactly two parameters")
        print(f'addfilter name = {name}, filter = {filter}' )
        self.filters[name] = filter
    def getfilters(self) -> bytes:
        """Return the filters."""
        return cloudpickle.dumps(self.filters)
class FilterBroker:
    """A wrapper (adapter) that hides the intricacies of serializing
    arguments and return values to the various FilterBrokerImpl methods."""
    def init(self, broker: FilterBrokerImpl):
        self.broker = broker
    def addfilter(self, name: str, filter: Filter) -> None:
        self.broker.addfilter(cloudpickle.dumps((name, filter)))
    def getfilters(self) -> dict[str, Filter]:
        return cloudpickle.loads(self.broker.getfilters())
def task(broker, name, n):
    def genfilter():
        def function(x):
            return x > n
        return function
    f = Filter(genfilter())
    print(f'result of executing filter named {name} with argument 2: {f(2)}')
    broker.addfilter(name, f)
    logging.debug(f"Filter added")
class FilterBrokerManager(BaseManager):
    pass
def main():
    FilterBrokerManager.register('FilterBroker', FilterBrokerImpl)
    with FilterBrokerManager() as manager:
        broker = FilterBroker(manager.FilterBroker())
        p1 = Process(target=task, args=(broker, 'test0', 0))
        p2 = Process(target=task, args=(broker, 'test5', 5))
        p1.start()
        p2.start()
        p1.join()
        p2.join()
        print('filters:', broker.getfilters())
if name == 'main':
    main()

Prints:

result of executing filter named test5 with argument 2: False
result of executing filter named test0 with argument 2: True
addfilter name = test5, filter = 
addfilter name = test0, filter = 
filters: {'test5': , 'test0': }