Python I/Oの複数待機について

Pythonで、名前付きパイプを4つ用意し、４つ同時に待機させ、一つでも書き込みが行われたら、ディスコードに通知を飛ばすプログラムを作成したいです。

普通にコードを書くと、名前付きパイプを４つ用意してしまうと、４つすべてにデータが書き込まれないと、処理が行われずやりたいことができませんでした。

そこで、スレッドを使って、並列処理を行おうと思い、添付のコードを書きました。

これで、３つまでは同時待機させることができ、意図する処理を行うことができたのですが、
４つ目が動作しなくなりました。
スレッドは、同時に処理できる数に制限があるのでしょうか？

また、そんな場合、どういう処理をすれば、制限なく、複数のパイプを個別で待機させることができるのでしょうか？
処理委成功すれば、１５個くらい同時待機させたいと考えています。

Python
1コード
2
3from discordwebhook import Discord
4import win32pipe
5import win32file
6import threading
7
8discord = Discord(url="URL")
9
10def test1():
11    test1_text = win32pipe.CreateNamedPipe(
12    r'\\.\pipe\test1_symbol', 
13    win32pipe.PIPE_ACCESS_DUPLEX,
14    win32pipe. PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE | win32pipe.PIPE_WAIT,
15    1, 256, 256, 0, None)
16    
17    win32pipe.ConnectNamedPipe(test1_text, None)
18    
19    buffer1 = b''
20    
21    while True:
22        hr, c_test1 = win32file.ReadFile(test1_text, 1)
23        buffer1 += c_test1
24        
25        if c_test1 == b'\n':
26            
27            discord.post(content='{}'.format(str(buffer1)))
28            buffer1 = b''
29        
30def test2():
31    test2_text = win32pipe.CreateNamedPipe(
32    r'\\.\pipe\test2_symbol', 
33    win32pipe.PIPE_ACCESS_DUPLEX,
34    win32pipe. PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE | win32pipe.PIPE_WAIT,
35    1, 256, 256, 0, None)
36    
37    win32pipe.ConnectNamedPipe(test2_text, None)
38    
39    buffer2 = b''
40    
41    while True:
42        hr, c_test2 = win32file.ReadFile(test2_text, 1)
43        buffer2 += c_test2
44        
45        if c_test2 == b'\n':
46            
47            discord.post(content='{}'.format(str(buffer2)))
48            buffer2 = b''
49        
50def test3():
51    test3_text = win32pipe.CreateNamedPipe(
52    r'\\.\pipe\test3_symbol', 
53    win32pipe.PIPE_ACCESS_DUPLEX,
54    win32pipe. PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE | win32pipe.PIPE_WAIT,
55    1, 256, 256, 0, None)
56    
57    win32pipe.ConnectNamedPipe(test3_text, None)
58    
59    buffer3 = b''
60    
61    while True:
62        hr, c_test3 = win32file.ReadFile(test3_text, 1)
63        buffer3 += c_test3
64        
65        if c_test3 == b'\n':
66            
67            discord.post(content='{}'.format(str(buffer3)))
68            buffer3 = b''
69        
70def test4():
71    test4_text = win32pipe.CreateNamedPipe(
72    r'\\.\pipe\test4_symbol', 
73    win32pipe.PIPE_ACCESS_DUPLEX,
74    win32pipe. PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE | win32pipe.PIPE_WAIT,
75    1, 256, 256, 0, None)
76    
77    win32pipe.ConnectNamedPipe(test4_text, None)
78    
79    buffer4 = b''
80    
81    while True:
82        hr, c_test4 = win32file.ReadFile(test4_text, 1)
83        buffer4 += c_test4
84        
85        if c_test4 == b'\n':
86            
87            discord.post(content='{}'.format(str(buffer4)))
88            buffer4 = b''
89            
90t1 = threading.Thread(target=test1)
91t2 = threading.Thread(target=test2)
92t3 = threading.Thread(target=test3)
93t4 = threading.Thread(target=test4)    
94        
95t1.start()
96
97t2.start()
98
99t3.start()
100
101t4.start()
102

お詳しい方、ご教授お願いいたします。

y_waiwai

2022/03/28 22:58

このままではコードが読めないので、質問を編集し、</>(コードの挿入)ボタンを押し、出てくる’’’の枠の中にコードを貼り付けてください

shinya117

2022/03/29 01:21

ご丁寧にありがとうございます。よろしくお願いします。

bsdfan

2022/03/29 08:09

関数名とパイプの変数名がごちゃごちゃになっていますが、実際に動作させているものでは、そこは大丈夫ですか？ > ４つ目が動作しなくなりました。というのが、具体的にどういうことなのかも書いたほうがいいです。何をやって、どういう結果を期待していたが、どうなったのか。

shinya117

2022/03/29 08:59

