PHP: allow xDS interop client to start RPCs asynchronously (grpc#25696)

* PHP: allow xDS interop client to start RPCs asynchronously

* Address review comments

* Remove adhoc test config

src/php/bin/run_xds_client.sh

@@ -0,0 +1,63 @@
+#!/bin/bash
+# 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.
+
+
+# This script is being launched from the run_xds_tests.py runner and is
+# responsible for managing child PHP processes.
+
+cleanup () {
+    echo "Trapped SIGTERM. Cleaning up..."
+    set -x
+    kill -9 $PID1
+    kill -9 $PID2
+    running=false
+    set +x
+}
+
+trap cleanup SIGTERM
+
+set -e
+cd $(dirname $0)/../../..
+root=$(pwd)
+
+# tmp_file1 contains the list of RPCs (and their spec) the parent PHP
+# process want executed
+tmp_file1=$(mktemp)
+# tmp_file2 contains the RPC result of each key initiated
+tmp_file2=$(mktemp)
+
+set -x
+# This is the PHP parent process, who is primarily responding to the
+# run_xds_tests.py runner's stats requests
+php -d extension=grpc.so -d extension=pthreads.so \
+    src/php/tests/interop/xds_client.php $1 $2 $3 $4 $5 $6 \
+    --tmp_file1=$tmp_file1 --tmp_file2=$tmp_file2 &
+PID1=$!
+
+# This script watches RPCs written to tmp_file1, spawn off more PHP
+# child processes to execute them, and writes the result to tmp_file2
+python3 -u src/php/bin/xds_manager.py \
+        --tmp_file1=$tmp_file1 --tmp_file2=$tmp_file2 \
+        --bootstrap_path=$GRPC_XDS_BOOTSTRAP &
+PID2=$!
+set +x
+
+# This will be killed by a SIGTERM signal from the run_xds_tests.py
+# runner
+running=true
+while $running
+do
+    sleep 1
+done

src/php/bin/xds_manager.py

@@ -0,0 +1,100 @@
+#!/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]
+                rpc_behavior = 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),
+                        '--rpc_behavior=' + rpc_behavior,
+                        '--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),
+                        '--rpc_behavior=' + rpc_behavior,
+                        '--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()

src/php/lib/Grpc/RpcServer.php

@@ -51,9 +51,7 @@ private function loadRequest($request) {
         $event = $this->call->startBatch([
             OP_RECV_MESSAGE => true,
         ]);
-        if (!$event->message) {
-            throw new Exception("Did not receive a proper message");
-        }
+        $request->clear();
         $request->mergeFromString($event->message);
         return $request;
     }

src/php/tests/interop/Grpc/Testing/XdsUpdateClientConfigureServiceStub.php

@@ -0,0 +1,6 @@
+<?php
+// DO NOT EDIT
+namespace Grpc\Testing;
+class XdsUpdateClientConfigureServiceStub {
+}
+