pl-topologicalcopy in pipelines

[jennydaman]

Abstract

Several limitations to the current support for "feed join" operations, implemented by pl-topologicalcopy, make it impossible for "feed join" to be used in pipelines. To support "feed join" operations in pipelines, here a solution is proposed: (A) ts plugins should be able to wait on multiple parents, (B) ts-plugins should be supported in pipelines by using Piping.id for values of the parameter plugininstances.

Definitions

Feed join: creating a plugin instance in a feed where its inputs are the files of two or more plugin instances in the same feed.

Background

"Topology synthesis" or ts-type plugins were conceived to be a type of ChRIS plugin which interacts with its own feed in special ways: in contrast to ds-type plugins, the inputs of a ts-plugin include more than just the files of its "previous"/parent plugin instance. Currently, ts-plugins are implemented as a special case in the backend:

ChRIS_ultron_backEnd/chris_backend/plugininstances/services/manager.py

Lines 281 to 329 in 5d3c2e2

	def get_ts_plugin_instance_input_objs(self):
	"""
	Get a tuple whose first element is a dictionary with keys that are the ids of
	each input plugin instance to this 'ts' plugin instance. The values of
	this dictionary are also dictionaries containing the output dir of the plugin
	instances and the list of all the objects under the output dir that match a
	regular expression. The second element of the tuple indicates the value of the
	'groupByInstance' flag for this 'ts' plugin instance.
	"""
	job_id = self.str_job_id
	# extract the 'ts' plugin's special parameters from the DB
	plg_inst_ids = regexs = []
	group_by_instance = False
	if self.c_plugin_inst.plugin.meta.type == 'ts':
	for param_inst in self.l_plugin_inst_param_instances:
	if param_inst.plugin_param.name == 'plugininstances':
	# string param that represents a comma-separated list of ids
	plg_inst_ids = param_inst.value.split(',') if param_inst.value else []
	elif param_inst.plugin_param.name == 'filter':
	# string param that represents a comma-separated list of regular expr
	regexs = param_inst.value.split(',') if param_inst.value else []
	elif param_inst.plugin_param.name == 'groupByInstance':
	group_by_instance = param_inst.value
	d_objs = {}
	for i, inst_id in enumerate(plg_inst_ids):
	try:
	plg_inst = PluginInstance.objects.get(pk=int(inst_id))
	except PluginInstance.DoesNotExist:
	logger.error(f"[CODE05,{job_id}]: Couldn't find any plugin instance with "
	f"id {inst_id} while processing input instances to 'ts' "
	f"plugin instance with id {self.c_plugin_inst.id}")
	self.c_plugin_inst.error_code = 'CODE05'
	raise
	output_path = plg_inst.get_output_path()
	try:
	l_ls = self.swift_manager.ls(output_path)
	except ClientException as e:
	logger.error(f'[CODE06,{job_id}]: Error while listing swift '
	f'storage files in {output_path}, detail: {str(e)}')
	self.c_plugin_inst.error_code = 'CODE06'
	raise
	if (i < len(regexs)) and regexs[i]:
	r = re.compile(regexs[i])
	d_objs[plg_inst.id] = {'output_path': output_path,
	'objs': [obj for obj in l_ls if r.search(obj)]}
	else:
	d_objs[plg_inst.id] = {'output_path': output_path,
	'objs': l_ls}
	return d_objs, group_by_instance

ts plugins can be described more specifically by explaining the code: a ts plugin is a ds plugin which is assumed to have the parameters plugininstances, filter, and groupByInstance. Before a ts plugin runs, CUBE executes some special code to collect the files from every plugin instance specified by plugininstances. These files are used as the input to the ts-plugin instance.

The ts-plugin pl-topologicalcopy is used to perform "join" operations.

In the current implementation there is a problem where ts plugin instances have a data dependency on the plugin instances specified by their plugininstances parameter, howeThis topic was discussed during the ChRIS roundtable on 2023-01-10ver this data dependency is not satisfied because it is possible for a ts-plugin to start running before its dependencies are in the state finishedSuccessfully. Due to this issue, ts-plugins cannot be used in pipelines.

A second issue is that the value for the parameter plugininstances can only be figured out at runtime, which makes it impossible to specify its value in a pipeline.

Proposal

To support feed join operations in pipelines, ts plugins should be able to wait on multiple parents, and pipelines should be allowed to specify a value for the parameter plugininstances by list index.

Wait on multiple parents

Simple as that. ts-plugins are currently implemented as "special cases" in manager.py, so why not implement them as a special case in tasks.py as well?

