☰

ptyバッファ容量の確認

tag-icon • tag-icon linux python pty
ptyバッファ容量の確認

私はシリアルポートをエミュレートするためにLinuxのttyサブシステムを使用しています。アナログシリアルポートは、物理シリアルポートが必要なアプリケーションで使用されます。私の場合、アプリケーションはDockerで実行されますが、Dockerはこの問題とは関係がないと思います。ただし、シミュレートされたシリアルポートを管理するアプリケーションがロックされている問題が発生しましたwrite。通話がブロックされないようにするか、いつブロックされるかを検出する方法があるかどうか疑問に思います。

シミュレートされたシリアルポートを管理し、それらの間でブロードキャストを実行するPythonアプリケーションがあります。つまり、あるアプリケーションがそのシリアルポートに書き込むと、ネットワーク上の他のすべてのシリアルポートが書き込み情報を受け取ります。アプリケーションは次のとおりです。

#!/usr/bin/env python3

import os
import pty
import selectors
import signal
import tty

# read the configuration
all_apps = ...


selector = selectors.DefaultSelector()


# create the emulated serial ports
master_fds = { }

for app in all_apps:
    app_master, app_slave = pty.openpty()

    tty.setraw(app_master)
    master_fds[app] = app_master
    selector.register(app_master, selectors.EVENT_READ, app_master)

    # make the symlink that gets picked up by the application
    os.symlink(os.ttyname(app_slave), "/run/my-app/{}/ttyS0".format(app))


# exit on SIGINT or SIGTERM
run_flag = True

def signal_handler(sig, stack):
    global run_flag
    run_flag = False

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)


# main loop
while run_flag:
    events = selector.select(timeout = 1)

    for key,mask in events:
        fd = key.data

        # read up to 1024 bytes of data from the app that sent the message
        data = os.read(fd, 1024)

        for app_fd in master_fds.values():
            # don't broadcast to the application that sent the message
            if fd == app_fd:
                continue

            # send the message to each application
            os.write(app_fd, data)


# cleanup
for key in master_fds.keys():
    os.remove("/run/my-app/{}/ttyS0".format(key))
    os.close(master_fds[key])

通常、このスクリプトはうまく機能します。アプリケーションが起動したら、ptyデバイスをシリアルポートに依存させ、アプリケーション間でメッセージをブロードキャストします。ただし、ある時点でスクリプトが中断される可能性があります。私はスクリプトが呼び出されていると思い、os.writeアプリケーションがptyマスターの1つに送信されたときにのみ呼び出されます。この状態では、スクリプトはシグナルに応答しません（ループの実行中にのみrun_flagをチェックしてos.writeループをブロックするため）。

私にとって理解できる唯一の説明は、少なくとも1つのアプリケーションがptyのスレーブ側でデータを正しく読み取ることができないことです。これが真であれば、カーネルの一部のptyサポートバッファがいっぱいになり、呼び出しがバッファをオーバーフローするwriteため、バッファが十分に使い果たされるまで呼び出しはブロックされます（これは決して発生しません）。

catポートのスレーブ側で実行して、エミュレートされたシリアルポートを消費することが時々可能であることがわかりました。ただし、os.write通話が通常ブロックされると、人の介入なしにブロックされないように十分に信頼できるアプリケーションが必要です。

ptyマスターへの呼び出しをブロックしないように、ptyサポートバッファの残りの容量を測定する方法（私のPythonスクリプトなど）はありますかwrite？

答え1

解決策を見つけたようですが、必ずしも理想的なものかどうかはわかりません。

ptyマスターを非ブロックモードに設定するようにスクリプトを編集できます。

import fcntl

...

for app in all_apps:
    app_master, app_slave = pty.openpty()

    flags = fcntl.fcntl(app_master, fcntl.F_GETFL)
    flags |= os.O_NONBLOCK
    fnctl.fcntl(app_master, fcntl.F_SETFL, flags)

その後、ホストに書き込むたびに書き込みが正常にブロックされたら、キャプチャしてBlockingIOError続行できます。

            # (try to) send the message to each application
            try:
                os.write(app_fd, data)
            except BlockingIOError:
                print("Caught BlockingIOError")

これはアプリケーションが誤動作してシリアルポートを読み取ることができないという基本的な問題を解決しませんが、このソリューションを使用すると、スクリプトは少なくともその操作を続けることができます。欠点は、一部のデータが失われる可能性があることです（ポートを消費しないアプリケーションにブロードキャスト）。

関連情報