grpc/src/php/bin/xds_manager.py

#!/usr/bin/env python
# Copyright 2021 gRPC authors.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Manage PHP child processes for the main PHP xDS Interop client"""

import argparse
import fcntl
import os
import subprocess

# This script is being launched from src/php/bin/run_xds_client.sh
# to manage PHP child processes which will send 1 RPC each
# asynchronously. This script keeps track of all those open
# processes and reports back to the main PHP interop client each
# of the child RPCs' status code.

if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--tmp_file1', nargs='?', default='')
    parser.add_argument('--tmp_file2', nargs='?', default='')
    parser.add_argument('--bootstrap_path', nargs='?', default='')
    args = parser.parse_args()
    server_address = ''
    rpcs_started = []
    open_processes = {}
    client_env = dict(os.environ)
    client_env['GRPC_XDS_BOOTSTRAP'] = args.bootstrap_path
    while True:
        # tmp_file1 contains a list of RPCs (and their spec) the parent process
        # wants executed
        f1 = open(args.tmp_file1, 'r+')
        fcntl.flock(f1, fcntl.LOCK_EX)
        while True:
            key = f1.readline()
            if not key:
                break
            key = key.strip()
            if key.startswith('server_address'):
                if not server_address:
                    server_address = key[15:]
            elif not key in rpcs_started:
                # format here needs to be in sync with
                # src/php/tests/interop/xds_client.php
                items = key.split('|')
                num = items[0]
                metadata = items[2]
                timeout_sec = items[3]
                if items[1] == 'UnaryCall':
                    p = subprocess.Popen([
                        'php', '-d', 'extension=grpc.so', '-d',
                        'extension=pthreads.so',
                        'src/php/tests/interop/xds_unary_call.php',
                        '--server=' + server_address, '--num=' + str(num),
                        '--metadata=' + metadata, '--timeout_sec=' + timeout_sec
                    ],
                                         env=client_env)
                elif items[1] == 'EmptyCall':
                    p = subprocess.Popen([
                        'php', '-d', 'extension=grpc.so', '-d',
                        'extension=pthreads.so',
                        'src/php/tests/interop/xds_empty_call.php',
                        '--server=' + server_address, '--num=' + str(num),
                        '--metadata=' + metadata, '--timeout=' + timeout_sec
                    ],
                                         env=client_env)
                else:
                    continue
                rpcs_started.append(key)
                open_processes[key] = p
        f1.truncate(0)
        fcntl.flock(f1, fcntl.LOCK_UN)
        f1.close()
        # tmp_file2 contains the RPC result of each key received from tmp_file1
        f2 = open(args.tmp_file2, 'a')
        fcntl.flock(f2, fcntl.LOCK_EX)
        keys_to_delete = []
        for key, process in open_processes.items():
            result = process.poll()
            if result is not None:
                # format here needs to be in sync with
                # src/php/tests/interop/xds_client.php
                f2.write(key + ',' + str(process.returncode) + "\n")
                keys_to_delete.append(key)
        for key in keys_to_delete:
            del open_processes[key]
        fcntl.flock(f2, fcntl.LOCK_UN)
        f2.close()