python3でsocket.recvを強制終了したい

Question

説明

Python3でマルチキャスト受信するプログラムを作成しているのですが、socket.recvをしている途中で通信を終了したい場合があります。
socketにタイムアウトを設定すれば受信タイムアウト発生後に通信終了を判定することができるのですが、受信タイムアウト発生前に割り込んで受信を終了させることはできないでしょうか。
例えば以下のコードですと、理想は待機後の8秒で終了することですが、実際はタイムアウトが2回発生した後の10秒後に終了してしまいます。
どうにかして8秒で強制終了させられないでしょうか。よろしくお願いします。

コード

global is_recv


def recv_func(addr, port, timeout):
    global is_recv

    is_recv = True

    with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_UDP) as sock:
        sock.settimeout(timeout)

        # マルチキャストJOIN
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.bind('', port)
        sock.setsockopt(socket.IPPROTO_IP,
                        socket.IP_ADD_MEMBERSHIP,
                        socket.inet_aton(addr) + socket.inet_aton('0.0.0.0'))

        # 受信
        while is_recv:
            try:
                data, addr = sock.recvfrom(4096)
                print('received from={}, len={}'.format(addr, len(data)))
            except Exception as ex:
                print(ex)
            finally:
                time.sleep(0.1)

        # マルチキャストLeave(略)
        pass


if __name__ == '__main__':
    global is_recv

    # 開始
    print('start: time={}'.format(datetime.now()))
    t = Thread(target=recv_func, args=('239.0.0.1', 5000, 5,))
    t.start()

    # 待機
    time.sleep(8)

    # 終了
    is_recv = False
    t.join()
    print('end: time={}'.format(datetime.now()))

追記

sleepの8秒は決められた値ではなく変動するため、終了時間を予想してtimeoutを設定する、ということはできません。
timeoutを0.1秒など小さな値にしてtry~exceptを使用すれば、is_recv=False直後にスレッド終了が可能ですが、Exceptionを大量発生させるのもスマートではないため、他の方法を模索しています。

Comments

[_Victorique__]

18秒の間違いでしょうか？

[y.nakamura]

いえ、動作は以下のようになっており、10秒の受信タイムアウト(2)の後にプログラム終了しているところを、08秒の待機sleep終了直後にプログラム終了したいと思っています。
08秒のスレッド終了指示でis_recvをFalseにするのではなく、何らかのsocket操作で強制終了できないでしょうか？

・00秒
　・プログラム開始
　・スレッド開始
　・受信開始(1)
　・待機sleep開始
・05秒
　・受信タイムアウト発生(1)
　・受信開始(2)
・08秒
　・待機sleep終了
　・スレッド終了指示
・10秒
　・受信タイムアウト発生(2)
　・スレッド終了
　・プログラム終了

[Kenji.Noguchi]

ソケットのタイムアウトを5秒から8秒に変更して、Timeoutが発生したらis_recv=Falseすればいいのでは？現状ソケットのタイムアウトが5秒だからメインスレッドが8秒スリープしてたら2回目のsock.recvがブロッキングするので5秒x2で10秒になるのは当然ですね。

[y.nakamura]

5秒×2で10秒になるのは承知しております。
またsleepの8秒は決められた値ではなく変動するため、終了時間を予想してtimeoutを設定する、ということはできません。
timeoutを0.1秒など小さな値にしてtry~exceptを使用すれば、is_recv=False後にすぐスレッド終了が可能ですが、Exceptionを大量発生させるのもスマートではないため、他の方法を模索しています。

[Kenji.Noguchi]

なるほど。ちょっと考えてみます。

Answer 1

ご質問の内容に興味が沸きましたので試してみました。

受信タイムアウト発生前に割り込んで受信を終了させることはできないでしょうか。

一見、__main__の方のメインスレッドで socket.close()を呼び出せばrecv_funcスレッドのsocket.recv_fromでExceptionで抜けるのではないかと思って試してみたですが、その方法ではダメでした。recv_funcスレッド内のrecv_fromから抜けてきません。

