Handling Asynchronous Steps

A step runs to completion before the flow continues. That is deliberate: the order of the steps in a flow is the order in which things happen, which is what keeps the log and the results tree readable. Some work does not fit that shape, though. A process that must keep running while other steps proceed, a background thread that collects measurements, a connection that must stay open: none of these can be finished inside one step.

Orcraft's answer is to split such work into two steps. A start step launches the work and returns at once, and a later read step collects its outcome. Between the two the flow does whatever must overlap, on the same endpoint or elsewhere. The thing that ties the two steps together is the object the start step created, a thread or a process handle, which it puts on its result. Such an object has no meaning off the machine that made it, so it does not travel to the controller. It stays on the endpoint, and the flow receives a reference to it. Handing that reference to a later step on the same endpoint gives that step the live object back. The framework reference calls these non-transferable values.

The Remote Thread Example test of the Endpoints group shows the pattern with a background thread that keeps solving mazes on an endpoint while the flow waits.

The start, wait and read steps

Source directory:

The start step builds a small task record with a counter, a list of results, a lock and a stop event, starts a daemon thread that solves mazes until the event is set, and puts the record on its result:

def _run_step(self) -> OrcraftTestStepResult:
	self.result = StepStartMazeThreadResult()

	exe_folder = normalize_long_path(self.config.maze_executable_folder)
	width = int(self.config.width)
	height = int(self.config.height)

	task = {
		"count": 0,
		"results": [],
		"lock": threading.Lock(),
		"stop_event": threading.Event(),
	}

	def _worker() -> None:
		while not task["stop_event"].is_set():
			try:
				result = MazeApp(exe_folder, "solve", width=width, height=height).run_silent()
				if result.solved:
					with task["lock"]:
						task["count"] += 1
						task["results"].append(result)
			except Exception:
				pass

	thread = threading.Thread(target=_worker, daemon=True)
	task["thread"] = thread
	thread.start()

	Logger.log(
		f"Started background maze-generation thread (width={width}, height={height})",
		module_tag=False,
	)

	self.result.task = task
	self.result.status = OrcraftStatus.PASS
	return self.result

task is a plain attribute of the result, not a declared result field, because it is meant for the flow and not for a column in the controller. It holds a thread and a lock, so it cannot be copied off the endpoint; what reaches the flow is a reference that stands for it.

The read step receives the task through a plain attribute of its config, which arrives as the live object again because the step runs on the same endpoint. It stops the thread, joins it, and turns the collected results into declared result fields:

def _run_step(self) -> OrcraftTestStepResult:
	self.result = StepReadMazeThreadResult()

	task = self.config.task
	if task is None:
		self.result.status = OrcraftStatus.ERROR
		self.result.error_description = "Task object is missing"
		return self.result

	task["stop_event"].set()

	task["thread"].join(timeout=float(self.config.join_timeout_sec))
	if task["thread"].is_alive():
		self.result.status = OrcraftStatus.ERROR
		self.result.error_description = "Background maze thread did not stop in time"
		return self.result

	with task["lock"]:
		count = int(task["count"])
		results = list(task["results"])

	step_counts = [result.steps for result in results]
	avg_steps = round(sum(step_counts) / len(step_counts), 1) if step_counts else 0

	self.result.mazes_generated = count
	self.result.avg_path_steps = avg_steps
	self.result.stopped = True

Between the two sits a wait step that only sleeps for the configured number of seconds. In a real test this is where the work that has to overlap with the thread would go.

The flow

The flow runs the three steps on the test endpoint and passes the start step's task to the read step:

def _run_flow(self) -> None:
	# Step 1: Start background maze-solving thread on test endpoint
	step_1_result = self.run_step(
		StepStartMazeThread,
		endpoint=self.config.test_endpoint,
		maze_executable_folder=self.config.maze_executable_folder,
		width=self.config.width,
		height=self.config.height,
	)

	if self.handle_errors(step_1_result, self.result):
		return

	# Step 2: Wait - thread keeps running, task object is held in step_1_result
	step_2_result = self.run_step(
		StepWait,
		endpoint=self.config.test_endpoint,
		wait_sec=self.config.wait_sec,
	)

	if self.handle_errors(step_2_result, self.result):
		self.run_step(
			StepReadMazeThread,
			endpoint=self.config.test_endpoint,
			task=step_1_result.task,
		)
		return

	# Step 3: Stop thread and collect results - using task from Step 1
	step_3_result = self.run_step(
		StepReadMazeThread,
		endpoint=self.config.test_endpoint,
		task=step_1_result.task,
	)

	if self.handle_errors(step_3_result, self.result):
		return

	self.result.mazes_generated = step_3_result.mazes_generated
	self.result.avg_path_steps = step_3_result.avg_path_steps

Note what happens when the wait step fails: the flow still runs the read step before returning, so the thread is stopped rather than left running on the endpoint. Whatever a start step launches, some step must always end it, on the failure path as well as on the happy one.

Running the example

Select the Remote Thread Example test, set its Maze Executable Folder, choose the Linux machine as the Test Endpoint, and raise Wait Sec from 2 to 5 to give the thread more time:

Click Update Plan, then Run and confirm. When the run finishes, search the log for Background maze thread stopped:

The timings tell the story. Start Maze Thread finished in no time at all, because it only launched the thread. Wait took its five seconds, while the thread kept solving mazes on the endpoint. Read Maze Thread then stopped the thread and reported 20 mazes generated, and the log goes on to print each of them.

Click Details for the run:

The three steps ran on GMG-TEST-3, and the values the read step declared, Mazes Generated and Avg Path Steps, are carried up to the flow and the run as result fields. The thread itself never appears anywhere: it lived and died on the endpoint, and the flow only ever held a reference to it.

The same pattern serves a process started with Popen, an open socket, or any resource that must outlive one step. The reference is valid only on the endpoint that created it; passing it to a step on another machine, or to a local step, raises an error rather than silently copying something that cannot be copied.

The thread reference is a value the framework refuses to copy. The next page, Remote Type Fidelity, is about the values it does copy, and how they keep their Python types on the way to an endpoint and back.