ChRIS_ultron_backEnd/chris_backend/plugininstances/tasks.py

Lines 72 to 84 in 5d3c2e2

	@shared_task(bind=True)
	@skip_if_running
	def schedule_waiting_plugin_instances(self): # task is passed info about itself
	"""
	Schedule the apps corresponding to all plugin instances in 'waiting' DB status
	and whom previous plugin instance is in 'finishedSuccessfully' DB status.
	"""
	instances = PluginInstance.objects.filter(status='waiting',
	previous__status='finishedSuccessfully')
	for plg_inst in instances:
	plg_inst.status = 'scheduled'
	plg_inst.save()
	run_plugin_instance.delay(plg_inst.id) # call async task

Counterarguments: algorithmic complexity, poor scalability: we don't care about scalability right now. A reasonable ChRIS deployment might look like <20,000 plugin instances, CUBE running on a single machine with 16GB RAM.

plugininstances parameter in pipelines

1. The plugininstances parameter in a pipeline should accept a CSV of list indices, just as what is accepted by the previous_index argument.
2. When serializing a ts-type piping, the value for plugininstances should be mapped from list indices to Piping.id.
3. During workflow creation, plugin instances should be created breadth-first
4. During creation of a plugin instance of a ts-type piping, the value for plugininstances should be mapped from Piping.id to PluginInstance.id

Example of a ts pipeline

1. Canonical JSON representation of example pipeline

"plugin_tree": [
  {
    "title": "Convert NIFTI to MINC",
    "plugin_name": "pl-nii2mnc-u8",
    "plugin_version": "1.0.1",
    "previous_index": null
  },
  {
    "title": "Left WM mask",
    "plugin_name": "pl-nums2mask",
    "plugin_version": "1.0.1",
    "previous_index": 0,
    "plugin_parameter_defaults": [
      {
        "name": "value",
        "default": 160
      }
    ]
  },
  {
    "title": "Right WM mask",
    "plugin_name": "pl-nums2mask",
    "plugin_version": "1.0.1",
    "previous_index": 0,
    "plugin_parameter_defaults": [
      {
        "name": "value",
        "default": 161
      }
    ]
  },
  {
    "title": "Join Left and Right masks into same directory",
    "plugin_name": "pl-topologicalcopy",
    "plugin_version": "0.3",
    "previous_index": 2,
    "plugin_parameter_defaults": [
      {
        "name": "plugininstances",
        "default": "1,2"
      }
    ]
  },
  {
    "title": "Surface extraction",
    "plugin_name": "pl-fetal-cp-surface-extract",
    "plugin_version": "1.1.0",
    "previous_index": 3
  }
]

ASCII art diagram of example pipeline

    O      0: pl-nii2mnc-u8
   / \
  /   \
 |     O   1: pl-nums2mask --value 160
 O     |   2: pl-nums2mask --value 161
  \   /
   \ /
    O      3: pl-topologicalcopy --plugininstances 1,2
    |
    |
    O      4: pl-fetal-cp-surface-extract

1.1. As a RFC #2 YAML file

Also proposed is for piping title to be used in the value for plugininstances instead of list indices.

name: "Example pipeline containing ts piping"
authors: "Jennings Zhang <[email protected]>"
category: "Example"
description: "Input is a T1 nuc NIFTI"
locked: false
plugin_tree:

  # pipeline root node
- title: "Convert NIFTI to MINC"
  plugin: pl-nii2mnc-u8 v1.0.1
  previous: null

  # two sibling children of root
- title: "Left WM mask"
  plugin: pl-nums2mask v1.0.1
  previous: "Convert NIFTI to MINC"
  plugin_parameter_defaults:
    value: 160
- title: "Right WM mask"
  plugin: pl-nums2mask v1.0.1
  previous: "Convert NIFTI to MINC"
  plugin_parameter_defaults:
    value: 161

  # feed join operation on siblings
- title: "Join Left and Right masks into same directory"
  plugin: pl-topologicalcopy v0.3
  previous: "Right WM mask"
  plugin_parameter_defaults:
    plugininstances: 'Left WM Mask,Right WM Mask'

  # process joined files
- title: "Surface extraction"
  plugin: pl-fetal-cp-surface-extract v1.1.0
  previous: "Join Left and Right masks into same directory"

2. Mapping list indices -> Piping.id

For example, CUBE assigns the following piping IDs:

id	title
76	Convert NIFTI to MINC
77	Left WM mask
78	Right WM mask
79	Join Left and Right masks into same directory
80	Surface extraction

