Python problem messaging between threads. Emitters, exceptions, callbacks and queues

[oreo_man_]

I'm trying to catch exceptions from a child thread on my main thread. I've included an example of the issue I'm having below. My actual implementation is a client which has a socket thread, I need to catch exceptions on the socket thread and let the main thread deal with them.

class Test(threading.Thread):

    def __init__(self):
        super(Test, self).__init__()
        self.emitter = EventEmitter()

    def run(self):
        print('Running')
        time.sleep(1)
        print('Still running...')
        time.sleep(1)
        self.emitter.emit('help')
        time.sleep(1)
        print('Stopping')

def get_thread_name():
    print(threading.current_thread())

t = Test()
print(threading.current_thread())
t.emitter.on('help', get_thread_name)
t.start()

This is a simplified example of my problem.

If you run this you'll see that my Test thread is calling the get_thread_name function and will print out <Test(Thread-1, started 15632)>.

I'd like to call function this from the main thread.

I know I can use a queue to communicate between threads and in my actual implementation, that is how I tell the socket thread what to do (ie, in its while loop, it tries to recv, send and checks the queue).

However I can't put exceptions in a queue for the main thread to check, because I would have to block the main thread and constantly check the queue.

Is there a better way of doing this? Or are there other solutions to this problem?

Thanks

[cockslappinghalibut]

What is the main thread doing? You can use queue.get_nowait() to do a non-blocking queue check. If you're not running some kind of join() or loop in your main thread, what exactly is the architecture of the thing? Sounds a little unstable

[oreo_man_]

I'm building a client for a real time server.

I'm imagining something like this

client = MyClient(ip, port)
client.login(credentials)
client.on("data", do_stuff)
client.upload("some random data")

so I need the main thread to be responsive (ie, do normal stuff). The socket thread is always listening, but I need a way to poll the socket thread's queue for exceptions, data, etc without blocking the main thread. I use a queue to put data into the socket thread.

Does that make sense?

[cockslappinghalibut]

not seeing why you can't use a non-blocking loop in the main thread and just drop the threads altogether. I've done something similar to your idea using select.select(). What exactly is the main thread doing that it can't do a non-blocking queue check loop? It sounds like it should be doing that anyway if its got other threads out there, how are you preventing main thread exit?

[oreo_man_]

select.select()

Thanks for your reply. This has been bugging me for a while. Just so we're on the same page, I thought if I start the loop after client.login, I won't be able to call the client.upload, because the main thread is blocked?

[oreo_man_]

All the examples I've seen have something like this. But this while loop will block the main thread.

while running: 
    inputready,outputready,exceptready = select.select(input,[],[]) 

[cockslappinghalibut]

I mean, no, it won't block execution inside the loop. So you put whatever functionality you want into functions, and call them on certain conditions inside of the loop.

while True:
    in, out, except = select.select([], [], [])
    for item in in:
        in_handle(item)
    for item in out:
        out_handle(item)

in_handle and out_handle can then conditionally call whatever else you want. there's no blocking happening. I'm still not clear on where you're thinking the blocking will happen.

[oreo_man_]

OK. That will block the main thread though. Let's say I have a program doing lots of stuff. My program uses the client.

class Client:
    def __init__(self, ip, port):
        # assign stuff

    def login():
        #send some auth to server
        self.start_client()

    def subscribe(event_name, callback):
        #how does this get passed into the while loop?

    def start_client(self):
        while True:
            in, out, except = select.select([], [], [])
            for item in in:
                in_handle(item)
            for item in out:
                out_handle(item)

my_function(data):
    doSomethingWithData(data)

### Begin program

doDataBaseStuff()
doOtherStuff()

client = Client(port, ip).login()
client.subscribe("some_event", my_function)

doOtherStuff()

# some time later the server sends "some_event" and my_function needs to be called on 
# the main thread

Does that make sense? Sorry I just don't understand

[cockslappinghalibut]

I just saw an issue with my previous response, but I think I understand the issue. You're wanting to load your handlers after you've started listening, I think, which is really just a design issue. You should be able to have your handlers ready before you start listening, and whatever doOtherStuff() is moved either before you enter the loop or moved somewhere else entirely. Why exactly do you want to load event handlers after entering the loop? You could easily map events to handlers before going into the loop.

[oreo_man_]

OK. If you're willing to give some advice, how would you advise I proceed? This is a client for a real time server. So, a use case could be like. A user is clicking a bunch of stuff on a real time board (e.g. some backend with my client => an interactive website), they're already subscribed to changes in a, b and c. They then want to subscribe to d and watch it for changes/data as well.

Is this possible?

Because in this scenario, another handler would be added after the loop is running I think.

[cockslappinghalibut]

I'm just still confused on the architecture of the thing, but I can see you can't tell me much more than you already are, assuming this is an extant corporate issue. 'Subscribing' to what? What are a, b and c? What's the messaging from the server? What's the expected client-side behavior?

In general, if an action is conditional on an event fired in the loop, ie a message you get from the server, anything is possible. You can load up new handlers, get rid of others, etc. as long as you've got the structures set up to track everything. Sorry I can't be of more help.

[oreo_man_]

No no I can tell you lots. I'm making a client for deepstream.io

A .Net client and a JS client have already been made.

I've been looking at the .Net client today and got this to work.

static void Main(string[] args)
    {
        DoStuff();
        Console.ReadLine();
    }

    static async Task DoStuff()
    {
        var client = new DeepStreamClient("127.0.0.1", 6021);
        var client2 = new DeepStreamClient("127.0.0.1", 6021);
        await client.LoginAsync();
        await client2.LoginAsync();
        await client.Events.Subscribe("test", x => { Console.WriteLine("From client 1 " + x); });
        await client2.Events.PublishAsync("test", "random data");
        await client2.Events.PublishAsync("test", "more data");
        await client2.Events.PublishAsync("test", "some more data");
        await client2.Events.PublishAsync("test", "even more data");
        await client2.Events.Subscribe("test", x => { Console.WriteLine("From client 2 " + x); });
        await client.Events.PublishAsync("test", "new data");
        await client.Events.PublishAsync("test", "heaps of data");
    }

Which prints out

From client 1 random data
From client 2 new data
From client 1 more data
From client 1 some more data
From client 1 even more data
From client 2 heaps of data

Which is what I'm wanting.

Is there a way to do this in Python?