src/llvm-project/lldb/test/API/functionalities/interactive_scripted_process/interactive_scripted_process.py - toolchain/rustc - Git at Google

 # Usage:
 # ./bin/lldb $LLVM/lldb/test/API/functionalities/interactive_scripted_process/main \
 #   -o "br set -p 'Break here'" \
 #   -o "command script import $LLVM/lldb/test/API/functionalities/interactive_scripted_process/interactive_scripted_process.py" \
 #   -o "create_mux" \
 #   -o "create_sub" \
 #   -o "br set -p 'also break here'" -o 'continue'

 import os, json, struct, signal, tempfile

 from threading import Thread
 from typing import Any, Dict

 import lldb
 from lldb.plugins.scripted_process import PassthroughScriptedProcess
 from lldb.plugins.scripted_process import PassthroughScriptedThread


 class MultiplexedScriptedProcess(PassthroughScriptedProcess):
     def __init__(self, exe_ctx: lldb.SBExecutionContext, args: lldb.SBStructuredData):
         super().__init__(exe_ctx, args)
         self.multiplexer = None
         if isinstance(self.driving_process, lldb.SBProcess) and self.driving_process:
             parity = args.GetValueForKey("parity")
             # TODO: Change to Walrus operator (:=) with oneline if assignment
             # Requires python 3.8
             val = parity.GetUnsignedIntegerValue()
             if val is not None:
                 self.parity = val

             # Turn PassthroughScriptedThread into MultiplexedScriptedThread
             for thread in self.threads.values():
                 thread.__class__ = MultiplexedScriptedThread

     def get_process_id(self) -> int:
         return self.parity + 420

     def launch(self, should_stop: bool = True) -> lldb.SBError:
         self.first_launch = True
         return lldb.SBError()

     def resume(self, should_stop: bool) -> lldb.SBError:
         if self.first_launch:
             self.first_launch = False
             return super().resume()
         else:
             if not self.multiplexer:
                 error = lldb.SBError("Multiplexer is not set.")
                 return error
             return self.multiplexer.resume(should_stop)

     def get_threads_info(self) -> Dict[int, Any]:
         if not self.multiplexer:
             return super().get_threads_info()
         filtered_threads = self.multiplexer.get_threads_info(pid=self.get_process_id())
         # Update the filtered thread class from PassthroughScriptedThread to MultiplexedScriptedThread
         return dict(
             map(
                 lambda pair: (pair[0], MultiplexedScriptedThread(pair[1])),
                 filtered_threads.items(),
             )
         )

     def create_breakpoint(self, addr, error, pid=None):
         if not self.multiplexer:
             error.SetErrorString("Multiplexer is not set.")
         return self.multiplexer.create_breakpoint(addr, error, self.get_process_id())

     def get_scripted_thread_plugin(self) -> str:
         return f"{MultiplexedScriptedThread.__module__}.{MultiplexedScriptedThread.__name__}"


 class MultiplexedScriptedThread(PassthroughScriptedThread):
     def get_name(self) -> str:
         parity = "Odd" if self.scripted_process.parity % 2 else "Even"
         return f"{parity}{MultiplexedScriptedThread.__name__}.thread-{self.idx}"


 class MultiplexerScriptedProcess(PassthroughScriptedProcess):
     listener = None
     multiplexed_processes = None

     def wait_for_driving_process_to_stop(self):
         def handle_process_state_event():
             # Update multiplexer process
             log("Updating interactive scripted process threads")
             dbg = self.driving_target.GetDebugger()
             new_driving_thread_ids = []
             for driving_thread in self.driving_process:
                 new_driving_thread_ids.append(driving_thread.id)
                 log(f"{len(self.threads)} New thread {hex(driving_thread.id)}")
                 structured_data = lldb.SBStructuredData()
                 structured_data.SetFromJSON(
                     json.dumps(
                         {
                             "driving_target_idx": dbg.GetIndexOfTarget(
                                 self.driving_target
                             ),
                             "thread_idx": driving_thread.GetIndexID(),
                         }
                     )
                 )

                 self.threads[driving_thread.id] = PassthroughScriptedThread(
                     self, structured_data
                 )

             for thread_id in self.threads:
                 if thread_id not in new_driving_thread_ids:
                     log(f"Removing old thread {hex(thread_id)}")
                     del self.threads[thread_id]

             print(f"New thread count: {len(self.threads)}")

             mux_process = self.target.GetProcess()
             mux_process.ForceScriptedState(lldb.eStateRunning)
             mux_process.ForceScriptedState(lldb.eStateStopped)

             for child_process in self.multiplexed_processes.values():
                 child_process.ForceScriptedState(lldb.eStateRunning)
                 child_process.ForceScriptedState(lldb.eStateStopped)

         event = lldb.SBEvent()
         while True:
             if not self.driving_process:
                 continue
             if self.listener.WaitForEvent(1, event):
                 event_mask = event.GetType()
                 if event_mask & lldb.SBProcess.eBroadcastBitStateChanged:
                     state = lldb.SBProcess.GetStateFromEvent(event)
                     log(f"Received public process state event: {state}")
                     if state == lldb.eStateStopped:
                         # If it's a stop event, iterate over the driving process
                         # thread, looking for a breakpoint stop reason, if internal
                         # continue.
                         handle_process_state_event()
             else:
                 continue

     def __init__(self, exe_ctx: lldb.SBExecutionContext, args: lldb.SBStructuredData):
         super().__init__(exe_ctx, args, launched_driving_process=False)
         if isinstance(self.driving_target, lldb.SBTarget) and self.driving_target:
             self.listener = lldb.SBListener(
                 "lldb.listener.multiplexer-scripted-process"
             )
             self.multiplexed_processes = {}

             # Copy breakpoints from real target to passthrough
             with tempfile.NamedTemporaryFile() as tf:
                 bkpt_file = lldb.SBFileSpec(tf.name)
                 error = self.driving_target.BreakpointsWriteToFile(bkpt_file)
                 if error.Fail():
                     log(
                         "Failed to save breakpoints from driving target (%s)"
                         % error.GetCString()
                     )
                 bkpts_list = lldb.SBBreakpointList(self.target)
                 error = self.target.BreakpointsCreateFromFile(bkpt_file, bkpts_list)
                 if error.Fail():
                     log(
                         "Failed create breakpoints from driving target \
                         (bkpt file: %s)"
                         % tf.name
                     )

             # Copy breakpoint from passthrough to real target
             if error.Success():
                 self.driving_target.DeleteAllBreakpoints()
                 for bkpt in self.target.breakpoints:
                     if bkpt.IsValid():
                         for bl in bkpt:
                             real_bpkt = self.driving_target.BreakpointCreateBySBAddress(
                                 bl.GetAddress()
                             )
                             if not real_bpkt.IsValid():
                                 log(
                                     "Failed to set breakpoint at address %s in \
                                     driving target"
                                     % hex(bl.GetLoadAddress())
                                 )

             self.listener_thread = Thread(
                 target=self.wait_for_driving_process_to_stop, daemon=True
             )
             self.listener_thread.start()

     def launch(self, should_stop: bool = True) -> lldb.SBError:
         if not self.driving_target:
             return lldb.SBError(
                 f"{self.__class__.__name__}.resume: Invalid driving target."
             )

         if self.driving_process:
             return lldb.SBError(
                 f"{self.__class__.__name__}.resume: Invalid driving process."
             )

         error = lldb.SBError()
         launch_info = lldb.SBLaunchInfo(None)
         launch_info.SetListener(self.listener)
         driving_process = self.driving_target.Launch(launch_info, error)

         if not driving_process or error.Fail():
             return error

         self.driving_process = driving_process

         for module in self.driving_target.modules:
             path = module.file.fullpath
             load_addr = module.GetObjectFileHeaderAddress().GetLoadAddress(
                 self.driving_target
             )
             self.loaded_images.append({"path": path, "load_addr": load_addr})

         self.first_resume = True
         return error

     def resume(self, should_stop: bool = True) -> lldb.SBError:
         if self.first_resume:
             # When we resume the multiplexer process for the first time,
             # we shouldn't do anything because lldb's execution machinery
             # will resume the driving process by itself.

             # Also, no need to update the multiplexer scripted process state
             # here because since it's listening for the real process stop events.
             # Once it receives the stop event from the driving process,
             # `wait_for_driving_process_to_stop` will update the multiplexer
             # state for us.

             self.first_resume = False
             return lldb.SBError()

         if not self.driving_process:
             return lldb.SBError(
                 f"{self.__class__.__name__}.resume: Invalid driving process."
             )

         return self.driving_process.Continue()

     def get_threads_info(self, pid: int = None) -> Dict[int, Any]:
         if not pid:
             return super().get_threads_info()
         parity = pid % 2
         return dict(filter(lambda pair: pair[0] % 2 == parity, self.threads.items()))

     def create_breakpoint(self, addr, error, pid=None):
         if not self.driving_target:
             error.SetErrorString("%s has no driving target." % self.__class__.__name__)
             return False

         def create_breakpoint_with_name(target, load_addr, name, error):
             addr = lldb.SBAddress(load_addr, target)
             if not addr.IsValid():
                 error.SetErrorString("Invalid breakpoint address %s" % hex(load_addr))
                 return False
             bkpt = target.BreakpointCreateBySBAddress(addr)
             if not bkpt.IsValid():
                 error.SetErrorString(
                     "Failed to create breakpoint at address %s"
                     % hex(addr.GetLoadAddress())
                 )
                 return False
             error = bkpt.AddNameWithErrorHandling(name)
             return error.Success()

         name = (
             "multiplexer_scripted_process"
             if not pid
             else f"multiplexed_scripted_process_{pid}"
         )

         if pid is not None:
             # This means that this method has been called from one of the
             # multiplexed scripted process. That also means that the multiplexer
             # target doesn't have this breakpoint created.
             mux_error = lldb.SBError()
             bkpt = create_breakpoint_with_name(self.target, addr, name, mux_error)
             if mux_error.Fail():
                 error.SetError(
                     "Failed to create breakpoint in multiplexer \
                                target: %s"
                     % mux_error.GetCString()
                 )
                 return False
         return create_breakpoint_with_name(self.driving_target, addr, name, error)


 def multiplex(mux_process, muxed_process):
     muxed_process.GetScriptedImplementation().multiplexer = (
         mux_process.GetScriptedImplementation()
     )
     mux_process.GetScriptedImplementation().multiplexed_processes[
         muxed_process.GetProcessID()
     ] = muxed_process


 def launch_scripted_process(target, class_name, dictionary):
     structured_data = lldb.SBStructuredData()
     structured_data.SetFromJSON(json.dumps(dictionary))

     launch_info = lldb.SBLaunchInfo(None)
     launch_info.SetProcessPluginName("ScriptedProcess")
     launch_info.SetScriptedProcessClassName(class_name)
     launch_info.SetScriptedProcessDictionary(structured_data)

     error = lldb.SBError()
     return target.Launch(launch_info, error)


 def duplicate_target(driving_target):
     error = lldb.SBError()
     exe = driving_target.executable.fullpath
     triple = driving_target.triple
     debugger = driving_target.GetDebugger()
     return debugger.CreateTargetWithFileAndTargetTriple(exe, triple)


 def create_mux_process(debugger, command, exe_ctx, result, dict):
     if not debugger.GetNumTargets() > 0:
         return result.SetError(
             "Interactive scripted processes requires one non scripted process."
         )

     debugger.SetAsync(True)

     driving_target = debugger.GetSelectedTarget()
     if not driving_target:
         return result.SetError("Driving target is invalid")

     # Create a seconde target for the multiplexer scripted process
     mux_target = duplicate_target(driving_target)
     if not mux_target:
         return result.SetError(
             "Couldn't duplicate driving target to launch multiplexer scripted process"
         )

     class_name = f"{__name__}.{MultiplexerScriptedProcess.__name__}"
     dictionary = {"driving_target_idx": debugger.GetIndexOfTarget(driving_target)}
     mux_process = launch_scripted_process(mux_target, class_name, dictionary)
     if not mux_process:
         return result.SetError("Couldn't launch multiplexer scripted process")


 def create_child_processes(debugger, command, exe_ctx, result, dict):
     if not debugger.GetNumTargets() >= 2:
         return result.SetError("Scripted Multiplexer process not setup")

     debugger.SetAsync(True)

     # Create a seconde target for the multiplexer scripted process
     mux_target = debugger.GetSelectedTarget()
     if not mux_target:
         return result.SetError("Couldn't get multiplexer scripted process target")
     mux_process = mux_target.GetProcess()
     if not mux_process:
         return result.SetError("Couldn't get multiplexer scripted process")

     driving_target = mux_process.GetScriptedImplementation().driving_target
     if not driving_target:
         return result.SetError("Driving target is invalid")

     # Create a target for the multiplexed even scripted process
     even_target = duplicate_target(driving_target)
     if not even_target:
         return result.SetError(
             "Couldn't duplicate driving target to launch multiplexed even scripted process"
         )

     class_name = f"{__name__}.{MultiplexedScriptedProcess.__name__}"
     dictionary = {"driving_target_idx": debugger.GetIndexOfTarget(mux_target)}
     dictionary["parity"] = 0
     even_process = launch_scripted_process(even_target, class_name, dictionary)
     if not even_process:
         return result.SetError("Couldn't launch multiplexed even scripted process")
     multiplex(mux_process, even_process)

     # Create a target for the multiplexed odd scripted process
     odd_target = duplicate_target(driving_target)
     if not odd_target:
         return result.SetError(
             "Couldn't duplicate driving target to launch multiplexed odd scripted process"
         )

     dictionary["parity"] = 1
     odd_process = launch_scripted_process(odd_target, class_name, dictionary)
     if not odd_process:
         return result.SetError("Couldn't launch multiplexed odd scripted process")
     multiplex(mux_process, odd_process)


 def log(message):
     # FIXME: For now, we discard the log message until we can pass it to an lldb
     # logging channel.
     should_log = False
     if should_log:
         print(message)


 def __lldb_init_module(dbg, dict):
     dbg.HandleCommand(
         "command script add -o -f interactive_scripted_process.create_mux_process create_mux"
     )
     dbg.HandleCommand(
         "command script add -o -f interactive_scripted_process.create_child_processes create_sub"
     )
	# Usage:
	# ./bin/lldb $LLVM/lldb/test/API/functionalities/interactive_scripted_process/main \
	# -o "br set -p 'Break here'" \
	# -o "command script import $LLVM/lldb/test/API/functionalities/interactive_scripted_process/interactive_scripted_process.py" \
	# -o "create_mux" \
	# -o "create_sub" \
	# -o "br set -p 'also break here'" -o 'continue'

	import os, json, struct, signal, tempfile

	from threading import Thread
	from typing import Any, Dict

	import lldb
	from lldb.plugins.scripted_process import PassthroughScriptedProcess
	from lldb.plugins.scripted_process import PassthroughScriptedThread


	class MultiplexedScriptedProcess(PassthroughScriptedProcess):
	def __init__(self, exe_ctx: lldb.SBExecutionContext, args: lldb.SBStructuredData):
	super().__init__(exe_ctx, args)
	self.multiplexer = None
	if isinstance(self.driving_process, lldb.SBProcess) and self.driving_process:
	parity = args.GetValueForKey("parity")
	# TODO: Change to Walrus operator (:=) with oneline if assignment
	# Requires python 3.8
	val = parity.GetUnsignedIntegerValue()
	if val is not None:
	self.parity = val

	# Turn PassthroughScriptedThread into MultiplexedScriptedThread
	for thread in self.threads.values():
	thread.__class__ = MultiplexedScriptedThread

	def get_process_id(self) -> int:
	return self.parity + 420

	def launch(self, should_stop: bool = True) -> lldb.SBError:
	self.first_launch = True
	return lldb.SBError()

	def resume(self, should_stop: bool) -> lldb.SBError:
	if self.first_launch:
	self.first_launch = False
	return super().resume()
	else:
	if not self.multiplexer:
	error = lldb.SBError("Multiplexer is not set.")
	return error
	return self.multiplexer.resume(should_stop)

	def get_threads_info(self) -> Dict[int, Any]:
	if not self.multiplexer:
	return super().get_threads_info()
	filtered_threads = self.multiplexer.get_threads_info(pid=self.get_process_id())
	# Update the filtered thread class from PassthroughScriptedThread to MultiplexedScriptedThread
	return dict(
	map(
	lambda pair: (pair[0], MultiplexedScriptedThread(pair[1])),
	filtered_threads.items(),
	)
	)

	def create_breakpoint(self, addr, error, pid=None):
	if not self.multiplexer:
	error.SetErrorString("Multiplexer is not set.")
	return self.multiplexer.create_breakpoint(addr, error, self.get_process_id())

	def get_scripted_thread_plugin(self) -> str:
	return f"{MultiplexedScriptedThread.__module__}.{MultiplexedScriptedThread.__name__}"


	class MultiplexedScriptedThread(PassthroughScriptedThread):
	def get_name(self) -> str:
	parity = "Odd" if self.scripted_process.parity % 2 else "Even"
	return f"{parity}{MultiplexedScriptedThread.__name__}.thread-{self.idx}"


	class MultiplexerScriptedProcess(PassthroughScriptedProcess):
	listener = None
	multiplexed_processes = None

	def wait_for_driving_process_to_stop(self):
	def handle_process_state_event():
	# Update multiplexer process
	log("Updating interactive scripted process threads")
	dbg = self.driving_target.GetDebugger()
	new_driving_thread_ids = []
	for driving_thread in self.driving_process:
	new_driving_thread_ids.append(driving_thread.id)
	log(f"{len(self.threads)} New thread {hex(driving_thread.id)}")
	structured_data = lldb.SBStructuredData()
	structured_data.SetFromJSON(
	json.dumps(
	{
	"driving_target_idx": dbg.GetIndexOfTarget(
	self.driving_target
	),
	"thread_idx": driving_thread.GetIndexID(),
	}
	)
	)

	self.threads[driving_thread.id] = PassthroughScriptedThread(
	self, structured_data
	)

	for thread_id in self.threads:
	if thread_id not in new_driving_thread_ids:
	log(f"Removing old thread {hex(thread_id)}")
	del self.threads[thread_id]

	print(f"New thread count: {len(self.threads)}")

	mux_process = self.target.GetProcess()
	mux_process.ForceScriptedState(lldb.eStateRunning)
	mux_process.ForceScriptedState(lldb.eStateStopped)

	for child_process in self.multiplexed_processes.values():
	child_process.ForceScriptedState(lldb.eStateRunning)
	child_process.ForceScriptedState(lldb.eStateStopped)

	event = lldb.SBEvent()
	while True:
	if not self.driving_process:
	continue
	if self.listener.WaitForEvent(1, event):
	event_mask = event.GetType()
	if event_mask & lldb.SBProcess.eBroadcastBitStateChanged:
	state = lldb.SBProcess.GetStateFromEvent(event)
	log(f"Received public process state event: {state}")
	if state == lldb.eStateStopped:
	# If it's a stop event, iterate over the driving process
	# thread, looking for a breakpoint stop reason, if internal
	# continue.
	handle_process_state_event()
	else:
	continue

	def __init__(self, exe_ctx: lldb.SBExecutionContext, args: lldb.SBStructuredData):
	super().__init__(exe_ctx, args, launched_driving_process=False)
	if isinstance(self.driving_target, lldb.SBTarget) and self.driving_target:
	self.listener = lldb.SBListener(
	"lldb.listener.multiplexer-scripted-process"
	)
	self.multiplexed_processes = {}

	# Copy breakpoints from real target to passthrough
	with tempfile.NamedTemporaryFile() as tf:
	bkpt_file = lldb.SBFileSpec(tf.name)
	error = self.driving_target.BreakpointsWriteToFile(bkpt_file)
	if error.Fail():
	log(
	"Failed to save breakpoints from driving target (%s)"
	% error.GetCString()
	)
	bkpts_list = lldb.SBBreakpointList(self.target)
	error = self.target.BreakpointsCreateFromFile(bkpt_file, bkpts_list)
	if error.Fail():
	log(
	"Failed create breakpoints from driving target \
	(bkpt file: %s)"
	% tf.name
	)

	# Copy breakpoint from passthrough to real target
	if error.Success():
	self.driving_target.DeleteAllBreakpoints()
	for bkpt in self.target.breakpoints:
	if bkpt.IsValid():
	for bl in bkpt:
	real_bpkt = self.driving_target.BreakpointCreateBySBAddress(
	bl.GetAddress()
	)
	if not real_bpkt.IsValid():
	log(
	"Failed to set breakpoint at address %s in \
	driving target"
	% hex(bl.GetLoadAddress())
	)

	self.listener_thread = Thread(
	target=self.wait_for_driving_process_to_stop, daemon=True
	)
	self.listener_thread.start()

	def launch(self, should_stop: bool = True) -> lldb.SBError:
	if not self.driving_target:
	return lldb.SBError(
	f"{self.__class__.__name__}.resume: Invalid driving target."
	)

	if self.driving_process:
	return lldb.SBError(
	f"{self.__class__.__name__}.resume: Invalid driving process."
	)

	error = lldb.SBError()
	launch_info = lldb.SBLaunchInfo(None)
	launch_info.SetListener(self.listener)
	driving_process = self.driving_target.Launch(launch_info, error)

	if not driving_process or error.Fail():
	return error

	self.driving_process = driving_process

	for module in self.driving_target.modules:
	path = module.file.fullpath
	load_addr = module.GetObjectFileHeaderAddress().GetLoadAddress(
	self.driving_target
	)
	self.loaded_images.append({"path": path, "load_addr": load_addr})

	self.first_resume = True
	return error

	def resume(self, should_stop: bool = True) -> lldb.SBError:
	if self.first_resume:
	# When we resume the multiplexer process for the first time,
	# we shouldn't do anything because lldb's execution machinery
	# will resume the driving process by itself.

	# Also, no need to update the multiplexer scripted process state
	# here because since it's listening for the real process stop events.
	# Once it receives the stop event from the driving process,
	# `wait_for_driving_process_to_stop` will update the multiplexer
	# state for us.

	self.first_resume = False
	return lldb.SBError()

	if not self.driving_process:
	return lldb.SBError(
	f"{self.__class__.__name__}.resume: Invalid driving process."
	)

	return self.driving_process.Continue()

	def get_threads_info(self, pid: int = None) -> Dict[int, Any]:
	if not pid:
	return super().get_threads_info()
	parity = pid % 2
	return dict(filter(lambda pair: pair[0] % 2 == parity, self.threads.items()))

	def create_breakpoint(self, addr, error, pid=None):
	if not self.driving_target:
	error.SetErrorString("%s has no driving target." % self.__class__.__name__)
	return False

	def create_breakpoint_with_name(target, load_addr, name, error):
	addr = lldb.SBAddress(load_addr, target)
	if not addr.IsValid():
	error.SetErrorString("Invalid breakpoint address %s" % hex(load_addr))
	return False
	bkpt = target.BreakpointCreateBySBAddress(addr)
	if not bkpt.IsValid():
	error.SetErrorString(
	"Failed to create breakpoint at address %s"
	% hex(addr.GetLoadAddress())
	)
	return False
	error = bkpt.AddNameWithErrorHandling(name)
	return error.Success()

	name = (
	"multiplexer_scripted_process"
	if not pid
	else f"multiplexed_scripted_process_{pid}"
	)

	if pid is not None:
	# This means that this method has been called from one of the
	# multiplexed scripted process. That also means that the multiplexer
	# target doesn't have this breakpoint created.
	mux_error = lldb.SBError()
	bkpt = create_breakpoint_with_name(self.target, addr, name, mux_error)
	if mux_error.Fail():
	error.SetError(
	"Failed to create breakpoint in multiplexer \
	target: %s"
	% mux_error.GetCString()
	)
	return False
	return create_breakpoint_with_name(self.driving_target, addr, name, error)


	def multiplex(mux_process, muxed_process):
	muxed_process.GetScriptedImplementation().multiplexer = (
	mux_process.GetScriptedImplementation()
	)
	mux_process.GetScriptedImplementation().multiplexed_processes[
	muxed_process.GetProcessID()
	] = muxed_process


	def launch_scripted_process(target, class_name, dictionary):
	structured_data = lldb.SBStructuredData()
	structured_data.SetFromJSON(json.dumps(dictionary))

	launch_info = lldb.SBLaunchInfo(None)
	launch_info.SetProcessPluginName("ScriptedProcess")
	launch_info.SetScriptedProcessClassName(class_name)
	launch_info.SetScriptedProcessDictionary(structured_data)

	error = lldb.SBError()
	return target.Launch(launch_info, error)


	def duplicate_target(driving_target):
	error = lldb.SBError()
	exe = driving_target.executable.fullpath
	triple = driving_target.triple
	debugger = driving_target.GetDebugger()
	return debugger.CreateTargetWithFileAndTargetTriple(exe, triple)


	def create_mux_process(debugger, command, exe_ctx, result, dict):
	if not debugger.GetNumTargets() > 0:
	return result.SetError(
	"Interactive scripted processes requires one non scripted process."
	)

	debugger.SetAsync(True)

	driving_target = debugger.GetSelectedTarget()
	if not driving_target:
	return result.SetError("Driving target is invalid")

	# Create a seconde target for the multiplexer scripted process
	mux_target = duplicate_target(driving_target)
	if not mux_target:
	return result.SetError(
	"Couldn't duplicate driving target to launch multiplexer scripted process"
	)

	class_name = f"{__name__}.{MultiplexerScriptedProcess.__name__}"
	dictionary = {"driving_target_idx": debugger.GetIndexOfTarget(driving_target)}
	mux_process = launch_scripted_process(mux_target, class_name, dictionary)
	if not mux_process:
	return result.SetError("Couldn't launch multiplexer scripted process")


	def create_child_processes(debugger, command, exe_ctx, result, dict):
	if not debugger.GetNumTargets() >= 2:
	return result.SetError("Scripted Multiplexer process not setup")

	debugger.SetAsync(True)

	# Create a seconde target for the multiplexer scripted process
	mux_target = debugger.GetSelectedTarget()
	if not mux_target:
	return result.SetError("Couldn't get multiplexer scripted process target")
	mux_process = mux_target.GetProcess()
	if not mux_process:
	return result.SetError("Couldn't get multiplexer scripted process")

	driving_target = mux_process.GetScriptedImplementation().driving_target
	if not driving_target:
	return result.SetError("Driving target is invalid")

	# Create a target for the multiplexed even scripted process
	even_target = duplicate_target(driving_target)
	if not even_target:
	return result.SetError(
	"Couldn't duplicate driving target to launch multiplexed even scripted process"
	)

	class_name = f"{__name__}.{MultiplexedScriptedProcess.__name__}"
	dictionary = {"driving_target_idx": debugger.GetIndexOfTarget(mux_target)}
	dictionary["parity"] = 0
	even_process = launch_scripted_process(even_target, class_name, dictionary)
	if not even_process:
	return result.SetError("Couldn't launch multiplexed even scripted process")
	multiplex(mux_process, even_process)

	# Create a target for the multiplexed odd scripted process
	odd_target = duplicate_target(driving_target)
	if not odd_target:
	return result.SetError(
	"Couldn't duplicate driving target to launch multiplexed odd scripted process"
	)

	dictionary["parity"] = 1
	odd_process = launch_scripted_process(odd_target, class_name, dictionary)
	if not odd_process:
	return result.SetError("Couldn't launch multiplexed odd scripted process")
	multiplex(mux_process, odd_process)


	def log(message):
	# FIXME: For now, we discard the log message until we can pass it to an lldb
	# logging channel.
	should_log = False
	if should_log:
	print(message)


	def __lldb_init_module(dbg, dict):
	dbg.HandleCommand(
	"command script add -o -f interactive_scripted_process.create_mux_process create_mux"
	)
	dbg.HandleCommand(
	"command script add -o -f interactive_scripted_process.create_child_processes create_sub"
	)