Python3 ping結果をCSVに出力したい

Question

目的

設置した複数のRaspberry Piがネットに繋がっているかをCSVファイルから確認したい。（直接手動でpingコマンドをいちいちうつのは面倒であるため、ファイルから確認できるようにしたい。）
ただし、pingを送信するのは別のRaspberry Piとする。

案

CSVファイルに予めRaspberry PiのIPアドレスを書いておく。
そのIPアドレスにpingを打って、応答があればCSVテーブルに現在日時を書く。
こうすれば、日時が最近のものであればネットに繋がっている、日時が古いものであればネットに繋がっていない、すべての日時が古いものであればping送信しようとしているラズパイがネットに繋がってないと解釈できると考えた。

ソースコードの説明

pingの送信間隔はあとから設定できるように変数に入れています。
ここでは、現在日時が00秒になったら、CSVにあるIPアドレスにpingを送信します（1分間隔）。
pingはラズパイごと順番にやっているとつっかえたときに時間の遅れが出そうなので、それぞれのスレッドで送信しようとしています。多分そうすればすべてのラズパイに対して同じ時間にping送信できる（？）。
ping応答があれば、そのIPアドレス行の更新日時列に現在時間のデータを書きます。なければ何もしません。

作成したソースコード

from datetime import datetime
import configparser
import csv
import re
import subprocess
import threading
import time

cnt = 0
exe_time = ["**:**:00"]


def read_csv():
    path = "/home/pi/デスクトップ/table.csv"
    data = []
    with open(path, "r", newline="", errors="ignore") as rf:
        f = csv.reader(rf, lineterminator="\r\n")
        for row in f:
            data.append(row)
    return data


def write_csv(row, data):
    path = "/home/pi/デスクトップ/table.csv"
    with open(path, "w", newline="", encoding="utf_8_sig") as wf:
        writer = csv.writer(wf, lineterminator="\r\n")
        now_str = datetime.now().strftime("%Y/%m/%d %H:%M:%S.%f")[:-3]
        for i in data:
            # ping成功時のデータは更新する
            if i[1] == row[1]:
                writer.writerow([row[0], row[1], row[2], now_str])
            # ping失敗時のデータはそのまま
            else:
                writer.writerow([i[0], i[1], i[2], i[3]])
    
    
# pingコマンド実行タイミング
# exe_time = ["**:**:00"] で 1分間に1回
def timing():
    global cnt, exe_time
    for i in range(len(exe_time)):
        exe_time[i] = exe_time[i].replace("*", r"\d")
        if re.match(exe_time[i], format(datetime.now(), "%H:%M:%S")) != None:
            if cnt == 0:
                cnt = 1
                return True
            return False
    # フラグ処理
    if cnt == 1:
            cnt = 0


def ping_exe(row, data):
    cmd = "ping -c 3 " + row[1]
    ret = subprocess.call(cmd.split(" "))

    # 試験的にpingの成功失敗デバイスを指定
    
    if row[1] == "192.168.0.20" or row[1] == "192.168.0.30":
        ret = 0
    else:
        ret = 1
    

    print("ret: {}".format(ret))
    if ret == 0:
        write_csv(row, data)
            

if __name__ == "__main__":
    try:
        while True:
            if timing() == True:
                data = read_csv()
                for row in data:
                    threading.Thread(target=lambda:ping_exe(row, data)).start()
            time.sleep(0.1)
      
    except KeyboardInterrupt:
        print("Ctrl + C")
        
    except Exception as e:
        print(type(e), e)



ホスト名,IPアドレス,場所,更新日時
user01,192.168.0.10,玄関,
user02,192.168.0.20,廊下,
user03,192.168.0.30,自室,

症状

ping応答したラズパイが複数ある場合に更新日時が反映されないものがある。
（たとえばこの例ではuser02とuser03が成功しているとして、反映されているのが片方だけであったりする。）

スレッドを使って1つのファイルに対して複数のアクセスがあるからこうなってしまうのでしょうか？

また、上の方法では時間データでネット接続を判断していますが他にいい方法があれば教えて下さい。

Comments

[meg_]

排他制御が必要かと思います。

[person]

