teratail
回答率
85.36%
teratail header banner
teratail header banner
質問するログイン新規登録
トップに関する質問Python　並列処理によるソケット通信　高速化

編集履歴

質問編集履歴

4

再度変更

2021/02/17 02:54

投稿

goki_gottan
goki_gottan

スコア168

test CHANGED
File without changes
test CHANGED
@@ -162,7 +162,7 @@
162
162
 
163
163
 
164
164
 
165
- with futures.ThreadPoolExecutor() as executor:
165
+ with futures.ThreadPoolExecutor(2) as executor:
166
166
 
167
167
 
168
168
 

3

指摘があったため

2021/02/17 02:54

投稿

goki_gottan
goki_gottan

スコア168

test CHANGED
File without changes
test CHANGED
@@ -60,126 +60,132 @@
60
60
 
61
61
  def func1(path1):
62
62
 
63
+ global j1
64
+
65
+ j1=j1+1
66
+
67
+
68
+
69
+ if j==1:
70
+
71
+ tcp_client1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
72
+
73
+ tcp_client1.connect((target_ip1, target_port1))
74
+
75
+
76
+
77
+ v = "***," + str(path1)
78
+
79
+ tcp_client1.send(v.encode(encoding='utf-8'))
80
+
81
+
82
+
83
+ response1 = tcp_client1.recv(buffer_size)
84
+
85
+ response1 = response1.decode(encoding='utf-8')
86
+
87
+
88
+
89
+ if j1==last:
90
+
91
+ tcp_client1.shutdown(socket.SHUT_RDWR)
92
+
93
+ tcp_client1.close()
94
+
95
+
96
+
97
+
98
+
99
+ def func2(path2):
100
+
101
+ global j2
102
+
103
+ j2 = j2+1
104
+
105
+
106
+
107
+ if j2==1:
108
+
109
+ tcp_client2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
110
+
111
+ tcp_client2.connect((target_ip2, target_port2))
112
+
113
+
114
+
115
+ v = "***," + str(path2)
116
+
63
117
 
64
118
 
65
- v = "***," + str(path1)
66
-
67
- tcp_client1.send(v.encode(encoding='utf-8'))
119
+ tcp_client2.send(v.encode(encoding='utf-8'))
68
-
69
-
70
-
120
+
121
+
122
+
71
- response1 = tcp_client1.recv(buffer_size)
123
+ response2 = tcp_client2.recv(buffer_size)
72
-
124
+
73
- response1 = response1.decode(encoding='utf-8')
125
+ response2 = response2.decode(encoding='utf-8')
74
-
75
-
76
-
77
-
78
-
126
+
127
+
128
+
79
- def func2(path2):
129
+ if j2==last:
80
-
81
-
82
-
130
+
83
- v = "***," + str(path2)
131
+ tcp_client2.shutdown(socket.SHUT_RDWR)
132
+
133
+ tcp_client2.close()
84
134
 
85
135
 
86
136
 
87
- tcp_client2.send(v.encode(encoding='utf-8'))
137
+ paths = list(input_dir.glob("**/*.jpg"))
138
+
88
-
139
+ border, last = len(paths)//2, len(paths)-len(paths)%2
140
+
141
+
142
+
89
-
143
+ buffer_size = 4096
144
+
145
+
146
+
90
-
147
+ t1=time.time()
148
+
149
+
150
+
151
+ future_list = []
152
+
153
+
154
+
155
+ global j1
156
+
157
+ global j2
158
+
159
+ j1=0
160
+
161
+ j2=0
162
+
163
+
164
+
165
+ with futures.ThreadPoolExecutor() as executor:
166
+
167
+
168
+
169
+ for path1, path2 in zip(paths[:border], paths[border:last]):
170
+
171
+
172
+
173
+ # タスクを追加する。
174
+
91
- response2 = tcp_client2.recv(buffer_size)
175
+ future1 = executor.submit(func1, str(path1))
92
-
176
+
93
- response2 = response2.decode(encoding='utf-8')
177
+ future2 = executor.submit(func2, str(path2))
94
178
 