質問用に、例として作成したもので、誤りがありましたので修正しました。ご指摘ありがとうございます。パイプを4つ用意し、クライアント側から書き込みがあったらディスコードに表示したいです。現在、マルチスレッドを使って、3つ同時待機させることができていますが、スレッドを使うと、限界値が3つなのか分かりませんが、4つ目がパイプへの入力があっても、出力できない状態です。 4つ同時待機で、①、②、③、④とあったときに、 ①に書き込まれたら1に書き込まれた内容を都度処理。①だけに書き込まれたのであれば、②、③、④は引き続き待機したいと言うのがやりたいことです。ただ、現在は①に書き込まれた時は他は待機、 ②に書き込まれた時も同様。 ③に書き込まれた時も同様。 ④に書き込まれた時は、④の出力の動作ができておらず、待機自体ができてないと思っています。複数(4つより大きい数)のパイプを個別で、同時待機させたいです。

bsdfan

2022/03/29 10:17

すでに回答ある通り、きちんと書けば問題なく動くはずなので、 - スレッドのスタートの順番を変える - テストのクライアントから書き込む順番を変えるなどしてみて、何か変わるか見てみるといいのかも。

行動規範の内容に同意します

回答1件

ベストアンサー

パイプを３０個作成した例です。
あなたが、提示されたソースにいくつか誤りがあったので、修正してあります。
サーバー側（あなたが作成しているスクリプト）

python3
1import win32pipe
2import win32file
3import threading
4
5
6def test(t_no):
7    print("test%d started" % t_no)
8    str = r'\\.\pipe\test%d_symbol' % t_no
9    ph = win32pipe.CreateNamedPipe(
10    str,
11    win32pipe.PIPE_ACCESS_DUPLEX,
12    win32pipe.PIPE_TYPE_BYTE | win32pipe.PIPE_READMODE_BYTE | win32pipe.PIPE_WAIT,
13    1, 256, 256, 0, None)
14    win32pipe.ConnectNamedPipe(ph, None)
15    buffer = b''
16
17    while True:
18        hr, c_test = win32file.ReadFile(ph, 1)
19        buffer += c_test
20        if c_test == b'\n':
21            str = buffer.decode('utf-8')
22            print("t_no={0} rcv={1}".format(t_no,str))
23            buffer = b''
24
25ths = []
26for i in range(1,31):
27    th = threading.Thread(target=test,args=(i,))
28    ths.append(th)
29
30for th in ths:
31    th.start()
32

クライアント側（パイプへ書き込む側）

import win32pipe
import win32file

phs = []
for i in range(1,31):
    str = r'\\.\pipe\test%d_symbol' % i
    ph = win32file.CreateFile(
    str,
    win32file.GENERIC_WRITE,
    0,
    None,
    win32file.OPEN_EXISTING,
    win32file.FILE_ATTRIBUTE_NORMAL,
    None)
    phs.append(ph)

print("START")
while True:
    no_c = input("No>>")
    no = int(no_c)
    if no == 99:
        break
    if no < 1 or no > 30:
        continue
    str = input("String>>")
    str += '\n'
    bdata = str.encode('utf-8')
    ph = phs[no-1]
    win32file.WriteFile(ph, bdata)

for i in range(1,11):
    phs[i].Close()

実行結果
サーバー側
test1 started
test2 started
test3 started
test4 started
test5 started
test6 started
test7 started
test8 started
test9 started
test10 started
test11 started
test12 started
test13 started
test14 started
test15 started
test16 started
test17 started
test18 started
test19 started
test20 started
test21 started
test22 started
test23 started
test24 started
test25 started
test26 started
test27 started
test28 started
test29 started
test30 started
t_no=1 rcv=AAAAAAAAAAAAA

t_no=2 rcv=BBBBBBBBBBBBBB

t_no=29 rcv=XXXXXXXXXXXXXXXX

t_no=30 rcv=ZZZZZZZZZZZZZZ

クライアント側
START
No>>1
String>>AAAAAAAAAAAAA
No>>2
String>>BBBBBBBBBBBBBB
No>>29
String>>XXXXXXXXXXXXXXXX
No>>30
String>>ZZZZZZZZZZZZZZ
No>>99

最初にサーバー側を起動します。
次に、クライアント側を起動します。
No>>のあとに、数字を入力します。
（１～30の数値です、１はスレッドのパイプへ書き込み、30はスレッド30のパイプへ書き込みます）
99を入力すると、終了します。
String>>のあとに、文字列を入力します。
入力した文字列が、サーバー側で読み取られ、画面に表示されます。
サーバー側は、文字列の表示のみ行っています。
ディスコードに通知を飛ばすところは、あなたのほうで組み込んでください。

投稿2022/03/29 09:25

tatsu99

総合スコア5438

shinya117

2022/03/29 21:17

ご丁寧にコードまで記載して頂きありがとうございます。とてもわかりやすかったです。今回、クライアント側はC言語で書いております。詳しくお話しすると、MQL4という言語です。質問させていただいたのは、スレッドが3つしか使えないと思っていましたが、今回のご回答でスレッドの制限が原因ではないことがわかりました。というのも、異なるパイプに一つづつ書き込むのは成功しますが、スレッドで準備した30のパイプ全てに書き込むと、処理が3つしか実行されない。ということが起こっています。 C言語のコードを添付させて頂き、見ていただくことは可能でしょうか。申し訳ありませんが、もし宜しければお願いしたいと思っております。よろしくお願いいたします。

行動規範の内容に同意します