私自身、C言語での生なソケットプログラミングで別のスレッドから強制的にソケットをcloseして処理を抜けさせるような(少々乱暴な)方法を使うことがあり、それでできるかと思ったのですが、PythonではNGのようです。

Python-3.5.7 のソースコードを読むと、ソケットのシステムコールを呼び出すC言語の関数 sock_call_exで、以下のようなコメントがありました。

// Python-3.5.7/Modules/socketmodule.c 

/* Call a socket function.
...省略

   sock_call_ex() must be called with the GIL held. The socket function is
   called with the GIL released. */

sock_call_ex関数内でのシステムコール呼び出し部分では、以下のように呼び出し前後でガードされています。

Py_BEGIN_ALLOW_THREADS
res = sock_func(s, data);
Py_END_ALLOW_THREADS

PythonのGIL(Global Interpreter Lock)で守られているようで、他スレッドで実行中のソケットのシステムコール(質問者さんの例ではrecv_from)を、他スレッドからキャンセルすることは無理そうです。

代替案として、元のコードとほとんど同じですがselect.selectを使って一定時間ごとにフラグをチェックする方法をご紹介します。質問者さんは「受信タイムアウトの例外を利用してフラグをチェックする方法がスマートではない」かんじがするとのことでしたが、selectを使えば意味的にも少しスマートです。また、ソケットに受信タイムアウトをセットする必要もなくなります。

# t1b.py - select
import socket
import time
import datetime
import threading
import select
import sys

global is_recv

def recv_func(addr, port, timeout):
    global is_recv

    is_recv = True

    with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_UDP) as sock:
#        sock.settimeout(timeout)

        # マルチキャストJOIN
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.setsockopt(socket.IPPROTO_IP,
                               socket.IP_ADD_MEMBERSHIP,
                               socket.inet_aton(addr) + socket.inet_aton('0.0.0.0'))

        # 本コードではブロックしてもしなくても同じ
        #sock.setblocking(0)
        rfds = [sock]

        # 受信
        while is_recv:
            try:
                # 0.5秒ごとにselectタイムアウトで is_recvフラグチェックの機会を与える
                r, _, _ = select.select(rfds, [], [], 0.5)
                for rs in r:
                    data, addr = sock.recvfrom(4096)
                    print('received from={}, len={}'.format(addr, len(data)))
            except Exception as ex:
                # print(ex)
                print('Exception: time={}, desc={}'.format(datetime.datetime.now(), ex))
            finally:
                time.sleep(0.1)

        # マルチキャストLeave(略)
        # pass
        print("recv_func leaves.")


if __name__ == '__main__':
    global is_recv

    # 開始
    print('main start: time={}'.format(datetime.datetime.now()))
    t = threading.Thread(target=recv_func, args=('239.0.0.1', 5000, 5,))
    t.start()

    # 待機
    time.sleep(8)
    is_recv = False

    # スレッド終了待ち
    t.join()

    # 終了
    print('main end: time={}'.format(datetime.datetime.now()))

Ubuntu16.04(x64) / Python3.5.2 での実行結果です。ほぼ8秒できれいに終わります。

@ubuntu1604-x64:/mnt/share$ sudo python3 t1b.py
main start: time=2019-04-03 15:30:53.343439
received from=('192.168.11.100', 0), len=36
received from=('192.168.11.100', 0), len=36
received from=('192.168.11.100', 0), len=36
received from=('192.168.11.100', 0), len=36
recv_func leaves.
main end: time=2019-04-03 15:31:01.818659
user01@ubuntu1604-x64:/mnt/share$ 

もうひとつ、今度はスレッドを使わない方法に直してみたものをご案内します。シグナルを使ったものです。recv_func関数は__main__で実行してしまい、sys.alarmで8秒後にシグナル(SIGALRM)を発生させ、非同期で動くシグナルハンドラーの中でソケットをcloseします。同時にis_recvフラグをOFFにし、抜けるようにするものです。