95
179
 
96
180
 
97
- paths = list(input_dir.glob("**/*.jpg"))
98
-
99
- border, last = len(paths)//2, len(paths)-len(paths)%2
100
-
101
-
102
-
103
- buffer_size = 4096
104
-
105
-
106
-
107
- tcp_client1 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
108
-
109
- tcp_client2 = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
110
-
111
-
112
-
113
- try:
114
-
115
- tcp_client1.connect((target_ip1, target_port1))
116
-
117
- tcp_client2.connect((target_ip2, target_port2))
118
-
119
- except ConnectionRefusedError:
120
-
121
- print("接続不良")
122
-
123
- sys.exit()
124
-
125
-
126
-
127
- t1=time.time()
128
-
129
-
130
-
131
- future_list = []
132
-
133
-
134
-
135
- with futures.ThreadPoolExecutor() as executor:
136
-
137
-
138
-
139
- for path1, path2 in zip(paths[:border], paths[border:last]):
140
-
141
-
142
-
143
- # タスクを追加する。
144
-
145
- future1 = executor.submit(func1, str(path1))
146
-
147
- # future1.setDaemon(True)
148
-
149
- future2 = executor.submit(func2, str(path2))
150
-
151
-
152
-
153
181
  # Future オブジェクトを記録する。
154
182
 
155
183
  future_list.append(future1)
156
184
 
157
185
  future_list.append(future2)
158
186
 
159
- if len(paths)%2 == 1:
160
-
161
- future3 = executor.submit(func2, str(paths[-1]))
162
-
163
- future_list.append(future3)
164
-
165
187
 
166
188
 
167
-
168
-
169
- tcp_client1.shutdown(socket.SHUT_RDWR)
170
-
171
- tcp_client1.close()
172
-
173
-
174
-
175
-
176
-
177
- tcp_client2.shutdown(socket.SHUT_RDWR)
178
-
179
- tcp_client2.close()
180
-
181
-
182
-
183
189
  t2 = time.time()
184
190
 
185
191
 

2

回答が得られないため

2021/02/17 02:52

投稿

goki_gottan
goki_gottan

スコア168

test CHANGED
File without changes
test CHANGED
@@ -1,10 +1,24 @@
1
+ 回答がなかなか得られなかったので、ちょっと質問を変えさせていただきます。
2
+
3
+ ソケットが込み合っている可能性があり、
4
+
5
+ https://qiita.com/maueki/items/8f1e190681682ea11c98
6
+
7
+ ↑こちらのサイトのようにブロッキング、ノンブロッキングを入れる必要があるのでしょうか。
8
+
9
+
10
+
1
- JAVAからPythonの入れ替えのため、ソケット通信にて、ソフト2個を並列処理にて動作させようとしておりますが、
11
+ Pythonで、ソケット通信にて、ソフト2個を並列処理にて動作させようとしておりますが、
2
12
 
3
13
  並列処理はできておるのですが、非常に重く、遅いです。また、固まり、最終的にソフトが落ちるときがあります。
4
14
 
5
15
 
6
16
 
7
17
  １個では落ちません。おそらく、並列処理が悪さしてると思うのですが、通信をどのようにしたら良いか、socket以外のライブラリー、または通信待機など入れるべきでしょうか？
18
+
19
+
20
+
21
+
8
22
 
9
23
 
10
24
 

1

2021/02/13 01:44

投稿

goki_gottan
goki_gottan

スコア168

test CHANGED
File without changes
test CHANGED
@@ -1,6 +1,10 @@
1
1
  JAVAからPythonの入れ替えのため、ソケット通信にて、ソフト2個を並列処理にて動作させようとしておりますが、
2
2
 
3
3
  並列処理はできておるのですが、非常に重く、遅いです。また、固まり、最終的にソフトが落ちるときがあります。
4
+
5
+
6
+
7
+ １個では落ちません。おそらく、並列処理が悪さしてると思うのですが、通信をどのようにしたら良いか、socket以外のライブラリー、または通信待機など入れるべきでしょうか？
4
8
 
5
9
 
6
10