Python equivalent of Golang's select on channels

Go has a select statement that works on channels. From the documentation:

> The select statement lets a goroutine wait on multiple communication
> operations.
>
> A select blocks until one of its cases can run, then it executes that
> case. It chooses one at random if multiple are ready.

Is there a Python equivalent of the following code:

package main

import "fmt"

func main() {
	c1 := make(chan int)
	c2 := make(chan int)
	quit := make(chan int)

	go func() {
		for i := 0; i < 10; i++ {
			c1 <- i
		}
		quit <- 0
	}()

	go func() {
		for i := 0; i < 2; i++ {
			c2 <- i
		}
	}()

	for {
		select {
		case <-c1:
			fmt.Println("Received value from c1")
		case <-c2:
			fmt.Println("Received value from c2")
		case <-quit:
			fmt.Println("quit")
			return
		}
	}
}

Output of this program:

Received value from c1
Received value from c1
Received value from c2
Received value from c1
Received value from c2
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from c1
Received value from c1
quit

Here's a pretty direct translation, but the "choosing which if multiple are ready" part works differently - it's just taking what came in first. Also this is like running your code with gomaxprocs(1).

import threading
import Queue

def main():
    c1 = Queue.Queue(maxsize=0)
    c2 = Queue.Queue(maxsize=0)
    quit = Queue.Queue(maxsize=0)

    def func1():
        for i in range(10):
            c1.put(i)
        quit.put(0)

    threading.Thread(target=func1).start()

    def func2():
        for i in range(2):
            c2.put(i)

    threading.Thread(target=func2).start()

    combined = Queue.Queue(maxsize=0)

    def listen_and_forward(queue):
        while True:
            combined.put((queue, queue.get()))

    t = threading.Thread(target=listen_and_forward, args=(c1,))
    t.daemon = True
    t.start()
    t = threading.Thread(target=listen_and_forward, args=(c2,))
    t.daemon = True
    t.start()
    t = threading.Thread(target=listen_and_forward, args=(quit,))
    t.daemon = True
    t.start()

    while True:
        which, message = combined.get()
        if which is c1:
            print 'Received value from c1'
        elif which is c2:
            print 'Received value from c2'
        elif which is quit:
            print 'Received value from quit'
            return
main()

The basic change is simulating the select with threads that combine messages. If you were going to use this pattern much, you might write some select code:

import threading
import Queue

def select(*queues):
    combined = Queue.Queue(maxsize=0)
    def listen_and_forward(queue):
        while True:
            combined.put((queue, queue.get()))
    for queue in queues:
        t = threading.Thread(target=listen_and_forward, args=(queue,))
        t.daemon = True
        t.start()
    while True:
        yield combined.get()

def main():

    c1 = Queue.Queue(maxsize=0)
    c2 = Queue.Queue(maxsize=0)
    quit = Queue.Queue(maxsize=0)

    def func1():
        for i in range(10):
            c1.put(i)
        quit.put(0)

    threading.Thread(target=func1).start()

    def func2():
        for i in range(2):
            c2.put(i)

    threading.Thread(target=func2).start()

    for which, msg in select(c1, c2, quit):
        if which is c1:
            print 'Received value from c1'
        elif which is c2:
            print 'Received value from c2'
        elif which is quit:
            print 'Received value from quit'
            return
main()

But...

Note that this select isn't quite the go one, though it doesn't matter for your program - a goroutine could send a result on a channel that would be queued up in the select and lost if we didn't always iterate over the select to completion!

Also consider the offset library by Benoit Chesneau. It is a port of the Go concurrency model to Python, using fibers under the covers.

He gave a presentation about this at PyCon APAC 2013:

• Slides
• Video

You can use multiprocessing.Pipe instead of chan, threading.Thread instead of go and select.select instead of select.

Here's a reimplementation of your go example in Python using this approach:

import random
from multiprocessing import Pipe
from select import select
from threading import Thread


def main():
    c1_r, c1_w = Pipe(duplex=False)
    c2_r, c2_w = Pipe(duplex=False)
    quit_r, quit_w = Pipe(duplex=False)

    def func1():
        for i in range(10):
            c1_w.send(i)
        quit_w.send(0)

    Thread(target=func1).start()

    def func2():
        for i in range(2):
            c2_w.send(i)

    Thread(target=func2).start()

    while True:
        ready, _, _ = select([c1_r, c2_r, quit_r], [], [])
        which = random.choice(ready)
        if which == c1_r:
            c1_r.recv()
            print 'Received value from c1'
        elif which == c2_r:
            c2_r.recv()
            print 'Received value from c2'
        elif which == quit_r and len(ready) == 1:
            quit_r.recv()
            print 'Received value from quit'
            return

if __name__ == '__main__':
    main()

This implementation is based upon @Thomas's implementation, but unlike @Thomas's it doesn't spawn extra threads to perform the select.

Tested on Linux with Python 2.7.13. Windows may behave differently as select is a Unixy thing.

