[grpc] Number of processes through arguments.

Author: arturgontijo

run_example_service.py

@@ -30,6 +30,12 @@ def main():
                         action="store_true",
                         dest="run_ssl",
                         help="start the daemon with SSL")
+    parser.add_argument("--mp",
+                        help="number of concurrent processes",
+                        metavar="NUMBER_OF_PROCESSES",
+                        default=1,
+                        type=int,
+                        required=False)
     args = parser.parse_args()
     root_path = pathlib.Path(__file__).absolute().parent
 
@@ -41,7 +47,8 @@ def main():
                                service_modules,
                                args.run_daemon,
                                args.daemon_config,
-                               args.run_ssl)
+                               args.run_ssl,
+                               args.mp)
 
     # Continuous checking all subprocess
     try:
@@ -56,7 +63,8 @@ def main():
         raise
 
 
-def start_all_services(cwd, service_modules, run_daemon, daemon_config, run_ssl):
+def start_all_services(cwd, service_modules,
+                       run_daemon, daemon_config, run_ssl, mp):
     """
     Loop through all service_modules and start them.
     For each one, an instance of Daemon "snetd" is created.
@@ -65,14 +73,14 @@ def start_all_services(cwd, service_modules, run_daemon, daemon_config, run_ssl)
     all_p = []
     for _, service_module in enumerate(service_modules):
         service_name = service_module.split(".")[-1]
-        log.info("Launching {} on port {}".format(str(registry[service_name]),
-                                                  service_module))
+        log.info("Launching {} on port {}".format(service_module,
+                                                  str(registry[service_name])))
         all_p += start_service(cwd, service_module,
-                               run_daemon, daemon_config, run_ssl)
+                               run_daemon, daemon_config, run_ssl, mp)
     return all_p
 
 
-def start_service(cwd, service_module, run_daemon, daemon_config, run_ssl):
+def start_service(cwd, service_module, run_daemon, daemon_config, run_ssl, mp):
     """
     Starts SNET Daemon ("snetd") and the python module of the service
     at the passed gRPC port.
@@ -98,11 +106,11 @@ def add_ssl_configs(conf):
                 all_p.append(start_snetd(str(cwd), config_file))
     service_name = service_module.split(".")[-1]
     grpc_port = registry[service_name]["grpc"]
-    p = subprocess.Popen([
-        sys.executable,
-        "-m", service_module,
-        "--grpc-port", str(grpc_port)],
-       cwd=str(cwd))
+    p = subprocess.Popen([sys.executable,
+                          "-m", service_module,
+                          "--grpc-port", str(grpc_port),
+                          "--mp", str(mp)],
+                         cwd=str(cwd))
     all_p.append(p)
     return all_p
 

service/example_service.py

@@ -8,7 +8,7 @@
 import sys
 
 import argparse
-import os.path
+import os
 import grpc
 
 from service import registry
@@ -56,11 +56,12 @@
 # derived from the protobuf codes.
 class CalculatorServicer(grpc_bt_grpc.CalculatorServicer):
     def __init__(self):
+        self.pid = os.getpid()
         self.a = 0
         self.b = 0
         self.result = 0
         # Just for debugging purpose.
-        _LOGGER.debug("CalculatorServicer created")
+        _LOGGER.debug("[{}] CalculatorServicer created".format(self.pid))
 
     # The method that will be exposed to the snet-cli call command.
     # request: incoming data
@@ -74,7 +75,10 @@ def add(self, request, context):
         self.result = Result()
 
         self.result.value = self.a + self.b
-        _LOGGER.debug("add({},{})={}".format(self.a, self.b, self.result.value))
+        _LOGGER.debug("[{}] add({},{})={}".format(self.pid,
+                                                  self.a,
+                                                  self.b,
+                                                  self.result.value))
         return self.result
 
     def sub(self, request, context):
@@ -83,7 +87,10 @@ def sub(self, request, context):
 
         self.result = Result()
         self.result.value = self.a - self.b
-        _LOGGER.debug("sub({},{})={}".format(self.a, self.b, self.result.value))
+        _LOGGER.debug("[{}] sub({},{})={}".format(self.pid,
+                                                  self.a,
+                                                  self.b,
+                                                  self.result.value))
         return self.result
 
     def mul(self, request, context):
@@ -92,7 +99,10 @@ def mul(self, request, context):
 
         self.result = Result()
         self.result.value = self.a * self.b
-        _LOGGER.debug("mul({},{})={}".format(self.a, self.b, self.result.value))
+        _LOGGER.debug("[{}] mul({},{})={}".format(self.pid,
+                                                  self.a,
+                                                  self.b,
+                                                  self.result.value))
         return self.result
 
     def div(self, request, context):
@@ -101,7 +111,10 @@ def div(self, request, context):
 
         self.result = Result()
         self.result.value = self.a / self.b
-        _LOGGER.debug("div({},{})={}".format(self.a, self.b, self.result.value))
+        _LOGGER.debug("[{}] div({},{})={}".format(self.pid,
+                                                  self.a,
+                                                  self.b,
+                                                  self.result.value))
         return self.result
 
 
@@ -154,12 +167,19 @@ def main():
                         default=registry[service_name]['grpc'],
                         type=int,
                         required=False)
-    args = parser.parse_args(sys.argv[1:])
+    parser.add_argument("--mp",
+                        help="number of concurrent processes",
+                        metavar="NUMBER_OF_PROCESSES",
+                        default=1,
+                        type=int,
+                        required=False)
+    args = parser.parse_args()
+
+    num_processes = _PROCESS_COUNT if args.mp > _PROCESS_COUNT else args.mp
     with reserve_port(args.grpc_port) as port:
-        _LOGGER.debug("Binding to port '%s'", port)
         sys.stdout.flush()
         workers = []
-        for _ in range(_PROCESS_COUNT):
+        for _ in range(num_processes):
             # NOTE: It is imperative that the worker subprocesses be forked before
             # any gRPC servers start up. See
             # https://github.com/grpc/grpc/issues/16001 for more details.