なお、with構文の終わりで自動でソケットをcloseする流れになるかと思いますが、テスト用のコードなので流用する際は適当に直してください。

# t1d.py
import socket
import time
import datetime
import threading
import sys
import signal

global is_recv
global g_sock

def signal_handler(signal, frame):
    global is_recv
    global g_sock
    print("SIGALRM!\n")
    g_sock.close()
    is_recv = False


def recv_func(addr, port, timeout):
    global is_recv
    global g_sock

    is_recv = True

    with socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_UDP) as sock:
#        sock.settimeout(timeout)
        g_sock = sock

        # マルチキャストJOIN
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        sock.setsockopt(socket.IPPROTO_IP,
                               socket.IP_ADD_MEMBERSHIP,
                               socket.inet_aton(addr) + socket.inet_aton('0.0.0.0'))

        # 受信
        while is_recv:
            try:
                data, addr = sock.recvfrom(4096)
                print('received from={}, len={}'.format(addr, len(data)))
            except Exception as ex:
                print(ex)
            finally:
                time.sleep(0.1)

        # マルチキャストLeave(略)
        pass
        print('recv_func leaves.')


if __name__ == '__main__':
    global is_recv

    signal.signal(signal.SIGALRM, signal_handler)

    # 開始
    print('main-start: time={}'.format(datetime.datetime.now()))
    # スレッドは使わない
    #t = threading.Thread(target=recv_func, args=('239.0.0.1', 5000, 5,))
    #t.start()

    # 8秒後にシグナルで通知
    signal.alarm(8)
    recv_func('239.0.0.1', 5000, 5)
    #time.sleep(8)

    # 終了
    print('main-end: time={}'.format(datetime.datetime.now()))

同、実行結果です。これもきれいにほぼ8秒で終わります。

user01@ubuntu1604-x64:/mnt/share$ sudo python3 t1d.py
main-start: time=2019-04-03 15:29:46.341175
received from=('192.168.11.100', 0), len=44
received from=('192.168.11.100', 0), len=44
SIGALRM!

[Errno 9] Bad file descriptor
recv_func leaves.
main-end: time=2019-04-03 15:29:54.445209
user01@ubuntu1604-x64:/mnt/share$ 

Comments

[y.nakamura]

recv中の他操作はコード上で明確にロックされているのですね。
selectモジュールを使用することで、期待通りの動作を実現することができました。
シグナルを使用したものも、別のシーンで利用することができそうです。
大変勉強になりました。ありがとうございます。

Answer 2

終了 でsys.exit()でもしてやれば強制的に終わりませんか。

Comments

[dodox86]

別回答の流れで試させていただいたのですが、興味深いことにメイン(__main__)のスレッドは終わるもののrecv_funcスレッドの方は終わらず、動いたままでした。(Ubuntu 16.04 64ビット / Python 3.5.2)

[y.nakamura]

ご回答ありがとうございます。
スレッド作成の際にdaemon=Trueを指定することで、メインスレッドと共にrecv_funcスレッドを強制終了することができました。

t = Thread(target=receive_func, args=('239.0.0.1', 5000, 5,), daemon=True)

その場合マルチキャストのLeaveをスキップしてしまいますが、強制終了可能ということで一案とさせていただきます。

Answer 3

コメントに書きましたが、ソケットのタイムアウトを8秒にして、タイムアウトが発生したらループから抜ければOKでしょう。

        while is_recv:
            try:
                data, addr = sock.recvfrom(4096)
                print('received from={}, len={}'.format(addr, len(data)))
            except socket.timeout:
                is_recv = False
            except Exception as ex:
                print(ex)
            finally:
                time.sleep(0.1)

（中略）

    t = Thread(target=recv_func, args=('239.0.0.1', 5000, 8,))

結果

start: time=2019-04-02 10:59:18.920342
end: time=2019-04-02 10:59:26.925506

Comments

[y.nakamura]

ご回答ありがとうございます。
条件等を質問内容に追記させていただきました。