Allow threaded operation

[dmopalmer]

Allow threaded operation, including a handler that places the messages on the queue and an event to stop the thread.

Includes an example in the README.md file, and a command line pygcn-threaded-listen that continuously counts seconds and has a --maxtime timeout parameter.

[dmopalmer]

Just a ping. Any comments on this PR?

[lpsinger]

I like the idea of a sentinel to terminate the loop. I'm not sure I'd do it this way, though. I have been thinking for a while of doing a complete rewrite using asyncio, which would give the caller much more control over concurrency.

[dmopalmer]

I actually read the documentations to see if you could use the queue itself as a sentinel. (e.g. for pipes, if a reader closes a pipe the writer can tell.)

No luck, but I found a bug in my example code: For a vanilla queue.Queue() you are supposed to to call q.task_done() after processing the result of each pop.

asyncio also sounds good (at the expense of anyone still using Python 2.7 or <3.4 unless you do a lot of work to make it compatible). Would it require that the rest of the code it is tying in to must also use asyncio for things like sockets?

[lpsinger]

asyncio also sounds good (at the expense of anyone still using Python 2.7 or <3.4 unless you do a lot of work to make it compatible). Would it require that the rest of the code it is tying in to must also use asyncio for things like sockets?

I think that the existing function would remain and would suffice for most users, but under the hood it would become a thin wrapper for starting a run loop and calling a coroutine.

[dmopalmer]

Is this a 'doing it now' extension, or is it a 'future plans' thing? (I am getting ready for the LIGO/VIRGO run which was my motivation for doing the multithread.)

So perfectly valid paths are:

1. I use my dmopalmer fork without it being merged into the lpsinger origin, then use asyncio when the origin is ready.
2. You say 'threading is good enough' and merge my fork.
3. I expect the asyncio features to be ready in time and write code towards that.
4. I learn enough asyncio to implement it myself.

I don't think I have time for 4. If asyncio is on a few-days horizon then I can take path 3. Otherwise it is your choice for 1. or 2. .

I won't be heartbroken if a better implementation means that my fork gets stubbed off; it was largely a learning experience.

[lpsinger]

Let me take a look at the asyncio stuff. I should know in a day or two if it's on the immediate horizon or if it's a longer term project.

[dmopalmer]

Just wanted to bump this at the 1 year point. I fixed a pep8 that prevented Travis from passing and also found that I had duplicated a function.

[lpsinger]

@dmopalmer, after O3 is over, I will rewrite this module using asyncio, at which point you will be able to interrupt the listener using the cancel() method.

As for this pull request, I'm sorry to say, but I'm hesitant to add a special code path for this particular style of sentinel-based cancellation. The reason is that I've designed PyGCN to be as simple as possible, and that means making no assumptions about the concurrency or threading setup of the surrounding program.

The listen() method is meant to run in an infinite loop until the program is stopped by SIGINT (Ctrl-C). If the listen() method is not going to be running on your main thread, then you can run it in a subprocess so that you can terminate it by sending it a signal, like this:

import gcn
import multiprocessing


def handler(payload, root):
    # put your GCN handling code here


if __name__ == '__main__':
    gcn_process = multiprocessing.Process(
        target=gcn.listen, kwargs={'handler': handler})
    gcn_process.start()

    try:
        # put your main program's work here
    finally:
        gcn_process.terminate()
        gcn_process.join()

Would this solution work for you? If so, I would be happy to add it to the README.

[dmopalmer]

I will continue to use my fork with threading until your asyncio version is available, but understand if you don't want to incorporate my pull request.

The README might include a suggestion to use have the handler() use a global queue.Queue to pass the (payload,root) values to the main program if that is desired.

Does calling gcn_process.terminate() cause any problems (dangling ports, etc.) that would have to be cleaned up in your code by a SIGTERM handler?

[lpsinger]

I will continue to use my fork with threading until your asyncio version is available, but understand if you don't want to incorporate my pull request.

Sorry for leaving this up in the air for so long.

The README might include a suggestion to use have the handler() use a global queue.Queue to pass the (payload,root) values to the main program if that is desired.

I'd be very happy if you would like to contribute the README section.

Does calling gcn_process.terminate() cause any problems (dangling ports, etc.) that would have to be cleaned up in your code by a SIGTERM handler?

I don't know. I'd have to check. Everything is cleaned up neatly if you terminate it with SIGINT, I don't know about SIGTERM.

This was why I didn't suggest starting a thread with daemon=True, since such threads do not get cleaned up.

[dmopalmer]

Is now a good time to re-open this pull request for threaded VOEvent operation?

We are gearing up our telescopes for more GRB (and GW when available)  response, and would prefer to use mainline rather than fork libraries where possible.

You didn’t want to change things during O3.  I don’t know if it has been decided whether the rest of O3 will resume or if LIGO will not restart until O4.  Either way, we are unlikely to get new GW candidates in the near future.

You were also thinking of using asyncio instead of threads.  Any progress on that?

[dmopalmer]

Is there any more decision on whether to add either threading (through this pull request) or asyncio to this package?

[lpsinger]

I would welcome a PR to re-implement the client using asyncio, which would provide greater flexibility for the user's code to multiplex multiple connections in a single thread.

I didn't accept the implementation in this PR partly because I felt like it made some application-specific assumptions that belonged in user code.