python - チャネルでの Golang の select に相当する Python

Question

Go には、チャネルで機能する select ステートメントがあります。ドキュメントから：

select ステートメントにより、ゴルーチンは複数の通信操作を待機できます。

select は、ケースの 1 つが実行できるようになるまでブロックし、その後、そのケースを実行します。複数の準備ができている場合は、ランダムに 1 つを選択します。

次のコードに相当する Python はありますか。

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
        }
    }
}

このプログラムの出力:

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

score 16 · Accepted Answer

これはかなり直接的な翻訳ですが、「複数の準備ができているかどうかを選択する」部分の動作は異なります-最初に来たものを取得するだけです. また、これはでコードを実行するようなもの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()

基本的な変更は、メッセージを結合するスレッドで選択をシミュレートすることです。このパターンを頻繁に使用する場合は、いくつかの選択コードを記述できます。

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()

しかし...

あなたのプログラムにとっては問題ではありませんが、この選択は完全に正しいものではないことに注意してください - ゴルーチンは選択でキューに入れられ、常に反復しなければ失われる結果をチャンネルに送信する可能性があります。選択して完成！

score 12 · Accepted Answer

また、Benoit Chesneau によるオフセットライブラリも検討してください。これは、カバーの下でファイバーを使用して、Go 同時実行モデルを Python に移植したものです。

彼は PyCon APAC 2013 でこれについてプレゼンテーションを行いました。

score 9 · Accepted Answer

multiprocessing.Pipeの代わりにchan、のthreading.Thread代わりに、goおよびのselect.select代わりに使用できますselect。

このアプローチを使用した Python での go の例の再実装は次のとおりです。

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()

この実装は @Thomas の実装に基づいていますが、@Thomas の実装とは異なり、選択を実行するために余分なスレッドを生成しません。

Linux で Python 2.7.13 を使用してテスト済み。select は Unixy のものであるため、Windows では動作が異なる場合があります。

編集：条件を追加したlen(ready) == 1ので、終了は他のパイプが排出された後にのみ処理されます。チャネルのサイズがゼロであるため、これは Go では必要ありません。そのため、に送信されたメッセージが受信されるまでfunc1メッセージを送信できません。@Sean Perry のコメントに感謝します。quit_wc1_w

score 4 · Accepted Answer

はい、すべてgolessで可能です。あなたはそれを試すことができます。

楽しんで ;-）

次に例を示します。

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)

score 4 · Accepted Answer

Python 3.5 にはasyncandというキーワードawaitがあり、実行中に一時停止できる関数を作成できるため、スレッドではなく偶数ループで実行できます。asyncio標準ライブラリはそれを提供しています。

Go ブロッキングチャネルの動作をより直接的にマッピングするには、この小さなライブラリselectを利用すると、サンプルコードは Python で非常によく似たものになります。

score 2 · Accepted Answer

これは、go 構文を模倣する別の試みです。

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()

python - チャネルでの Golang の select に相当する Python

7 に答える 7

しかし...

Related

Reference