The plugin parameter defaults for piping 79 will be:

{
  "url": "https://example.com/api/v1/pipelines/string-parameter/2555/",
  "id": 2555,
  "value": "77,78",
  "type": "string",
  "plugin_piping_id": 79,
  "plugin_piping_title": "Join Left and Right masks into same directory",
  "previous_plugin_piping_id": 78,
  "param_name": "plugininstances",
  "param_id": 571,
  "plugin_piping": "https://example.com/api/v1/pipelines/pipings/79/",
  "plugin_name": "pl-topologicalcopy",
  "plugin_version": "0.3",
  "plugin_id": 102,
  "plugin_param": "https://example.com/api/v1/plugins/parameters/571/"
}

3-4 Breadth-first creation of plugin instances

First, user requests workflow to be created.
CUBE creates plugin instances from pipings breadth-first:

piping	plugininstance
76	111
77	112
78	113

Piping 79 is a ts-type piping, which is handled as a special case. Its parameter plugininstances is mapped from piping IDs to plugin instance IDs: the pipeline string parameter value 77,78 gets transformed to a plugin instance parameter value 112,113.

Final result:

piping	plugininstance
76	111
77	112
78	113
79	114
80	115

ASCII art diagram of created workflow

    O      111: pl-nii2mnc-u8
   / \
  /   \
 |     O   112: pl-nums2mask --value 160
 O     |   113: pl-nums2mask --value 161
  \   /
   \ /
    O      114: pl-topologicalcopy --plugininstances 112,113
    |
    |
    O      115: pl-fetal-cp-surface-extract

Summary of Value Mapping

title	list index	piping	plugin instance
Convert NIFTI to MINC	0	76	111
Left WM mask	1	77	112
Right WM mask	2	78	113
Join Left and Right masks into same directory	3	79	114
Surface extraction	4	80	115

Transformation of parameter value for plugininstances

1. As a YAML pipeline: "Rename left mask files,Rename right mask files"
2. As a JSON pipeline: "1,2"
3. As a pipeline in CUBE: "77,78"
4. As a workflow: "112,113"

Alternative Solutions

Dynamic plugins: see #482 tl;dr since a reasonableimplementation for dynamic plugins would require new features in CUBE anyways, better we just add this proposal instead.

Multiple parents: see #483 tl;dr all plugins may have [0, n) parents, a plugin with >1 parent will automatically do a feed join. This solution would be too much work.

Summary and Considerations:

• API is backwards-compatible
• upholds and extends the current idea of ts-type plugins
• new behavior where pl-topologicalcopy waits on its dependencies
• new behavior allowing pl-topologicalcopy to be used in pipelines
• CUBE performance and scalability worsened

Meeting Attendance

This topic was discussed during the ChRIS roundtable on 2023-01-10

In attendance: Rudolph, Jennings, Gideon, Sandip

Jorge was absent.

[jennydaman]

Decided

These four proposed changes shall be implemented, in order:

1. ts-type plugins shall wait on all plugin instances given by --plugininstances

Behavior will change, no change to API.

2. CUBE should accept RFC2 "YAML" pipelines format.

Users should be able to upload a YAML pipeline via POST api/v1/pipelines/ with Content-Type denoting that it's a RFC2 YAML file, e.g. application/vnd.chris.pipeline+yaml.

2.1 CUBE should save uploaded pipeline files to object storage and they should be retrievable.

Suggested implementation:an endpoint such as api/v1/pipelines/N/sourcefile/ should return the originally uploaded YAML file of a pipeline.

To register a pipeline from CUBE to another CUBE, it should be possible to POST api/v1/pipelines/ with a body {"pipeline_url": "https://another.cube.example.com/api/v1/pipelines/N/sourcefile/"}...

3. Allow ts-type plugins to be in pipelines, behavior as aforementioned.

API addition.

[jennydaman]

Cycle detection is not necessary for a correct implementation of multiple inheritance in ChRIS pipelines.

"Correct" meaning acyclic.

Concern: pipelines containing ts-plugins might contain cycles.

Layman’s explanation:

1. “Nodes” are added to a feed one at a time.
2. Existing nodes may not have any of their connections changed.

Think of a family tree. Is it possible for a person to be their own biological parent, grandparent, great-grandparent, great-great-grandparent, …? No! But what if it were possible for a child to have only a single parent, or 3 parents? Still, probably not...

Formal proof

Setup