pingを打つのは同時として、CSV書き込みをする際は書き込み前にスレッドをロックして書き込んだらリリースするようなイメージでしょうか？

[dodox86]

Pythonはマルチスレッドで書いてもGIL(Global Interpreter Lock)で実質、シングルスレッドのような動き、効果になることがありますが、ファイルの出力の競合は置いておいて、現状、望みの処理になっていますか？　見た目で問題ないならそれでも良いと思いますが。しかし依然としてマルチスレッドに伴うファイルなどのリソースの排他の意識は必要です。お使いのラズパイOSのバージョンやPython3のバージョンの詳細を質問文中に追記しましょう。

[person]

ファイルの書き込み以外（1分間隔での処理実行、pingの送信）については私からは問題ないように見えます。
ただし、pingの送信には送受信が成功した場合と失敗した場合
の処理を見極めたいため、ソースコード上ではヘッダと192.168.0.10は失敗、192.168.0.20と192.168.0.30は成功とあとから結果を上書きしています。

OSバージョン
$ lsb_release -a
No LSB modules are available.
Distributor ID: Raspbian
Description: Raspbian GNU/Linux 9.13 (stretch)
Release: 9.13
Codename: stretch

Pythonのバージョン
$ python --version
Python 2.7.13
$ python3 --version
Python 3.5.3

Answer 1

スレッドを使って1つのファイルに対して複数のアクセスがあるからこうなってしまうのでしょうか？

その通りです。

複数のスレッドから同時アクセスする場合、「排他制御」が必要になりますが、
無駄な待ち時間が発生したり、コードが複雑化する傾向があるため、

データの書き込みなどは、そのファイルを扱うスレッドをひとつにして、
(スレッドセーフな)同期キューを通じてデータのやり取りを行うことで、
同時アクセスを発生させないようにます。

また、データの更新に関しては、
CSV ファイルの更新は、更新の際すべてのデータを再書き込みする必要があるため、
何らかの障害があり強制終了した場合にデータ破損のリスクが出てきます。

CSVファイルは追記以外のデータ変更には向かないため、

データベース等 (大規模なDBではなく組み込み用途のもの。
標準モジュールであれば sqlite3 等) を用いて、
CSVファイルが必要なら、別途データベースから生成するようにすると良いです。

---

• mainスレッド: 1分毎のスケジューリング
• A: pingを送るスレッド (※ 複数の場合 ThreadPoolExecutor が便利です)
• B: データを書き込み(更新)するスレッド

A -> B へ同期キュー経由で更新するデータを渡す。

但し、この方法で同期が保証されるのは、同じプロセス内のみです。
別プロセスからの同時アクセスは起こる可能性があります。
この点の簡易的な解決策は、データベースを導入することです。

---

解決策(まとめ):
「書き込みタイミングの同期」と「CSVファイルの更新」の２点について、

• データ更新用のスレッドの導入
• データベースの導入

データベースの利用を提案しますが、質問はCSVファイルとの事なので、
もし、CSVファイルで運用したい（もしくは必要がある）のであれば、
スレッド間・プロセス間の排他制御を、必要に応じて自分で行う必要があります。
threading.Lock や fcntl.lockf 辺りを調べてみてください。

Comments

[person]

回答ありがとうございます。

データベースを利用するとして、
(1) CSVを読み込んでデータベースにINSERT
(2) ping結果によってデータベースの列をUPDATE（したり、しなかったり）
(3) データベースのデータをCSVに出力
となると思います。
すべてのラズパイ（user01〜03の計3台）のステータスが決まったかどうかは取得することは可能ですか？
おそらくステータスが確定したかどうかでCSV出力の処理に移ると思うので。

[teamikl]

ステータス所得の部分は出来ていて、
複数のスレッドでの処理が終了したかどうかを知る、ということですよね？

ThreadPoolExecutor を使うなら、concurrent.futures.as_completed
で一連のスレッドでの実行の完了を待つことが出来ます。

もしくは、(concurrentモジュールを使わない場合)
UPDATEしない場合もキューに何らかのデータを送り
単純にキューの読み出し側でカウントする等でしょうか。
（pingの応答が長すぎると不整合が起きるので、この方法の場合はタイムアウト設定が必要)

[teamikl]