Edit: I added the len(ready) == 1 condition so quit is only handled after the other pipes are drained. This isn't required in Go as the channels are zero sized, so func1 can't send a message to quit_w until after the message sent to c1_w has been received. Thanks to the comment by @Sean Perry.

With Python 3.5 there are the keywords async and await which make it possible to have functions which can be suspended in execution and thus are able to run on an evenloop instead of threads. The asyncio std lib is offering one.

To more directly map the behaviour of Go blocking channels and select you might make use of this small library and then your example code would look very similar in Python.

Yes, all are possible with goless. You can try it.

Have Fun

Here is an example:

c1 = goless.chan()
c2 = goless.chan()

def func1():
    time.sleep(1)
    c1.send('one')
goless.go(func1)

def func2():
    time.sleep(2)
    c2.send('two')
goless.go(func2)

for i in range(2):
    case, val = goless.select([goless.rcase(c1), goless.rcase(c2)])
    print(val)

Here's another, an attempt at imitating the go syntax:

from threading import Thread
from Queue import Queue

def main():

    c1 = Queue.Queue(maxsize=0)
    c2 = Queue.Queue(maxsize=0)
    quit = Queue.Queue(maxsize=0)

    Thread(target=lambda: [c1.put(i) for i in range(10)] or quit.put(0)).start()
    Thread(target=lambda: [c2.put(i) for i in range(2)]).start()

    for which, msg in select(c1, c2, quit):
        if which is c1:
            print 'Received value from c1'
        elif which is c2:
            print 'Received value from c2'
        elif which is quit:
            print 'Received value from quit'
            return

def select(*queues):
    combined = Queue.Queue(maxsize=0)
    def listen_and_forward(queue):
        while True:
            combined.put((queue, queue.get()))
    for queue in queues:
        t = Thread(target=listen_and_forward, args=(queue,))
        t.daemon = True
        t.start()
    while True:
        yield combined.get()

main()

For completeness: Go-style channels, including working select are available as part of pygolang:

ch1 = chan()    # synchronous channel
ch2 = chan(3)   # channel with buffer of size 3

def _():
    ch1.send('a')
    ch2.send('b')
go(_)

ch1.recv()      # will give 'a'
ch2.recv_()     # will give ('b', True)

_, _rx = select(
    ch1.recv,           # 0
    ch2.recv_,          # 1
    (ch2.send, obj2),   # 2
    default,            # 3
)
if _ == 0:
    # _rx is what was received from ch1
    ...
if _ == 1:
    # _rx is (rx, ok) of what was received from ch2
    ...
if _ == 2:
    # we know obj2 was sent to ch2
    ...
if _ == 3:
    # default case
    ...

offset (see https://stackoverflow.com/a/19143696/9456786) also seems interesting.

goless (see https://stackoverflow.com/a/39269599/9456786), unfortunately, has weak select implementation, which by design does not work properly on synchronous channels.

There are several answers here that use queue.Queue and threading.Thread to simulate the select behaviour but it's not necessary. You can extend queue.Queue like this:

import queue
import os
import select

class EQueue(queue.Queue):
    def __init__(self, *args, **kwargs)
        self._fd = os.eventfd(flags=0x00004001)
        super().__init__(*args, **kwargs)

    def put(self, *args, **kwargs):
        super().put(*args, **kwargs)
        eventfd_write(self._fd, 1)

    def get(self, *args, **kwargs):
        os.eventfd_read(self._fd)
        super().get(*args, **kwargs)

    def fileno(self):
        return self._fd

    def __del__(self):
        os.close(self._fd)

This adds an extra semaphore around the queue and, crucially, one that is accessible through a file descriptor. This means you can now wait on this queue with select.select(). So the above examples that use queues and threads can be rewritten without the extra threads:

def main():

    c1 = EQueue(maxsize=0)
    c2 = EQueue(maxsize=0)
    quit = EQueue(maxsize=0)

    def func1():
        for i in range(10):
            c1.put(i)
        quit.put(0)

    threading.Thread(target=func1).start()

    def func2():
        for i in range(2):
            c2.put(i)

    threading.Thread(target=func2).start()

    rx, _, _ = select.select([c1, c2, quit], [], []):
        if c1 in rx:
            msg = c1.get()
            print 'Received value from c1'
        elif c2 in rx:
            msg = c2.get()
            print 'Received value from c2'
        elif quit in rx:
            print 'Received value from quit'
            return
main()

The main function here is fairly similar to that given by @alkasm above but there is no custom implementation of select and no thread-per-queue to collect all the separate queues into one; it relies on the operating system to tell you when a queue has items available.

Note that os.eventfd was only added in Python 3.10 but implementing it in ctypes is fairly trivial or there is the eventfd package on PyPI. The latter also supports Windows, unlike the other options, simulating eventfds with pipes. The python doco claims that eventfds are only available on Linux systems running glibc >= 2.8 but muslc also supports them.