1. ChRIS feeds and pipelines can be modeled as directed graphs, where plugin instances are vertices, and there exists [0, n) directed edges from each plugin instance to their previous plugin instance.
   • fs plugin instances have exactly 0 directed edges leaving
   • ds plugin instances have exactly 1 directed edge leaving
   • ts plugin instances have 1 or more directed edges leaving (though a pl-tsdircopy has functionally 0 directed edges leaving)
2. Pipeline instances or workflows, which are valid graphs of plugin instances, can be created from valid ChRIS pipelines where the pipeline instance is the same graph as the pipeline, therefore all statements true about graphs modeling plugin instances are also true about pipelines.
3. ChRIS feeds and pipelines can be thought of being created one vertex at a time.
   1. There is an implicit constraint that: to add a vertex v and edges E to graph G, all edges in E must be directed to existing vertices in graph G. That is to say, all of a plugin instance's previous plugin instances must exist (obviously).
   2. Let's say you were to try adding two vertices v1, v2 and all edges in E1, E2 to graph G simultaneously. To satisfy the implicit constraint, v1 and v2 would be invalid if any edges in E1 or E2 are directed to a vertex outside of G. This includes where v1 had an edge to v2 or vice-versa: in this case, v1 and v2 are invalid.
   3. It can be shown that adding multiple vertices simultaneously (where there are no edges between vertices to be added) is functionally equivalent to adding them one by one in arbitrary order.
   4. Comment: the Python code to check whether (v, E) is a valid addition to G, i.e. "given a list of plugin instance IDs, do they all exist?" is pretty easy to write.
4. Observation: removing edges or vertices from an acyclic graph will never create a cycle. I.e. deleting a plugin instance is not a problem.
5. In this model, the direction of edges are from added vertex to existing vertex (from plugin instance to its previous) but the direction can be reversed and the implications still hold.

Problem

Suppose G is a graph constructed by adding vertices to it one by one. A set of [0, n) directed edges E are added each time vertex v is added to G where every edge e in E is from v to an existing vertex in G. Prove that all graphs constructed this manner are acyclic.

Restating the Problem

The "worst case" would be where G has the maximum number of edges allowed. That is, for every vertex to be added v, E is a set of edges from v to every vertex in G (read: in a pipeline, every non-root piping is a pl-topologicalcopy to every other piping).

• Let N be the set of all natural numbers (0, 1, 2, 3, ..., infinity)
• Let S be the set of "worst case" graphs for all sizes n where n is in N

Every possible graph which is constructed as described in the above section is a subgraph of some graph in S.

Lemma: E is acyclic implies R is acyclic where R is a subgraph of E

Solution

1. base case: n=0, empty graph is acyclic
2. base case: n=1, a vertex with no edges is acyclic
3. base case: n=2, two vertices with one edge is acyclic (base cases n=1 and n=2 are unnecessary for the proof but are stated for clarity)
4. Inductive hypothesis: assume n=k holds
5. Show n=k+1 holds: to show G in S of size k+1 is acyclic:
   1. Since there is no edge to v, it is not part of a cycle
   2. Every existing vertex u in G directed to by an edge in the set of edges to be added E becomes neighbors of v, but because (1), the edge v -> u does not create cycles because there are no edges to v
   3. (Unnecessary statement) since no edges are created between existing vertices in G, we have checked all possible edges for the case n=k+1 and have shown that G is acyclic
6. It is shown that n = k + 1 holds => n = k is also proven
7. All members of S are acyclic. The "worst case" is shown to be acyclic for all possible graph sizes, therefore all graphs constructed in this manner are acyclic.

QED

The Importance of Breadth-First Search

The above proof can guarantee no cycles in ChRIS pipelines containing ts-type pipings.

Pipelines must be parsed in BFS order. Pipeline instances and workflows must be constructed in BFS order.

Let's look at a simple example:

    A
   / \
  B   C
   \ /
    D

• BFS order: A, B, C, D. Everything ok!
• DFS order: A, B, D, C. D cannot be created because C does not yet exist.

Parsing pipeines, creating pipeline instances/workflows in BFS order assures that all valid plugin instances/pipings which may be a "previous" to the next item to be added, exist.

[jennydaman]

Some more ways to think about it:

If a set of plugin instances can be scheduled, it is implied that they are connected and acyclic.

A circular data structure requires mutability to construct, however ChRIS pipelines are represented immutably.

[jennydaman]

@rudolphpienaar says:

I think it's straight up easier to just say, "To create a cycle in a graph you need to connect a descendant as an ancestor. However, at the time you create the node, the descendant doesn't exist, ergo there is no way to create a cycle"