Klein app with deferred

Question

I am exploring Klein and Deferred. In the following example I am trying to increment a number using a child process and return it via Future. I am able to receive the Future call back.

The problem is that deferred object never calls the cb() function and the request made to endpoint never returns. Please help me identify the problem.

Following is my server.py code

from klein import Klein
from twisted.internet.defer import inlineCallbacks, returnValue
import Process4

if __name__ == '__main__':
    app = Klein()

    @app.route('/visit')
    @inlineCallbacks
    def get_num_visit(request):        
        try:
            resp = yield Process4.get_visitor_num()
            req.setResponseCode(200)
            returnValue('Visited = {}'.format(resp))
        except Exception as e:
            req.setResponseCode(500)
            returnValue('error {}'.format(e))

    print('starting server')
    app.run('0.0.0.0', 5005)

Following is Process4.py code

from multiprocessing import Process
from concurrent.futures import Future
from time import sleep
from twisted.internet.defer import Deferred

def foo(x):
    result = x+1
    sleep(3)
    return result


class MyProcess(Process):

    def __init__(self, target, args):
        super().__init__()
        self.target = target
        self.args = args
        self.f = Future()
        self.visit = 0

    def run(self):
        r = foo(self.visit)
        self.f.set_result(result=r)

def cb(result):
    print('visitor number {}'.format(result))
    return result

def eb(err):
    print('error occurred {}'.format(err))
    return err


def future_to_deferred(future):
    d = Deferred()

    def callback(f):
        e = f.exception()
        if e:
            d.errback(e)
        else:
            d.callback(f.result())

    future.add_done_callback(callback)
    return d

def get_visitor_num():
    p1 = MyProcess(target=foo, args=None)
    d = future_to_deferred(p1.f)
    p1.start()
    d.addCallback(cb)
    d.addErrback(eb)
    sleep(1)
    return d

Edit 1

Adding callbacks before starting the process p1 solves the problem of calling cb() function. But still the http request made to the endpoint does not return.

Answer 1

It turns out that setting future result self.f.set_result(result=r) in the run() method triggers the callback() method in the child process, where no thread is waiting for the result to be returned!

So to get the callback() function triggered in the MainProcess I had to get the result from the child-process using a multiprocess Queue using a worker thread in the MainProcess and then set the future result.

@notorious.no Thanks for reply. One thing which I noticed is that reactor.callFromThread does switches result from worker thread to MainThread in my modified code however d.callback(f.result()) works just fine but returns result from worker thread.

Following is the modified working code

server.py

from klein import Klein
from twisted.internet.defer import inlineCallbacks, returnValue


import Process4

if __name__ == '__main__':
    app = Klein()
    visit_count = 0

    @app.route('/visit')
    @inlineCallbacks
    def get_num_visit(req):
        global visit_count
        try:
            resp = yield Process4.get_visitor_num(visit_count)
            req.setResponseCode(200)
            visit_count = resp
            returnValue('Visited = {}'.format(resp))
        except Exception as e:
            req.setResponseCode(500)
            returnValue('error {}'.format(e))

    print('starting server')
    app.run('0.0.0.0', 5005)

Process4.py

from multiprocessing import Process, Queue
from concurrent.futures import Future
from time import sleep
from twisted.internet.defer import Deferred
import threading
from twisted.internet import reactor


def foo(x, q):
    result = x+1
    sleep(3)
    print('setting result, {}'.format(result))
    q.put(result)


class MyProcess(Process):

    def __init__(self, target, args):
        super().__init__()
        self.target = target
        self.args = args
        self.visit = 0

    def run(self):
        self.target(*self.args)


def future_to_deferred(future):
    d = Deferred()

    def callback(f):
        e = f.exception()
        print('inside callback {}'.format(threading.current_thread().name))
        if e:
            print('calling errback')
            d.errback(e)
            # reactor.callFromThread(d.errback, e)
        else:
            print('calling callback with result {}'.format(f.result()))
            # d.callback(f.result())
            reactor.callFromThread(d.callback, f.result())
    future.add_done_callback(callback)
    return d


def wait(q,f):
    r = q.get(block=True)
    f.set_result(r)


def get_visitor_num(x):

    def cb(result):
        print('inside cb visitor number {} {}'.format(result, threading.current_thread().name))
        return result

    def eb(err):
        print('inside eb error occurred {}'.format(err))
        return err

    f = Future()
    q = Queue()
    p1 = MyProcess(target=foo, args=(x,q,))

    wait_thread = threading.Thread(target=wait, args=(q,f,))
    wait_thread.start()

    defr = future_to_deferred(f)
    defr.addCallback(cb)
    defr.addErrback(eb)
    p1.start()
    print('returning deferred')
    return defr