常連

### 前提・実現したいこと

データセットがキーのアルファベット記号と要素の数字からなる辞書型で与えられており、
数字はリストに格納されています。リストの長さはキーによって様々です。
各キーの要素の数字間でそれぞれ、たすき掛けで掛け算し、最も大きいものを採用した上で、
配列の長さで割った値を２つのキー間の値として出力します。

これを大規模なデータセットで実行することを実現したいです。
[関連する前回の質問](https://teratail.com/questions/139409)

### 発生している問題・エラーメッセージ
小規模なデータセットでは問題なく実行できますが、
このコードで、データセットのキーの数が大規模（例えば3万など）な場合、
１つのキーに対してたすき掛けの計算で比較する対象も膨大になるため、
現行のコードのまま実行すると膨大な時間がかかってしまいます。

そこで、並列処理を行うために、以下のコードを書いて、
辞書のキー別に平行で実行しようとしています。
現在のコードではキー別に実行する方法がわからないため、まず、全ての処理を３つのスレッドで平行して行う形式になっていますが、それでも以下のエラーが出てしまっている状況です。

```
#３つのスレッドで同じエラーが出ている状況です
Exception in thread Thread-8:
Traceback (most recent call last):
  File "/Users/username/anaconda/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/Users/username/anaconda/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
TypeError: 'dict' object is not callable

Exception in thread Thread-6:
Traceback (most recent call last):
  File "/Users/username/anaconda/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/Users/username/anaconda/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
TypeError: 'dict' object is not callable

Exception in thread Thread-7:
Traceback (most recent call last):
  File "/Users/username/anaconda/lib/python3.6/threading.py", line 916, in _bootstrap_inner
    self.run()
  File "/Users/username/anaconda/lib/python3.6/threading.py", line 864, in run
    self._target(*self._args, **self._kwargs)
TypeError: 'dict' object is not callable
```

### 該当のソースコード

```python
def compute(data, output):
    for alph, nums in data.items():
        avg = {}
        my_list = data[alph]
        for target_alph, target_nums in data.items():
            target_list = data[target_alph]
            if alph == target_alph:
                continue
            max_nums = []

            for i in my_list:
                max_num = 0
                for j in target_list:
                    result = i * j
                    if result is not None and result > max_num:
                        max_num = result
                max_nums.append(max_num) 
            avg[target_alph] = sum(max_nums) / len(max_nums)
            output[alph] = avg
    return output
    
data = {
    'A':[1, 3, 5, 2, 1, 8, 9],
    'B':[9, 4, 3], 
    'C':[8, 5, 5, 6, 1]
}

output = {}

compute(data, output)

thA = threading.Thread(target=compute(data, output)) #data['A']
thB = threading.Thread(target=compute(data, output))#data['B']
thC = threading.Thread(target=compute(data, output)) #data['C']

thA.start(); print ('A started.')
thB.start(); print ('B started.')
thC.start(); print ('C started.')

thA.join(); print ('A completed.')
thB.join(); print ('B completed.')
thC.join(); print ('C completed.')
```

### 試したこと
試したことは、関数化した計算部分のコードと並列処理に関する簡単なコードです。
これらを組み合わせる段階でつまづいています。

関数化した計算部分のコード
```python
def compute(data, output):
    for alph, nums in data.items():
        avg = {}
        my_list = data[alph]
        for target_alph, target_nums in data.items():
            target_list = data[target_alph]
            if alph == target_alph:
                continue
            max_nums = []

            for i in my_list:
                max_num = 0
                for j in target_list:
                    result = i * j
                    if result is not None and result > max_num:
                        max_num = result
                max_nums.append(max_num) 
            avg[target_alph] = sum(max_nums) / len(max_nums)
            output[alph] = avg
    return output
    
data = {
    'A':[1, 3, 5, 2, 1, 8, 9],
    'B':[9, 4, 3], 
    'C':[8, 5, 5, 6, 1]
}

output = {}

compute(data, output)
```
出力
```
{'A': {'B': 37.285714285714285, 'C': 33.142857142857146},
 'B': {'A': 48.0, 'C': 42.666666666666664},
 'C': {'A': 45.0, 'B': 45.0}}
```


[並列処理に関する簡単なコード](https://uxmilk.jp/21092)
```python
import threading
 
def compute():
    x = 0
    times = 2 ** 26
    while x < times:
        x += 1
 
th1 = threading.Thread(target=compute)
th2 = threading.Thread(target=compute)

th1.start(); print '1 started.'
th2.start(); print '2 started.'

th1.join(); print '1 completed.'
th2.join(); print '2 completed.'
```


### 補足情報（FW/ツールのバージョンなど）
Python3.6

Pythonのマルチスレッド処理に関するエラー

### 実現したいこと

① websocketを使ってデータをリアルタイムで取得
② サブプロセスをマルチプロセスで実行
③ ①で取得したデータをpipeで②のサブプロセスに送って処理する

### 前提

PythonでBybitのAPIを叩いています。websocketを用いてデータを逐次取得しながら、サブプロセスで分析するようなコードを書こうとしています。
構成としては、websocketを動かすプロセスはデータを取得することだけをさせて、サブプロセスでは一定時間ごとにそのデータを加工して分析するというように、作業を二つのプロセスに分けようとしています。
とりあえず、全体の動きを見るために、websocketを動かすメインプロセスからpipeで 送られてきたデータを、サブプロセスでプリントすると言うだけのプログラムを書きました。
ここでは、サブプロセスでのデータのプリントを実現したいです。

プラットフォームはmacOSです。
本番環境はLinax（ubuntu）です。mac上でうまくいけば、Linax上でも大丈夫かなと考えていますが、安易でしょうか？

### 発生している問題・エラーメッセージ

特にエラーは発生していませんが、別プロセスからのデータの表示も全くされません。ps でプロセスIDを確認すると、プログラムが停止しているわけではなさそうでした。

### 該当のソースコード

```Python
import json
import websocket
from pprint import pprint
import multiprocessing as mp

class MyWS():
    def __init__(self):
        self.endpoint_public = 'wss://stream.bybit.com/v5/public/linear'
        self.resv_conn, self.send_conn = mp.Pipe()
        self.connect(self.endpoint_public)
        self.child = mp.Process(target=self.create_df, args=(self.resv_conn, ))
        self.child.start()
        self.child.join()

    def connect(self, endpoint):
        self.ws = websocket.WebSocketApp(url = endpoint,
                                    on_open = self.on_open,                                         
                                    on_message = self.on_message,
                                    on_error = self.on_error,
                                    on_close = self.on_close,)
        self.ws.run_forever()

    def on_message(self, ws, message):
        try:
            message = json.loads(message)
            if message.get('data'):
                for d in message.get('data'):
                    self.send_conn.send(d)  ### ここでデータをサブプロセスに送る。
            # pprint(message) ### A ###
        except Exception as e:
            self.on_error(ws, e)

    def on_open(self, ws):
        self.ws.send(json.dumps({"op":"subscribe", "args":["publicTrade.BTCUSDT"]}))
    
    def exit(self):
        self.ws.close()

    def on_error(self, ws, error):
        print(error)
        self.exit()

    def on_close(self, ws, close_status_code, close_msg):
        print("### closed ###")

    # データ処理用のサブプロセス
    def create_df(self, connection):
        while True:
            received_data = connection.recv()
            if received_data:
                print(received_data)


if __name__ == '__main__':  
    myws = MyWS()



```

### 試したこと

関数on_messageの中の「### A ###」の箇所を有効にすると、データは取得できています。
ですので、pipeでうまく送信できていないか、そもそもサブプロセスが動いていないのか、そのような問題だと思います。しかし、問題の切り分けができません。どなたか、websocketとマルチプロセスに詳しい方、教えていただけないでしょうか。

追記
コンストラクタ__init__()のコードの順序を変えました。
```Python
    def __init__(self):
        self.endpoint_public = 'wss://stream.bybit.com/v5/public/linear'
        self.resv_conn, self.send_conn = mp.Pipe()
        # 以下のようにコードの順序を変更
        self.child = mp.Process(target=self.create_df, args=(self.resv_conn, ))
        self.child.start()
        self.child.join()
        self.connect(self.endpoint_public)
```



Pythonでwebsocketのから逐次送ったデータをマルチプロセスで処理したいです

### 前提

Rust で並行処理を行うプログラムを作ろうとしています。

既存の変数を、そのデータをコピーせずに、``Mutex`` や ``Arc`` で包んだり、包みを外したりすることは可能でしょうか？

### 実現したいこと

例えば、次のプログラムを考えます。

```Rust
use std::sync::{Arc, Mutex};
use std::thread;

fn main() {

    let arr = [[0, 1, 2], [3, 4, 5], [6, 7, 8]];
    let lock = Arc::new(Mutex::new(arr)); // <1> ここで、arr をコピーせずに包みたい

    let lock0 = lock.clone();
    let th0 = thread::spawn(move || {
        let mut arr = lock0.lock().unwrap();
        *arr = [[0, 0, 0], [0, 0, 0], [0, 0, 0]];
    });

    th0.join().unwrap();
    println!("{:?}", lock.lock().unwrap()); // [[0, 0, 0], [0, 0, 0], [0, 0, 0]]

    // <2> このあと、lock の包みを外して配列データを使用したい
}
```

コメントの ``<1>`` の箇所で、``Arc<Mutex<[[i32; 3]; 3]>>`` 型を宣言していますが、アドレスを調べたところ、``arr`` のデータがコピーされて、``Mutex`` 内に格納されているようです。ここで、``arr`` の所有権を奪い、コピーを行うことなく、``Arc<Mutex<T>>`` で包む方法はありますでしょうか？

それから、コメントの ``<2>`` の箇所で、データをコピーすることなく包みを外す、つまり、再び ``[[i32; 3]; 3]`` 型を生成するような方法はありますでしょうか？

あるいは、配列コピーを行わずに ``Arc`` や ``Mutex`` を使用するならば、スコープの初めから最後まで一貫して包んでおくことしかできないのでしょうか？

ご回答よろしくお願いします。


Rust で既存の変数を Mutex や Arc で包みたい

# 状況
自作のモジュール内で`coucurrent.futures.ProcessPoolExecutor`を用いて並列処理を実装したいと考えています。
下記のコードで`MAX_WORKERS = 16`として実行したところ、`BrokenProcessPool`が発生しました。

```Python3
from concurrent import futures

class ExampleClass:
    def hoge(self):
        with futures.ProcessPoolExecutor(max_workers=MAX_WORKERS) as executor:
            results = executor.map(ExampleClass.fuga, arg)

    @staticmethod
    def fuga(arg):
        #何かしらの処理
```
発生するエラー
`concurrent.futures.process.BrokenProcessPool: A process in the process pool was terminated abruptly while the future was running or pending.`

# 試したこと
自分で原因を探ったところ、OSがプロセスをkillしている可能性があるとのことで、プロセス数を減らせばエラーが発生しないと考え`MAX_WORKERS = 8`とした場合には`BrokenProcessPool`は発生しませんでした。
また、公式のドキュメントを読んだところ、ProcessPoolExecutorの項で「If max_workers is None or not given, it will default to the number of processors on the machine.」とあったため`MAX_WORKERS = None`としたところ、`BrokenProcessPool`が発生しました。

また、`concurrent.futures.ProcessPoolExecutor`で実装する前に`multiprocessing.Pool`で実装していたのですが、その際は`MAX_WORKERS = 16`としてもエラーは発生しませんでした。
```
from multiprocessing import Pool
    ~~略~~
        with Pool(MAX_WORKERS) as p:
            results = p.map(ExampleClass.fuga, arg)
    ~~略~~
```

タスクマネージャーでCPUの使用率を観測しながら実行してみたところ、CPU使用率が100%に近づいた場合に`BrokenProcessPool`が発生するような印象を受けました。`MAX_WORKERS = 8`とした場合のCPU使用率は60~70%ほどでした。

# 知りたいこと
0. `BrokenProcessPool`が発生する原因、条件が知りたい。

0. `multiprocessing.Pool`では実行できるのに、`concurrent.futures.ProcessPoolExecutor`では実行できない理由が知りたい。
0. `MAX_WORKERS`の最適な数(エラーが発生せずに安全にタスクを完了できる数)の目安が知りたい。
0. 安全に並列処理を実行できる実装が知りたい。現在は`MAX_WORKERS = 8`とすれば一応動くが、他のタスクの実行中や現在のマシンよりも性能の劣るマシン上での実行などで、使用できるリソースが少ない状況でもエラーを出さずに実行したい。

また、「使用しているCPUは8コア16スレッドなので、`MAX_WORKERS`を増加させると`MAX_WORKERS = 16`まではタスクが高速になる可能性がある。」という認識は正しいでしょうか。

並列処理にはあまり詳しくないため、前提知識などが抜け落ちていた場合はご教示いただければ幸いです。
よろしくおねがいします。

# 環境
**CPU**
	Intel(R) Core(TM) i7-10700F CPU @ 2.90GHz

	基本速度:	2.90 GHz
	ソケット:	1
	コア:	8
	論理プロセッサ数:	16
	仮想化:	有効

**OS**
Windows10

**Python ver.**
Python 3.10.0

[Python]並列処理でプロセス数を増やすとエラーが発生する

#concurrent.futuresを用いてマルチプロセスを行い、その際プロセス間で変数の共有を行いたいです。

threadingやTreadPoolExecutorを用いたマルチスレッドでは実現できましたが、
ProcessPoolExecutorの使い方がわかりませんでした。
[リンク先](https://qiita.com/t_okkan/items/4127a87177ed2b2db148)なども調べてみた結果、マルチプロセスではグローバル変数を用いても、プロセス間で値が共有されないことはわかりましたが、value,arrayをどのように使えばよいか理解できませんでした。
```python3
import concurrent.futures
import time
 
a = 0


def func1():
    print("[1]スタート")
    time.sleep(5)
    print("5秒経過")
    global a
    a = 1
           
def func2():
    print("[2]スタート")
    while True:
        if (a == 1):
            print("値の共有を確認")
            break
                

if __name__ == "__main__":

    with concurrent.futures.TreadPoolExecutor(max_workers=2) as executor:
        executor.submit(func1)
        executor.submit(func2)
```

上記のコードはマルチスレッドで実際に動かすことができたものです。
（細かい処理は省略して、変数の値を共有できたか確認する部分のみにしています）

func1とfunc2をマルチプロセスで動かし、値を共有するにはどのようにコードを書けばよいでしょうか。
よろしくお願いいたします。

[Python]マルチプロセスでメモリ共有を行いたい

C++でスレッド処理(並行処理)の限界は、
CPUのスレッド数(論理コア数)をこえることはできないもしくは超えると処理が著しく低下ということでしょうか？

タスクマネージャーにスレッド数2000という記載があり(前後差あり)
C++の並行処理はどこのスレッドを使っているのかわからず質問させていただきました
よろしくお願いします。

C++のスレッド処理はどこでおこなっているのか？

pythonのtwintを使ってtwitterをスクレイピングするプログラムを実行しています。
1000tweetずつスクレイピングして、jsonファイルとして出力させる関数を5つfor文で回す仕組みです。
1つの関数の処理が終わるのを待たずsleep(5)で次の関数を走らせたいので、今回threadingを採用してみました。
途中まではうまくいくんですが、最初の方でシャットダウンしてしまいます。
ご教示お願いいたします。

```python
import twint
from datetime import datetime
import shutil
import time
import threading

def scraping(n):

    def gettweet(n):
        nowdate = datetime.now()
        strdate = "{0:%Y-%m-%d %H:%M:%S}".format(nowdate)
        c = twint.Config()
        c.Limit = 1000
        c.Search = 'lang:ja'
        c.Store_json = True
        namedate = "{0:%Y%m%d%H%M%S}".format(nowdate)
        strfilename = 'result' + namedate + '_' + str(n) + '.json'
        c.Output = strfilename
        twint.run.Search(c)
        return strfilename

    while True:
        filename = gettweet(n)
        shutil.move(filename, 'temp/')

for i in range(1, 6):
    sc = threading.Thread(target=scraping, args=(i,))
    sc.start()
    time.sleep(5)

```
エラー内容
```ここに言語を入力
RuntimeError: cannot schedule new futures after shutdown
```

やりたいこと
scraping(1)→無限ループ
処理待たずにscraping(2)→無限ループ
処理待たずにscraping(3)→無限ループ
処理待たずにscraping(4)→無限ループ
処理待たずにscraping(5)→無限ループ

pythonでthreadingを使って無限ループの関数を並行処理させたいが途中でシャットダウンする

### 知りたいこと
今日マルチスレッドについて触れたばかりです。
全体的に並列処理についての理解がありません。


OpenMPを使用してマルチスレッド化を試みているのですが、

並列計算をする際にCPUに記載されたコア数/スレッド数を超えたスレッド数を設定できます。
それによって計算速度も上昇します。
これが一体何を意味するのか、ということが知りたいです。

cpuはi7 9700F,8コア8スレッドです。
[https://ark.intel.com/content/www/jp/ja/ark/products/193738/intel-core-i7-9700f-processor-12m-cache-up-to-4-70-ghz.html](https://ark.intel.com/content/www/jp/ja/ark/products/193738/intel-core-i7-9700f-processor-12m-cache-up-to-4-70-ghz.html)

### 実装内容
並列計算は以下のように書いていて
```C++
#include <omp.h> // OpenMP
~~~~省略~~~~
int N = 1000000;
#pragma omp parallel for num_threads(num) // numの値がスレッド数にあたる
	for(int i = 0; i < N; i++)process(); // 100万の処理
```
process()は
```C++
	int num = 0;
	for(int i = 0; i < 10000; ++i) // 内部で1万ループする
	{
		++num;
	}
```

process内1個の計算時間は0.02msくらい

### 計測結果

スレッド数とその速度を表にまとめた図が下です。
![イメージ説明](349908fb44cfc0d2730f12d5802edcd3.png)

1~8スレッドまではCPUのコアが利用できるのでわかります。
16~512までCPUコア数、スレッド数超えた状態で指定すると、指定スレッド数が倍々で上がるにつれ計算効率が最大3%くらい上昇しています。
なぜ上昇するのかもよくわかってないんですが、
一番知りたいのは、CPUに記載されてるスレッド数を越して設定して大丈夫なのか？という点です。
どなたかご教授いただけますと幸いです。