/
common.py
640 lines (567 loc) · 25.3 KB
/
common.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
# This code is part of Qiskit.
#
# (C) Copyright IBM 2022.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.
"""Common preset passmanager generators."""
import collections
from typing import Optional
from qiskit.circuit.equivalence_library import SessionEquivalenceLibrary as sel
from qiskit.circuit.controlflow import CONTROL_FLOW_OP_NAMES
from qiskit.passmanager.flow_controllers import ConditionalController
from qiskit.transpiler.passmanager import PassManager
from qiskit.transpiler.passes import Error
from qiskit.transpiler.passes import BasisTranslator
from qiskit.transpiler.passes import Unroll3qOrMore
from qiskit.transpiler.passes import Collect2qBlocks
from qiskit.transpiler.passes import Collect1qRuns
from qiskit.transpiler.passes import ConsolidateBlocks
from qiskit.transpiler.passes import UnitarySynthesis
from qiskit.transpiler.passes import HighLevelSynthesis
from qiskit.transpiler.passes import CheckMap
from qiskit.transpiler.passes import GateDirection
from qiskit.transpiler.passes import BarrierBeforeFinalMeasurements
from qiskit.transpiler.passes import CheckGateDirection
from qiskit.transpiler.passes import TimeUnitConversion
from qiskit.transpiler.passes import ALAPScheduleAnalysis
from qiskit.transpiler.passes import ASAPScheduleAnalysis
from qiskit.transpiler.passes import FullAncillaAllocation
from qiskit.transpiler.passes import EnlargeWithAncilla
from qiskit.transpiler.passes import ApplyLayout
from qiskit.transpiler.passes import RemoveResetInZeroState
from qiskit.transpiler.passes import FilterOpNodes
from qiskit.transpiler.passes import ValidatePulseGates
from qiskit.transpiler.passes import PadDelay
from qiskit.transpiler.passes import InstructionDurationCheck
from qiskit.transpiler.passes import ConstrainedReschedule
from qiskit.transpiler.passes import PulseGates
from qiskit.transpiler.passes import ContainsInstruction
from qiskit.transpiler.passes import VF2PostLayout
from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason
from qiskit.transpiler.passes.layout.vf2_post_layout import VF2PostLayoutStopReason
from qiskit.transpiler.exceptions import TranspilerError
from qiskit.transpiler.layout import Layout
_ControlFlowState = collections.namedtuple("_ControlFlowState", ("working", "not_working"))
# Any method neither known good nor known bad (i.e. not a Terra-internal pass) is passed through
# without error, since it is being supplied by a plugin and we don't have any knowledge of these.
_CONTROL_FLOW_STATES = {
"layout_method": _ControlFlowState(working={"trivial", "dense", "sabre"}, not_working=set()),
"routing_method": _ControlFlowState(
working={"none", "stochastic", "sabre"}, not_working={"lookahead", "basic"}
),
"translation_method": _ControlFlowState(
working={"translator", "synthesis"},
not_working=set(),
),
"optimization_method": _ControlFlowState(working=set(), not_working=set()),
"scheduling_method": _ControlFlowState(working=set(), not_working={"alap", "asap"}),
}
def _has_control_flow(property_set):
return any(property_set[f"contains_{x}"] for x in CONTROL_FLOW_OP_NAMES)
def _without_control_flow(property_set):
return not any(property_set[f"contains_{x}"] for x in CONTROL_FLOW_OP_NAMES)
class _InvalidControlFlowForBackend:
# Explicitly stateful closure to allow pickling.
def __init__(self, basis_gates=(), target=None):
if target is not None:
self.unsupported = [op for op in CONTROL_FLOW_OP_NAMES if op not in target]
elif basis_gates is not None:
basis_gates = set(basis_gates)
self.unsupported = [op for op in CONTROL_FLOW_OP_NAMES if op not in basis_gates]
else:
# Pass manager without basis gates or target; assume everything's valid.
self.unsupported = []
def message(self, property_set):
"""Create an error message for the given property set."""
fails = [x for x in self.unsupported if property_set[f"contains_{x}"]]
if len(fails) == 1:
return f"The control-flow construct '{fails[0]}' is not supported by the backend."
return (
f"The control-flow constructs [{', '.join(repr(op) for op in fails)}]"
" are not supported by the backend."
)
def condition(self, property_set):
"""Checkable condition for the given property set."""
return any(property_set[f"contains_{x}"] for x in self.unsupported)
def generate_control_flow_options_check(
layout_method=None,
routing_method=None,
translation_method=None,
optimization_method=None,
scheduling_method=None,
basis_gates=(),
target=None,
):
"""Generate a pass manager that, when run on a DAG that contains control flow, fails with an
error message explaining the invalid options, and what could be used instead.
Returns:
PassManager: a pass manager that populates the ``contains_x`` properties for each of the
control-flow operations, and raises an error if any of the given options do not support
control flow, but a circuit with control flow is given.
"""
bad_options = []
message = "Some options cannot be used with control flow."
for stage, given in [
("layout", layout_method),
("routing", routing_method),
("translation", translation_method),
("optimization", optimization_method),
("scheduling", scheduling_method),
]:
option = stage + "_method"
method_states = _CONTROL_FLOW_STATES[option]
if given is not None and given in method_states.not_working:
if method_states.working:
message += (
f" Got {option}='{given}', but valid values are {list(method_states.working)}."
)
else:
message += (
f" Got {option}='{given}', but the entire {stage} stage is not supported."
)
bad_options.append(option)
out = PassManager()
out.append(ContainsInstruction(CONTROL_FLOW_OP_NAMES, recurse=False))
if bad_options:
out.append(ConditionalController(Error(message), condition=_has_control_flow))
backend_control = _InvalidControlFlowForBackend(basis_gates, target)
out.append(
ConditionalController(Error(backend_control.message), condition=backend_control.condition)
)
return out
def generate_error_on_control_flow(message):
"""Get a pass manager that always raises an error if control flow is present in a given
circuit."""
out = PassManager()
out.append(ContainsInstruction(CONTROL_FLOW_OP_NAMES, recurse=False))
out.append(ConditionalController(Error(message), condition=_has_control_flow))
return out
def if_has_control_flow_else(if_present, if_absent):
"""Generate a pass manager that will run the passes in ``if_present`` if the given circuit
has control-flow operations in it, and those in ``if_absent`` if it doesn't."""
if isinstance(if_present, PassManager):
if_present = if_present.to_flow_controller()
if isinstance(if_absent, PassManager):
if_absent = if_absent.to_flow_controller()
out = PassManager()
out.append(ContainsInstruction(CONTROL_FLOW_OP_NAMES, recurse=False))
out.append(ConditionalController(if_present, condition=_has_control_flow))
out.append(ConditionalController(if_absent, condition=_without_control_flow))
return out
def generate_unroll_3q(
target,
basis_gates=None,
approximation_degree=None,
unitary_synthesis_method="default",
unitary_synthesis_plugin_config=None,
hls_config=None,
):
"""Generate an unroll >3q :class:`~qiskit.transpiler.PassManager`
Args:
target (Target): the :class:`~.Target` object representing the backend
basis_gates (list): A list of str gate names that represent the basis
gates on the backend target
approximation_degree (Optional[float]): The heuristic approximation degree to
use. Can be between 0 and 1.
unitary_synthesis_method (str): The unitary synthesis method to use. You can see
a list of installed plugins with :func:`.unitary_synthesis_plugin_names`.
unitary_synthesis_plugin_config (dict): The optional dictionary plugin
configuration, this is plugin specific refer to the specified plugin's
documentation for how to use.
hls_config (HLSConfig): An optional configuration class to use for
:class:`~qiskit.transpiler.passes.HighLevelSynthesis` pass.
Specifies how to synthesize various high-level objects.
Returns:
PassManager: The unroll 3q or more pass manager
"""
unroll_3q = PassManager()
unroll_3q.append(
UnitarySynthesis(
basis_gates,
approximation_degree=approximation_degree,
method=unitary_synthesis_method,
min_qubits=3,
plugin_config=unitary_synthesis_plugin_config,
target=target,
)
)
unroll_3q.append(
HighLevelSynthesis(
hls_config=hls_config,
coupling_map=None,
target=target,
use_qubit_indices=False,
equivalence_library=sel,
basis_gates=basis_gates,
min_qubits=3,
)
)
# If there are no target instructions revert to using unroll3qormore so
# routing works.
if basis_gates is None and target is None:
unroll_3q.append(Unroll3qOrMore(target, basis_gates))
else:
unroll_3q.append(BasisTranslator(sel, basis_gates, target=target, min_qubits=3))
return unroll_3q
def generate_embed_passmanager(coupling_map):
"""Generate a layout embedding :class:`~qiskit.transpiler.PassManager`
This is used to generate a :class:`~qiskit.transpiler.PassManager` object
that can be used to expand and apply an initial layout to a circuit
Args:
coupling_map (Union[CouplingMap, Target): The coupling map for the backend to embed
the circuit to.
Returns:
PassManager: The embedding passmanager that assumes the layout property
set has been set in earlier stages
"""
return PassManager([FullAncillaAllocation(coupling_map), EnlargeWithAncilla(), ApplyLayout()])
def _layout_not_perfect(property_set):
"""Return ``True`` if the first attempt at layout has been checked and found to be imperfect.
In this case, perfection means "does not require any swap routing"."""
return property_set["is_swap_mapped"] is not None and not property_set["is_swap_mapped"]
def _apply_post_layout_condition(property_set):
# if VF2 Post layout found a solution we need to re-apply the better
# layout. Otherwise we can skip apply layout.
return (
property_set["VF2PostLayout_stop_reason"] is not None
and property_set["VF2PostLayout_stop_reason"] is VF2PostLayoutStopReason.SOLUTION_FOUND
)
def generate_routing_passmanager(
routing_pass,
target,
coupling_map=None,
vf2_call_limit=None,
backend_properties=None,
seed_transpiler=None,
check_trivial=False,
use_barrier_before_measurement=True,
vf2_max_trials=None,
):
"""Generate a routing :class:`~qiskit.transpiler.PassManager`
Args:
routing_pass (TransformationPass): The pass which will perform the
routing
target (Target): the :class:`~.Target` object representing the backend
coupling_map (CouplingMap): The coupling map of the backend to route
for
vf2_call_limit (int): The internal call limit for the vf2 post layout
pass. If this is ``None`` or ``0`` the vf2 post layout will not be
run.
backend_properties (BackendProperties): Properties of a backend to
synthesize for (e.g. gate fidelities).
seed_transpiler (int): Sets random seed for the stochastic parts of
the transpiler.
check_trivial (bool): If set to true this will condition running the
:class:`~.VF2PostLayout` pass after routing on whether a trivial
layout was tried and was found to not be perfect. This is only
needed if the constructed pass manager runs :class:`~.TrivialLayout`
as a first layout attempt and uses it if it's a perfect layout
(as is the case with preset pass manager level 1).
use_barrier_before_measurement (bool): If true (the default) the
:class:`~.BarrierBeforeFinalMeasurements` transpiler pass will be run prior to the
specified pass in the ``routing_pass`` argument.
vf2_max_trials (int): The maximum number of trials to run VF2 when
evaluating the vf2 post layout
pass. If this is ``None`` or ``0`` the vf2 post layout will not be run.
Returns:
PassManager: The routing pass manager
"""
def _run_post_layout_condition(property_set):
# If we check trivial layout and the found trivial layout was not perfect also
# ensure VF2 initial layout was not used before running vf2 post layout
if not check_trivial or _layout_not_perfect(property_set):
vf2_stop_reason = property_set["VF2Layout_stop_reason"]
if vf2_stop_reason is None or vf2_stop_reason != VF2LayoutStopReason.SOLUTION_FOUND:
return True
return False
routing = PassManager()
if target is not None:
routing.append(CheckMap(target, property_set_field="routing_not_needed"))
else:
routing.append(CheckMap(coupling_map, property_set_field="routing_not_needed"))
def _swap_condition(property_set):
return not property_set["routing_not_needed"]
if use_barrier_before_measurement:
routing.append(
ConditionalController(
[
BarrierBeforeFinalMeasurements(
label="qiskit.transpiler.internal.routing.protection.barrier"
),
routing_pass,
],
condition=_swap_condition,
)
)
else:
routing.append(ConditionalController(routing_pass, condition=_swap_condition))
is_vf2_fully_bounded = vf2_call_limit and vf2_max_trials
if (target is not None or backend_properties is not None) and is_vf2_fully_bounded:
routing.append(
ConditionalController(
VF2PostLayout(
target,
coupling_map,
backend_properties,
seed_transpiler,
call_limit=vf2_call_limit,
max_trials=vf2_max_trials,
strict_direction=False,
),
condition=_run_post_layout_condition,
)
)
routing.append(ConditionalController(ApplyLayout(), condition=_apply_post_layout_condition))
def filter_fn(node):
return (
getattr(node.op, "label", None)
!= "qiskit.transpiler.internal.routing.protection.barrier"
)
routing.append([FilterOpNodes(filter_fn)])
return routing
def generate_pre_op_passmanager(target=None, coupling_map=None, remove_reset_in_zero=False):
"""Generate a pre-optimization loop :class:`~qiskit.transpiler.PassManager`
This pass manager will check to ensure that directionality from the coupling
map is respected
Args:
target (Target): the :class:`~.Target` object representing the backend
coupling_map (CouplingMap): The coupling map to use
remove_reset_in_zero (bool): If ``True`` include the remove reset in
zero pass in the generated PassManager
Returns:
PassManager: The pass manager
"""
pre_opt = PassManager()
if coupling_map:
pre_opt.append(CheckGateDirection(coupling_map, target=target))
def _direction_condition(property_set):
return not property_set["is_direction_mapped"]
pre_opt.append(
ConditionalController(
[GateDirection(coupling_map, target=target)],
condition=_direction_condition,
)
)
if remove_reset_in_zero:
pre_opt.append(RemoveResetInZeroState())
return pre_opt
def generate_translation_passmanager(
target,
basis_gates=None,
method="translator",
approximation_degree=None,
coupling_map=None,
backend_props=None,
unitary_synthesis_method="default",
unitary_synthesis_plugin_config=None,
hls_config=None,
):
"""Generate a basis translation :class:`~qiskit.transpiler.PassManager`
Args:
target (Target): the :class:`~.Target` object representing the backend
basis_gates (list): A list of str gate names that represent the basis
gates on the backend target
method (str): The basis translation method to use
approximation_degree (Optional[float]): The heuristic approximation degree to
use. Can be between 0 and 1.
coupling_map (CouplingMap): the coupling map of the backend
in case synthesis is done on a physical circuit. The
directionality of the coupling_map will be taken into
account if pulse_optimize is True/None and natural_direction
is True/None.
unitary_synthesis_plugin_config (dict): The optional dictionary plugin
configuration, this is plugin specific refer to the specified plugin's
documentation for how to use.
backend_props (BackendProperties): Properties of a backend to
synthesize for (e.g. gate fidelities).
unitary_synthesis_method (str): The unitary synthesis method to use. You can
see a list of installed plugins with :func:`.unitary_synthesis_plugin_names`.
hls_config (HLSConfig): An optional configuration class to use for
:class:`~qiskit.transpiler.passes.HighLevelSynthesis` pass.
Specifies how to synthesize various high-level objects.
Returns:
PassManager: The basis translation pass manager
Raises:
TranspilerError: If the ``method`` kwarg is not a valid value
"""
if method == "translator":
unroll = [
# Use unitary synthesis for basis aware decomposition of
# UnitaryGates before custom unrolling
UnitarySynthesis(
basis_gates,
approximation_degree=approximation_degree,
coupling_map=coupling_map,
backend_props=backend_props,
plugin_config=unitary_synthesis_plugin_config,
method=unitary_synthesis_method,
target=target,
),
HighLevelSynthesis(
hls_config=hls_config,
coupling_map=coupling_map,
target=target,
use_qubit_indices=True,
equivalence_library=sel,
basis_gates=basis_gates,
),
BasisTranslator(sel, basis_gates, target),
]
elif method == "synthesis":
unroll = [
# # Use unitary synthesis for basis aware decomposition of
# UnitaryGates > 2q before collection
UnitarySynthesis(
basis_gates,
approximation_degree=approximation_degree,
coupling_map=coupling_map,
backend_props=backend_props,
plugin_config=unitary_synthesis_plugin_config,
method=unitary_synthesis_method,
min_qubits=3,
target=target,
),
HighLevelSynthesis(
hls_config=hls_config,
coupling_map=coupling_map,
target=target,
use_qubit_indices=True,
basis_gates=basis_gates,
min_qubits=3,
),
Unroll3qOrMore(target=target, basis_gates=basis_gates),
Collect2qBlocks(),
Collect1qRuns(),
ConsolidateBlocks(
basis_gates=basis_gates, target=target, approximation_degree=approximation_degree
),
UnitarySynthesis(
basis_gates=basis_gates,
approximation_degree=approximation_degree,
coupling_map=coupling_map,
backend_props=backend_props,
plugin_config=unitary_synthesis_plugin_config,
method=unitary_synthesis_method,
target=target,
),
HighLevelSynthesis(
hls_config=hls_config,
coupling_map=coupling_map,
target=target,
use_qubit_indices=True,
basis_gates=basis_gates,
),
]
else:
raise TranspilerError("Invalid translation method %s." % method)
return PassManager(unroll)
def generate_scheduling(
instruction_durations, scheduling_method, timing_constraints, inst_map, target=None
):
"""Generate a post optimization scheduling :class:`~qiskit.transpiler.PassManager`
Args:
instruction_durations (dict): The dictionary of instruction durations
scheduling_method (str): The scheduling method to use, can either be
``'asap'``/``'as_soon_as_possible'`` or
``'alap'``/``'as_late_as_possible'``
timing_constraints (TimingConstraints): Hardware time alignment restrictions.
inst_map (InstructionScheduleMap): Mapping object that maps gate to schedule.
target (Target): The :class:`~.Target` object representing the backend
Returns:
PassManager: The scheduling pass manager
Raises:
TranspilerError: If the ``scheduling_method`` kwarg is not a valid value
"""
scheduling = PassManager()
if inst_map and inst_map.has_custom_gate():
scheduling.append(PulseGates(inst_map=inst_map, target=target))
if scheduling_method:
# Do scheduling after unit conversion.
scheduler = {
"alap": ALAPScheduleAnalysis,
"as_late_as_possible": ALAPScheduleAnalysis,
"asap": ASAPScheduleAnalysis,
"as_soon_as_possible": ASAPScheduleAnalysis,
}
scheduling.append(TimeUnitConversion(instruction_durations, target=target))
try:
scheduling.append(scheduler[scheduling_method](instruction_durations, target=target))
except KeyError as ex:
raise TranspilerError("Invalid scheduling method %s." % scheduling_method) from ex
elif instruction_durations:
# No scheduling. But do unit conversion for delays.
def _contains_delay(property_set):
return property_set["contains_delay"]
scheduling.append(ContainsInstruction("delay"))
scheduling.append(
ConditionalController(
TimeUnitConversion(instruction_durations, target=target), condition=_contains_delay
)
)
if (
timing_constraints.granularity != 1
or timing_constraints.min_length != 1
or timing_constraints.acquire_alignment != 1
or timing_constraints.pulse_alignment != 1
):
# Run alignment analysis regardless of scheduling.
def _require_alignment(property_set):
return property_set["reschedule_required"]
scheduling.append(
InstructionDurationCheck(
acquire_alignment=timing_constraints.acquire_alignment,
pulse_alignment=timing_constraints.pulse_alignment,
target=target,
)
)
scheduling.append(
ConditionalController(
ConstrainedReschedule(
acquire_alignment=timing_constraints.acquire_alignment,
pulse_alignment=timing_constraints.pulse_alignment,
target=target,
),
condition=_require_alignment,
)
)
scheduling.append(
ValidatePulseGates(
granularity=timing_constraints.granularity,
min_length=timing_constraints.min_length,
target=target,
)
)
if scheduling_method:
# Call padding pass if circuit is scheduled
scheduling.append(PadDelay(target=target))
return scheduling
VF2Limits = collections.namedtuple("VF2Limits", ("call_limit", "max_trials"))
def get_vf2_limits(
optimization_level: int,
layout_method: Optional[str] = None,
initial_layout: Optional[Layout] = None,
) -> VF2Limits:
"""Get the VF2 limits for VF2-based layout passes.
Returns:
VF2Limits: An namedtuple with optional elements
``call_limit`` and ``max_trials``.
"""
limits = VF2Limits(None, None)
if layout_method is None and initial_layout is None:
if optimization_level in {1, 2}:
limits = VF2Limits(
int(5e4), # Set call limit to ~100ms with rustworkx 0.10.2
2500, # Limits layout scoring to < 600ms on ~400 qubit devices
)
elif optimization_level == 3:
limits = VF2Limits(
int(3e7), # Set call limit to ~60 sec with rustworkx 0.10.2
250000, # Limits layout scoring to < 60 sec on ~400 qubit devices
)
return limits