＞(1) CSVを読み込んでデータベースにINSERT
＞(2) ping結果によってデータベースの列をUPDATE（したり、しなかったり）
＞(3) データベースのデータをCSVに出力

この仕様なら、CSVへの同時書き込みは発生しなくなるので
データベースは使わなくても、リストに結果を格納して
ソートしてCSVへ出力でも良いですね。

入力ファイルと出力ファイルが同じ且つ
全文書き換えになるので、障害時のファイル破損には対策が必要ですが。
書き込みの同期の問題は解消されます。

データベースを導入する場合、
読み込み元をデータベースにしないと、
後者のファイル破損問題は解消できません。

[person]

> 入力ファイルと出力ファイルが同じ且つ
> 全文書き換えになるので、障害時のファイル破損には対策が必要ですが。
> 書き込みの同期の問題は解消されます。

障害時とはラズパイとファイルサーバ間の通信障害のことでしょうか。
書き込みしてる最中に通信が切れてファイルが破損するイメージであってますか？

> データベースを導入する場合、
> 読み込み元をデータベースにしないと、
> 後者のファイル破損問題は解消できません。

ping送信先のIPアドレスのテーブルをデータベースで作成しておいて、結果のみをCSV出力するということでしょうか？

[teamikl]

＞障害時とはラズパイとファイルサーバ間の通信障害のことでしょうか。
書き込みしてる最中に通信が切れてファイルが破損するイメージであってますか？

ファイルサーバが何処に出てくるのかわかりませんが、ping の通信ではなく、
書き込みプロセスが正常に完了する前に中断されてしまった場合です。

OS自体の障害も含みますが、
プログラムを強制終了した場合等でも起こりえます。＞ファイル破損

＞ping送信先のIPアドレスのテーブルをデータベースで作成しておいて、結果のみをCSV出力するということでしょうか？

はい。
読み込み元がCSVファイルのままでは、
データベース導入の利点がなくなてしまうので。

[person]

IPアドレステーブルはCSVで用意できたほうがいいので、ThreadPoolExecutorを使おうかなと思います。

ただ、下のように作ったサンプルをwhile内の1分間隔処理のif文の中に書くと、「pingの応答が長すぎると不整合」と同じようなことになりそうと思ったのですが大丈夫でしょうか。
次の処理実行をスキップするようなイメージになるのでしょうか。
（1分間隔の処理の場合、次の00秒が来ても、スレッドが終わってないのでwhile内でas_completedで止まってしまい、処理実行されない？）

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed

def func(n, str):
____for i in range(100):
________print(i + n)
____return str

executor = ThreadPoolExecutor(3)
threads = []
thread1 = executor.submit(lambda: func(0, "thread1"))
thread2 = executor.submit(lambda: func(100, "thread2"))
thread3 = executor.submit(lambda: func(500, "thread3"))
threads.append(thread1)
threads.append(thread2)
threads.append(thread3)

for thread in as_completed(threads):
____print("end: " + thread.result())

print("all end")

[teamikl]

> 1分間隔処理のif文の中に書く

as_completed は、各スレッドでの実行完了を待つため、同スレッド内であれば
「後から開始した処理が先に終了する」ような不整合は起きませんが、
１分間隔のスケジューリングを阻害してしまう可能性があります。

- as_complated の引数に timeoutを設定
- pingにも timeoutを設定

＞IPアドレステーブルはCSVで用意できたほうがいいので

であれば、読み込むCSVと、出力するファイルは別にしたほうが良いです。
前の更新日時を何処に記録するか・・・という課題は出てきますが。

mainスレッド内でCSV出力する場合、
open()直後のタイミングで Ctrl-C が押されると、
ファイル破損に繋がります。

タイミング次第なので、実際のプログラムで狙って再現することは難しいですが、
意図的に再現するなら、適当な中身のあるファイルを作成して、
pythonで 書き込みモードでopenだけして終了してみてください。
ファイルサイズが0バイトになっているはずです。

[person]

タイムアウト設定することにします。

Ctrl C については、ラズパイを直接操作することはないので、今のところはこれについては考えないことにします。
（ラズパイを起動したら、Pythonを自動起動して後はずっと放置のため）