Merge pull request #94 from xincunli-sonic/xincun/cherry-pick-3373

Cherry Pick PR: 3373

config/main.py

@@ -1,6 +1,7 @@
 #!/usr/sbin/env python
 
 import click
+import concurrent.futures
 import ipaddress
 import json
 import jsonpatch
@@ -14,6 +15,7 @@
 import itertools
 import copy
 import tempfile
+import threading
 
 from collections import OrderedDict
 from generic_config_updater.generic_updater import GenericUpdater, ConfigFormat, extract_scope
@@ -1083,6 +1085,11 @@ def validate_gre_type(ctx, _, value):
     except ValueError:
         raise click.UsageError("{} is not a valid GRE type".format(value))
 
+
+def apply_patch_wrapper(args):
+    return apply_patch_for_scope(*args)
+
+
 # Function to apply patch for a single ASIC.
 def apply_patch_for_scope(scope_changes, results, config_format, verbose, dry_run, ignore_non_yang_tables, ignore_path):
     scope, changes = scope_changes
@@ -1091,16 +1098,19 @@ def apply_patch_for_scope(scope_changes, results, config_format, verbose, dry_ru
         scope = multi_asic.DEFAULT_NAMESPACE
 
     scope_for_log = scope if scope else HOST_NAMESPACE
+    thread_id = threading.get_ident()
+    log.log_notice(f"apply_patch_for_scope started for {scope_for_log} by {changes} in thread:{thread_id}")
+
     try:
         # Call apply_patch with the ASIC-specific changes and predefined parameters
         GenericUpdater(scope=scope).apply_patch(jsonpatch.JsonPatch(changes),
-                                                    config_format,
-                                                    verbose,
-                                                    dry_run,
-                                                    ignore_non_yang_tables,
-                                                    ignore_path)
+                                                config_format,
+                                                verbose,
+                                                dry_run,
+                                                ignore_non_yang_tables,
+                                                ignore_path)
         results[scope_for_log] = {"success": True, "message": "Success"}
-        log.log_notice(f"'apply-patch' executed successfully for {scope_for_log} by {changes}")
+        log.log_notice(f"'apply-patch' executed successfully for {scope_for_log} by {changes} in thread:{thread_id}")
     except Exception as e:
         results[scope_for_log] = {"success": False, "message": str(e)}
         log.log_error(f"'apply-patch' executed failed for {scope_for_log} by {changes} due to {str(e)}")
@@ -1314,11 +1324,12 @@ def print_dry_run_message(dry_run):
                help='format of config of the patch is either ConfigDb(ABNF) or SonicYang',
                show_default=True)
 @click.option('-d', '--dry-run', is_flag=True, default=False, help='test out the command without affecting config state')
+@click.option('-p', '--parallel', is_flag=True, default=False, help='applying the change to all ASICs parallelly')
 @click.option('-n', '--ignore-non-yang-tables', is_flag=True, default=False, help='ignore validation for tables without YANG models', hidden=True)
 @click.option('-i', '--ignore-path', multiple=True, help='ignore validation for config specified by given path which is a JsonPointer', hidden=True)
 @click.option('-v', '--verbose', is_flag=True, default=False, help='print additional details of what the operation is doing')
 @click.pass_context
-def apply_patch(ctx, patch_file_path, format, dry_run, ignore_non_yang_tables, ignore_path, verbose):
+def apply_patch(ctx, patch_file_path, format, dry_run, parallel, ignore_non_yang_tables, ignore_path, verbose):
     """Apply given patch of updates to Config. A patch is a JsonPatch which follows rfc6902.
        This command can be used do partial updates to the config with minimum disruption to running processes.
        It allows addition as well as deletion of configs. The patch file represents a diff of ConfigDb(ABNF)
@@ -1364,8 +1375,26 @@ def apply_patch(ctx, patch_file_path, format, dry_run, ignore_non_yang_tables, i
                 changes_by_scope[asic] = []
 
         # Apply changes for each scope
-        for scope_changes in changes_by_scope.items():
-            apply_patch_for_scope(scope_changes, results, config_format, verbose, dry_run, ignore_non_yang_tables, ignore_path)
+        if parallel:
+            with concurrent.futures.ThreadPoolExecutor() as executor:
+                # Prepare the argument tuples
+                arguments = [(scope_changes, results, config_format,
+                              verbose, dry_run, ignore_non_yang_tables, ignore_path)
+                             for scope_changes in changes_by_scope.items()]
+
+                # Submit all tasks and wait for them to complete
+                futures = [executor.submit(apply_patch_wrapper, args) for args in arguments]
+
+                # Wait for all tasks to complete
+                concurrent.futures.wait(futures)
+        else:
+            for scope_changes in changes_by_scope.items():
+                apply_patch_for_scope(scope_changes,
+                                      results,
+                                      config_format,
+                                      verbose, dry_run,
+                                      ignore_non_yang_tables,
+                                      ignore_path)
 
         # Check if any updates failed
         failures = [scope for scope, result in results.items() if not result['success']]