Multiprocessing: Expose underlying pipe in queues

[TarkaSteve]

BPO	3831
Nosy	@anacrolix

^{Note: these values reflect the state of the issue at the time it was migrated and might not reflect the current state.}

Show more details

GitHub fields:

assignee = None
closed_at = None
created_at = <Date 2008-09-11.01:13:58.188>
labels = ['extension-modules', 'type-feature']
title = 'Multiprocessing: Expose underlying pipe in queues'
updated_at = <Date 2014-02-27.22:38:01.352>
user = 'https://bugs.python.org/TarkaSteve'

bugs.python.org fields:

activity = <Date 2014-02-27.22:38:01.352>
actor = 'jmehnle'
assignee = 'jnoller'
closed = False
closed_date = None
closer = None
components = ['Extension Modules']
creation = <Date 2008-09-11.01:13:58.188>
creator = 'TarkaSteve'
dependencies = []
files = []
hgrepos = []
issue_num = 3831
keywords = []
message_count = 6.0
messages = ['73001', '73019', '73069', '85168', '107371', '133358']
nosy_count = 9.0
nosy_names = ['forest', 'roudkerk', 'jnoller', 'TarkaSteve', 'asksol', 'anacrolix', 'andresfreund', 'jmehnle', 'barcc']
pr_nums = []
priority = 'low'
resolution = None
stage = None
status = 'open'
superseder = None
type = 'enhancement'
url = 'https://bugs.python.org/issue3831'
versions = ['Python 3.2']

[TarkaSteve]

Both Connection and Pipe objects expose their underlying file
descriptors, which is useful for accessing them in an event-driven
manner. However the higher-level Queue does not make the underlying
pipe available; to get at them you must access private member attributes
which is fragile. It would be good to have a well-defined API for
accessing either the pipe objects or the underlying FDs.

[jnoller]

Steve, I'm a little nervous about exposing the underlying Queue pipes in
an official API simply due to the fact that it is an advanced use case,
and we do want to keep the API simple, and relatively close to the core
Queue implementation.

Do you have an example use-case/code? This will not be making it into
python 2.6/3.0 - it will need to wait until 2.7 and 3.1 at very least.

[TarkaSteve]

Hi Jesse,

The use-case I had is mind is enabling asyncronous (i.e. select() style)
notification of data being available on the queue, which is more elegant
(and efficient) than polling with get(). Example code showing how this
works with GTK is available here:

http://haltcondition.net/?p=2319

I understand wanting to keep the API simple, however the underlying
posix-level file-descriptors for pipes and connections are already
exposed via fileno(); the actual API change would just be a matter of
changing '_reader' and '_writer' to 'reader' and 'writer' (implicitly
guaranteeing their consistency).

Would it help if I attached a patch showing the proposed change to the
module? I realise the API is frozen now but it would be nice to have
this further down the line.

[jnoller]

Can you please provide an example w.r.t to how you would handle the case
where poll()/recv returns partial bytes of the object instead of a full
object?

[andresfreund]

As soon as some bytes are signalled as being available one can simply do a normal get(). I don't really see the problem here?
Sure, the get() might not be completely non-blocking (especially if the transferred event is more than the size of a pipe-buffer) but I have a hard time seing that as a problem as that should be both rare and only last a short time.

My personal use-case is being able to efficiently wait for evens from different queues - using the standard api one currently can only do that by busy looping...

The biggest thing I see where you have to be careful here is some stomping herd phenomenon you will get into if you have multiple readers doing a poll().
Namely *all* off those processes will awake and run into .get() which isnt exactly nice, but thats hardly solvable on python level.

[anacrolix]

I look forward to this, or something similar. Inspiration can be taken from Golangs's select behaviour on channels.

select {
case i1 = <-c1:
print("received ", i1, " from c1\n")
case c2 <- i2:
print("sent ", i2, " to c2\n")

default:
print("no communication\n")
}

[tapetersen]

This would be useful much as commented in code using the style of go of waiting for multiple channels.
There is already precedence in exposing this for multiprocessing.connection.Connection objects that have the same issue with early notification of a partial object. Being able to add multiprocessing.Queue as well as Pipe and Connection objects in multiprocessing.wait` would be useful and avoid awkward event handling or polling.

Maybe exposing the reader is too much but either adding a fileno() function that would allow it to be used by most select apis or following the example of multiprocessing.Popenand exposing a sentinel that only guarantees that it is awaitable by multiprocessing.connection.wait() seems reasonable without exposing too much internals.

[tapetersen]

The same would also be true for the rest of the synchronisation primitives in the multiprocessing library. Right now the only way to do this seems to be to resort to having threads waiting for each non-selectable primitive that can then either set a central Event or post to a Pipe/Connection that can be selected on.