0

私が書いたudpサーバーのテストハーネスを書いています。アイデアは、 multiprocessing.pool を使用して、サーバーの複数のインスタンスに対して subprocess.call コマンドを実行することです(この質問による)。udp サーバーは、3 つのプロセス (メイン、http インターフェイス、および udp サーバー) で構成される別の python プログラムです。テスト スクリプトを作成しようとして、CTRL + C でテスト スクリプトを強制終了すると、アクティビティ モニターで常に 2 ～ 5 個の python2.7 プロセスがハングします。

私の試み：

  1. バニラサブプロセス
  2. 端末に接続されていない bash を処理するサブプロセス (この質問を参照)
  3. multiprocess.pool は、単純な test.py スクリプトを使用してハング プロセスも作成します (以下を参照)。

サーバーコードにサブプロセスキルを追加しようとしましたが、役に立ちませんでした。端末内でサーバーを実行すると、CTRL+C に正しく応答します。(1)と(2)はtest.pyスクリプトに問題がないので、サーバー内の子プロセスに何かあると思います。

1）サブプロセスを使用するだけ：

args0 = [sys.executable, server_path, '-p', '7000', '-t', '8000', '-u', '9000']
args1 = [sys.executable, server_path, '-p', '7001', '-t', '8001', '-u', '9001']

p0 = subprocess.Popen(args0)
p1 = subprocess.Popen(args1)

2) stdout、pipe、os.kill でサブプロセスを使用する:

p0 = subprocess.Popen(  args0, 
                        stdout=subprocess.PIPE, 
                        stderr=subprocess.STDOUT, 
                        close_fds=True)
print "running for 5s..."
time.sleep(5)
os.kill(p0.pid, signal.SIGTERM)

3a) test_harness.py (プールあり)

import sys, os, time
import subprocess, multiprocessing

def work(cmd):
    return subprocess.call(cmd) # , shell=False

def main():
    server_path = os.getcwd() + '/' + 'test.py'

    args0 = [sys.executable, server_path, '-p', '7000']
    # args1 = [sys.executable, server_path, '-p', '7001']
    tasks = [args0]

    pool = multiprocessing.Pool(processes=1)
    pool.map_async(work, tasks)
    time.sleep(3)
    pool.terminate()

if __name__ == '__main__':
    main()

3b) test.py

def main():
    while True:
        a = 1

if __name__ == '__main__':
    main()
4

1 に答える 1

0

やっと動作しました。このコードは、テスト ハーネス内のサブプロセスとして、2 つのマルチスレッド サーバー インスタンスを生成します。サブプロセスからすべてのコンソール出力を取得し、端末からテスト ハーネスを CTRL+C すると、すべてのサブプロセスも終了します。subprocess.Popen最終的にはうまくいきませんでした。

def runInstance(args):
    # NOTE: We don't do the stdout or stderr args
    p = subprocess.call(args, 
                        # stdout=subprocess.PIPE, 
                        # stderr=subprocess.STDOUT, 
                        close_fds=True, 
                        shell=True
                        )

def main():
    print "Starting test harness..."
    server_path = os.getcwd() + '/' + 'server.py'
    args0 = [sys.executable, server_path, '-p', '7000', '-t', '8000', '-u', '9000']
    args1 = [sys.executable, server_path, '-p', '7001', '-t', '8001', '-u', '9001']

    # Start server instances
    # NOTE: It is target=runInstance, not target=runInstance()
    p0 = multiprocessing.Process(target=runInstance, args=(' '.join(args0),))
    p1 = multiprocessing.Process(target=runInstance, args=(' '.join(args1),))
    p0.start()
    p1.start()
于 2013-06-27T00:12:00.423 に答える