Repository: Qiskit/qiskit-terra (python)

Issue:
Big overhead in parallel transpile for large target systems
### Environment

- **Qiskit Terra version**: `0.19.2`
- **Python version**: 3.8
- **Operating system**: Linux


### What is happening?

There is a large overhead when doing parallel transpilation for circuits targeting larger systems.  In the case below they are targeting the 127Q system, but the circuits themselves consist of only `x` gates and are already the correct width.  Thus transpilation should just be a pass-through.  However, in parallel mode (quad-core machine) it takes 330sec to transpile.  If I turn off parallel transpile it takes 3sec.

```python

from qiskit import *
from mthree.circuits import balanced_cal_strings, balanced_cal_circuits

b_strs = balanced_cal_strings(127)
cal_circs = balanced_cal_circuits(b_strs)

backend = provider.backend.ibm_washington
trans_circs = transpile(cal_circs, backend, optimization_level=0)
```

### How can we reproduce the issue?

Run above

### What should happen?

Parallel should not give 100x overhead. 

### Any suggestions?

I am guessing this is copying overhead from forking when the systems are large.

Pull request description:
Leverage Python SharedMemory for common transpile args
<!--
⚠️ If you do not respect this template, your pull request will be closed.
⚠️ Your pull request title should be short detailed and understandable for all.
⚠️ Also, please add a release note file using reno if the change needs to be
  documented in the release notes.
⚠️ If your pull request fixes an open issue, please link to the issue.

- [ ] I have added the tests to cover my changes.
- [ ] I have updated the documentation accordingly.
- [ ] I have read the CONTRIBUTING document.
-->

### Summary

This commit uses the Python shared memory library to reduce the overhead
of launching parallel processes as part of transpile. As the size of the
backends grow the payload size we're serializing and copying between
worker processes is also increasing. When we're running a lot of small
circuits at once on a big backend we can easily spend far more time
dealing with IO overhead than running the transpilation. By using shared
memory this reduces the overhead to only serializing and copying it once
and then each worker process just needs to serializing it. While this
doesn't remove all the overhead it should reduce the impact somewhat.

### Details and comments

Partially Addresses: #7741

TODO:

- [x] Fix failing tests
- [x] Cleanup code
- [x] Only use shared memory if we're going to run in parallel (the overhead is high for serial execution)

Patch:
diff --git a/qiskit/compiler/transpiler.py b/qiskit/compiler/transpiler.py
index 39ffc2a1d..ac9ccc7c0 100644
--- a/qiskit/compiler/transpiler.py
+++ b/qiskit/compiler/transpiler.py
@@ -10,11 +10,18 @@
 # copyright notice, and modified files need to carry a notice indicating
 # that they have been altered from the originals.
 
+# pylint: disable=import-error,invalid-sequence-index
+
 """Circuit transpile function"""
+import io
+from itertools import cycle
 import logging
-import warnings
+import os
+import pickle
+import sys
 from time import time
 from typing import List, Union, Dict, Callable, Any, Optional, Tuple, Iterable
+import warnings
 
 from qiskit import user_config
 from qiskit.circuit.quantumcircuit import QuantumCircuit
@@ -25,7 +32,7 @@ from qiskit.providers.backend import Backend
 from qiskit.providers.models import BackendProperties
 from qiskit.providers.models.backendproperties import Gate
 from qiskit.pulse import Schedule, InstructionScheduleMap
-from qiskit.tools.parallel import parallel_map
+from qiskit.tools import parallel
 from qiskit.transpiler import Layout, CouplingMap, PropertySet
 from qiskit.transpiler.basepasses import BasePass
 from qiskit.transpiler.exceptions import TranspilerError
@@ -41,6 +48,12 @@ from qiskit.transpiler.preset_passmanagers import (
 from qiskit.transpiler.timing_constraints import TimingConstraints
 from qiskit.transpiler.target import Target, target_to_backend_properties
 
+if sys.version_info >= (3, 8):
+    from multiprocessing.shared_memory import SharedMemory  # pylint: disable=no-name-in-module
+    from multiprocessing.managers import SharedMemoryManager  # pylint: disable=no-name-in-module
+else:
+    from shared_memory import SharedMemory, SharedMemoryManager
+
 logger = logging.getLogger(__name__)
 
 
@@ -274,8 +287,7 @@ def transpile(
             UserWarning,
         )
 
-    # Get transpile_args to configure the circuit transpilation job(s)
-    transpile_args = _parse_transpile_args(
+    unique_transpile_args, shared_args = _parse_transpile_args(
         circuits,
         backend,
         basis_gates,
@@ -299,12 +311,44 @@ def transpile(
         unitary_synthesis_plugin_config,
         target,
     )
-
-    _check_circuits_coupling_map(circuits, transpile_args, backend)
-
-    # Transpile circuits in parallel
-    circuits = parallel_map(_transpile_circuit, list(zip(circuits, transpile_args)))
-
+    # Get transpile_args to configure the circuit transpilation job(s)
+    if coupling_map in unique_transpile_args:
+        cmap_conf = unique_transpile_args["coupling_map"]
+    else:
+        cmap_conf = [shared_args["coupling_map"]] * len(circuits)
+    _check_circuits_coupling_map(circuits, cmap_conf, backend)
+    if (
+        len(circuits) > 1
+        and os.getenv("QISKIT_IN_PARALLEL", "FALSE") == "FALSE"
+        and parallel.PARALLEL_DEFAULT
+    ):
+        with SharedMemoryManager() as smm:
+            with io.BytesIO() as buf:
+                pickle.dump(shared_args, buf)
+                data = buf.getvalue()
+            smb = smm.SharedMemory(size=len(data))
+            smb.buf[: len(data)] = data[:]
+            # Transpile circuits in parallel
+            circuits = parallel.parallel_map(
+                _transpile_circuit,
+                list(zip(circuits, cycle([smb.name]), unique_transpile_args)),
+            )
+    else:
+        output_circuits = []
+        for circuit, unique_args in zip(circuits, unique_transpile_args):
+            transpile_config, pass_manager = _combine_args(shared_args, unique_args)
+            output_circuits.append(
+                _serial_transpile_circuit(
+                    circuit,
+                    pass_manager,
+                    transpile_config["callback"],
+                    transpile_config["output_name"],
+                    transpile_config["backend_num_qubits"],
+                    transpile_config["faulty_qubits_map"],
+                    transpile_config["pass_manager_config"].backend_properties,
+                )
+            )
+        circuits = output_circuits
     end_time = time()
     _log_transpile_time(start_time, end_time)
 
@@ -314,11 +358,9 @@ def transpile(
         return circuits[0]
 
 
-def _check_circuits_coupling_map(circuits, transpile_args, backend):
+def _check_circuits_coupling_map(circuits, cmap_conf, backend):
     # Check circuit width against number of qubits in coupling_map(s)
-    coupling_maps_list = list(
-        config["pass_manager_config"].coupling_map for config in transpile_args
-    )
+    coupling_maps_list = cmap_conf
     for circuit, parsed_coupling_map in zip(circuits, coupling_maps_list):
         # If coupling_map is not None or num_qubits == 1
         num_qubits = len(circuit.qubits)
@@ -347,25 +389,19 @@ def _log_transpile_time(start_time, end_time):
     logger.info(log_msg)
 
 
-def _transpile_circuit(circuit_config_tuple: Tuple[QuantumCircuit, Dict]) -> QuantumCircuit:
-    """Select a PassManager and run a single circuit through it.
-    Args:
-        circuit_config_tuple (tuple):
-            circuit (QuantumCircuit): circuit to transpile
-            transpile_config (dict): configuration dictating how to transpile. The
-                dictionary has the following format:
-                {'optimization_level': int,
-                 'output_name': string,
-                 'callback': callable,
-                 'pass_manager_config': PassManagerConfig}
-    Returns:
-        The transpiled circuit
-    Raises:
-        TranspilerError: if transpile_config is not valid or transpilation incurs error
-    """
-    circuit, transpile_config = circuit_config_tuple
+def _combine_args(shared_transpiler_args, unique_config):
+    # Pop optimization_level to exclude it from the kwargs when building a
+    # PassManagerConfig
+    level = shared_transpiler_args.pop("optimization_level")
+    pass_manager_config = shared_transpiler_args
+    pass_manager_config.update(unique_config.pop("pass_manager_config"))
+    pass_manager_config = PassManagerConfig(**pass_manager_config)
+    # restore optimization_level in the input shared dict in case it's used again
+    # in the same process
+    shared_transpiler_args["optimization_level"] = level
 
-    pass_manager_config = transpile_config["pass_manager_config"]
+    transpile_config = unique_config
+    transpile_config["pass_manager_config"] = pass_manager_config
 
     if transpile_config["faulty_qubits_map"]:
         pass_manager_config.initial_layout = _remap_layout_faulty_backend(
@@ -373,8 +409,6 @@ def _transpile_circuit(circuit_config_tuple: Tuple[QuantumCircuit, Dict]) -> Qua
         )
 
     # we choose an appropriate one based on desired optimization level
-    level = transpile_config["optimization_level"]
-
     if level == 0:
         pass_manager = level_0_pass_manager(pass_manager_config)
     elif level == 1:
@@ -385,6 +419,53 @@ def _transpile_circuit(circuit_config_tuple: Tuple[QuantumCircuit, Dict]) -> Qua
         pass_manager = level_3_pass_manager(pass_manager_config)
     else:
         raise TranspilerError("optimization_level can range from 0 to 3.")
+    return transpile_config, pass_manager
+
+
+def _serial_transpile_circuit(
+    circuit,
+    pass_manager,
+    callback,
+    output_name,
+    num_qubits,
+    faulty_qubits_map=None,
+    backend_prop=None,
+):
+    result = pass_manager.run(circuit, callback=callback, output_name=output_name)
+    if faulty_qubits_map:
+        return _remap_circuit_faulty_backend(
+            result,
+            num_qubits,
+            backend_prop,
+            faulty_qubits_map,
+        )
+
+    return result
+
+
+def _transpile_circuit(circuit_config_tuple: Tuple[QuantumCircuit, str, Dict]) -> QuantumCircuit:
+    """Select a PassManager and run a single circuit through it.
+    Args:
+        circuit_config_tuple (tuple):
+            circuit (QuantumCircuit): circuit to transpile
+            name (str): The name of the shared memory object containing a pickled dict of shared
+                arguments between parallel works
+            unique_config (dict): configuration dictating unique arguments for transpile.
+    Returns:
+        The transpiled circuit
+    Raises:
+        TranspilerError: if transpile_config is not valid or transpilation incurs error
+    """
+    circuit, name, unique_config = circuit_config_tuple
+    existing_shm = SharedMemory(name=name)
+    try:
+        with io.BytesIO(existing_shm.buf) as buf:
+            shared_transpiler_args = pickle.load(buf)
+    finally:
+        existing_shm.close()
+
+    transpile_config, pass_manager = _combine_args(shared_transpiler_args, unique_config)
+    pass_manager_config = transpile_config["pass_manager_config"]
 
     result = pass_manager.run(
         circuit, callback=transpile_config["callback"], output_name=transpile_config["output_name"]
@@ -479,7 +560,7 @@ def _parse_transpile_args(
     unitary_synthesis_method,
     unitary_synthesis_plugin_config,
     target,
-) -> List[Dict]:
+) -> Tuple[List[Dict], Dict]:
     """Resolve the various types of args allowed to the transpile() function through
     duck typing, overriding args, etc. Refer to the transpile() docstring for details on
     what types of inputs are allowed.
@@ -489,7 +570,8 @@ def _parse_transpile_args(
     arg has more priority than the arg set by backend).
 
     Returns:
-        list[dicts]: a list of transpile parameters.
+        Tuple[list[dict], dict]: a tuple contain a list of unique transpile parameter dicts and
+        the second element contains a dict of shared transpiler argument across all circuits.
 
     Raises:
         TranspilerError: If instruction_durations are required but not supplied or found.
@@ -518,91 +600,69 @@ def _parse_transpile_args(
         if backend_properties is None:
             backend_properties = target_to_backend_properties(target)
 
-    basis_gates = _parse_basis_gates(basis_gates, backend, circuits)
-    inst_map = _parse_inst_map(inst_map, backend, num_circuits)
+    basis_gates = _parse_basis_gates(basis_gates, backend)
+    initial_layout = _parse_initial_layout(initial_layout, circuits)
+    inst_map = _parse_inst_map(inst_map, backend)
     faulty_qubits_map = _parse_faulty_qubits_map(backend, num_circuits)
-    coupling_map = _parse_coupling_map(coupling_map, backend, num_circuits)
-    backend_properties = _parse_backend_properties(backend_properties, backend, num_circuits)
+    coupling_map = _parse_coupling_map(coupling_map, backend)
+    backend_properties = _parse_backend_properties(backend_properties, backend)
     backend_num_qubits = _parse_backend_num_qubits(backend, num_circuits)
-    initial_layout = _parse_initial_layout(initial_layout, circuits)
-    layout_method = _parse_layout_method(layout_method, num_circuits)
-    routing_method = _parse_routing_method(routing_method, num_circuits)
-    translation_method = _parse_translation_method(translation_method, num_circuits)
-    approximation_degree = _parse_approximation_degree(approximation_degree, num_circuits)
-    unitary_synthesis_method = _parse_unitary_synthesis_method(
-        unitary_synthesis_method, num_circuits
-    )
-    unitary_synthesis_plugin_config = _parse_unitary_plugin_config(
-        unitary_synthesis_plugin_config, num_circuits
-    )
-    seed_transpiler = _parse_seed_transpiler(seed_transpiler, num_circuits)
-    optimization_level = _parse_optimization_level(optimization_level, num_circuits)
+    approximation_degree = _parse_approximation_degree(approximation_degree)
     output_name = _parse_output_name(output_name, circuits)
     callback = _parse_callback(callback, num_circuits)
     durations = _parse_instruction_durations(backend, instruction_durations, dt, circuits)
-    scheduling_method = _parse_scheduling_method(scheduling_method, num_circuits)
     timing_constraints = _parse_timing_constraints(backend, timing_constraints, num_circuits)
-    target = _parse_target(backend, target, num_circuits)
+    target = _parse_target(backend, target)
     if scheduling_method and any(d is None for d in durations):
         raise TranspilerError(
             "Transpiling a circuit with a scheduling method"
             "requires a backend or instruction_durations."
         )
+    unique_dict = {
+        "callback": callback,
+        "output_name": output_name,
+        "faulty_qubits_map": faulty_qubits_map,
+        "backend_num_qubits": backend_num_qubits,
+    }
+    shared_dict = {
+        "optimization_level": optimization_level,
+        "basis_gates": basis_gates,
+    }
 
     list_transpile_args = []
-    for kwargs in _zip_dict(
-        {
-            "basis_gates": basis_gates,
-            "inst_map": inst_map,
-            "coupling_map": coupling_map,
-            "backend_properties": backend_properties,
-            "initial_layout": initial_layout,
-            "layout_method": layout_method,
-            "routing_method": routing_method,
-            "translation_method": translation_method,
-            "scheduling_method": scheduling_method,
-            "durations": durations,
-            "approximation_degree": approximation_degree,
-            "timing_constraints": timing_constraints,
-            "seed_transpiler": seed_transpiler,
-            "optimization_level": optimization_level,
-            "output_name": output_name,
-            "callback": callback,
-            "backend_num_qubits": backend_num_qubits,
-            "faulty_qubits_map": faulty_qubits_map,
-            "unitary_synthesis_method": unitary_synthesis_method,
-            "unitary_synthesis_plugin_config": unitary_synthesis_plugin_config,
-            "target": target,
-        }
-    ):
+    for key, value in {
+        "inst_map": inst_map,
+        "coupling_map": coupling_map,
+        "backend_properties": backend_properties,
+        "approximation_degree": approximation_degree,
+        "initial_layout": initial_layout,
+        "layout_method": layout_method,
+        "routing_method": routing_method,
+        "translation_method": translation_method,
+        "scheduling_method": scheduling_method,
+        "instruction_durations": durations,
+        "timing_constraints": timing_constraints,
+        "seed_transpiler": seed_transpiler,
+        "unitary_synthesis_method": unitary_synthesis_method,
+        "unitary_synthesis_plugin_config": unitary_synthesis_plugin_config,
+        "target": target,
+    }.items():
+        if isinstance(value, list):
+            unique_dict[key] = value
+        else:
+            shared_dict[key] = value
+
+    for kwargs in _zip_dict(unique_dict):
         transpile_args = {
-            "pass_manager_config": PassManagerConfig(
-                basis_gates=kwargs["basis_gates"],
-                inst_map=kwargs["inst_map"],
-                coupling_map=kwargs["coupling_map"],
-                backend_properties=kwargs["backend_properties"],
-                initial_layout=kwargs["initial_layout"],
-                layout_method=kwargs["layout_method"],
-                routing_method=kwargs["routing_method"],
-                translation_method=kwargs["translation_method"],
-                scheduling_method=kwargs["scheduling_method"],
-                instruction_durations=kwargs["durations"],
-                approximation_degree=kwargs["approximation_degree"],
-                timing_constraints=kwargs["timing_constraints"],
-                seed_transpiler=kwargs["seed_transpiler"],
-                unitary_synthesis_method=kwargs["unitary_synthesis_method"],
-                unitary_synthesis_plugin_config=kwargs["unitary_synthesis_plugin_config"],
-                target=kwargs["target"],
-            ),
-            "optimization_level": kwargs["optimization_level"],
-            "output_name": kwargs["output_name"],
-            "callback": kwargs["callback"],
-            "backend_num_qubits": kwargs["backend_num_qubits"],
-            "faulty_qubits_map": kwargs["faulty_qubits_map"],
+            "output_name": kwargs.pop("output_name"),
+            "callback": kwargs.pop("callback"),
+            "faulty_qubits_map": kwargs.pop("faulty_qubits_map"),
+            "backend_num_qubits": kwargs.pop("backend_num_qubits"),
+            "pass_manager_config": kwargs,
         }
         list_transpile_args.append(transpile_args)
 
-    return list_transpile_args
+    return list_transpile_args, shared_dict
 
 
 def _create_faulty_qubits_map(backend):
@@ -641,7 +701,7 @@ def _create_faulty_qubits_map(backend):
     return faulty_qubits_map
 
 
-def _parse_basis_gates(basis_gates, backend, circuits):
+def _parse_basis_gates(basis_gates, backend):
     # try getting basis_gates from user, else backend
     if basis_gates is None:
         backend_version = getattr(backend, "version", 0)
@@ -650,16 +710,10 @@ def _parse_basis_gates(basis_gates, backend, circuits):
                 basis_gates = getattr(backend.configuration(), "basis_gates", None)
         else:
             basis_gates = backend.operation_names
-    # basis_gates could be None, or a list of basis, e.g. ['u3', 'cx']
-    if basis_gates is None or (
-        isinstance(basis_gates, list) and all(isinstance(i, str) for i in basis_gates)
-    ):
-        basis_gates = [basis_gates] * len(circuits)
-
     return basis_gates
 
 
-def _parse_inst_map(inst_map, backend, num_circuits):
+def _parse_inst_map(inst_map, backend):
     # try getting inst_map from user, else backend
     if inst_map is None:
         backend_version = getattr(backend, "version", 0)
@@ -668,14 +722,10 @@ def _parse_inst_map(inst_map, backend, num_circuits):
                 inst_map = getattr(backend.defaults(), "instruction_schedule_map", None)
         else:
             inst_map = backend.target.instruction_schedule_map()
-    # inst_maps could be None, or single entry
-    if inst_map is None or isinstance(inst_map, InstructionScheduleMap):
-        inst_map = [inst_map] * num_circuits
-
     return inst_map
 
 
-def _parse_coupling_map(coupling_map, backend, num_circuits):
+def _parse_coupling_map(coupling_map, backend):
     # try getting coupling_map from user, else backend
     if coupling_map is None:
         backend_version = getattr(backend, "version", 0)
@@ -707,18 +757,17 @@ def _parse_coupling_map(coupling_map, backend, num_circuits):
 
     # coupling_map could be None, or a list of lists, e.g. [[0, 1], [2, 1]]
     if coupling_map is None or isinstance(coupling_map, CouplingMap):
-        coupling_map = [coupling_map] * num_circuits
-    elif isinstance(coupling_map, list) and all(
+        return coupling_map
+    if isinstance(coupling_map, list) and all(
         isinstance(i, list) and len(i) == 2 for i in coupling_map
     ):
-        coupling_map = [coupling_map] * num_circuits
+        return CouplingMap(coupling_map)
 
     coupling_map = [CouplingMap(cm) if isinstance(cm, list) else cm for cm in coupling_map]
-
     return coupling_map
 
 
-def _parse_backend_properties(backend_properties, backend, num_circuits):
+def _parse_backend_properties(backend_properties, backend):
     # try getting backend_properties from user, else backend
     if backend_properties is None:
         backend_version = getattr(backend, "version", 0)
@@ -754,8 +803,6 @@ def _parse_backend_properties(backend_properties, backend, num_circuits):
                     backend_properties.gates = gates
         else:
             backend_properties = target_to_backend_properties(backend.target)
-    if not isinstance(backend_properties, list):
-        backend_properties = [backend_properties] * num_circuits
     return backend_properties
 
 
@@ -807,36 +854,9 @@ def _parse_initial_layout(initial_layout, circuits):
         # even if one layout, but multiple circuits, the layout needs to be adapted for each
         initial_layout = [_layout_from_raw(initial_layout, circ) for circ in circuits]
 
-    if not isinstance(initial_layout, list):
-        initial_layout = [initial_layout] * len(circuits)
-
     return initial_layout
 
 
-def _parse_layout_method(layout_method, num_circuits):
-    if not isinstance(layout_method, list):
-        layout_method = [layout_method] * num_circuits
-    return layout_method
-
-
-def _parse_routing_method(routing_method, num_circuits):
-    if not isinstance(routing_method, list):
-        routing_method = [routing_method] * num_circuits
-    return routing_method
-
-
-def _parse_translation_method(translation_method, num_circuits):
-    if not isinstance(translation_method, list):
-        translation_method = [translation_method] * num_circuits
-    return translation_method
-
-
-def _parse_scheduling_method(scheduling_method, num_circuits):
-    if not isinstance(scheduling_method, list):
-        scheduling_method = [scheduling_method] * num_circuits
-    return scheduling_method
-
-
 def _parse_instruction_durations(backend, inst_durations, dt, circuits):
     """Create a list of ``InstructionDuration``s. If ``inst_durations`` is provided,
     the backend will be ignored, otherwise, the durations will be populated from the
@@ -874,47 +894,25 @@ def _parse_instruction_durations(backend, inst_durations, dt, circuits):
     return durations
 
 
-def _parse_approximation_degree(approximation_degree, num_circuits):
+def _parse_approximation_degree(approximation_degree):
+    if approximation_degree is None:
+        return None
     if not isinstance(approximation_degree, list):
-        approximation_degree = [approximation_degree] * num_circuits
-    if not all(0.0 <= d <= 1.0 for d in approximation_degree if d):
-        raise TranspilerError("Approximation degree must be in [0.0, 1.0]")
+        if approximation_degree < 0.0 or approximation_degree > 1.0:
+            raise TranspilerError("Approximation degree must be in [0.0, 1.0]")
+    else:
+        if not all(0.0 <= d <= 1.0 for d in approximation_degree if d):
+            raise TranspilerError("Approximation degree must be in [0.0, 1.0]")
     return approximation_degree
 
 
-def _parse_unitary_synthesis_method(unitary_synthesis_method, num_circuits):
-    if not isinstance(unitary_synthesis_method, list):
-        unitary_synthesis_method = [unitary_synthesis_method] * num_circuits
-    return unitary_synthesis_method
-
-
-def _parse_unitary_plugin_config(unitary_synthesis_plugin_config, num_circuits):
-    if not isinstance(unitary_synthesis_plugin_config, list):
-        unitary_synthesis_plugin_config = [unitary_synthesis_plugin_config] * num_circuits
-    return unitary_synthesis_plugin_config
-
-
-def _parse_target(backend, target, num_circuits):
+def _parse_target(backend, target):
     backend_target = getattr(backend, "target", None)
     if target is None:
         target = backend_target
-    if not isinstance(target, list):
-        target = [target] * num_circuits
     return target
 
 
-def _parse_seed_transpiler(seed_transpiler, num_circuits):
-    if not isinstance(seed_transpiler, list):
-        seed_transpiler = [seed_transpiler] * num_circuits
-    return seed_transpiler
-
-
-def _parse_optimization_level(optimization_level, num_circuits):
-    if not isinstance(optimization_level, list):
-        optimization_level = [optimization_level] * num_circuits
-    return optimization_level
-
-
 def _parse_callback(callback, num_circuits):
     if not isinstance(callback, list):
         callback = [callback] * num_circuits
diff --git a/qiskit/transpiler/passes/synthesis/unitary_synthesis.py b/qiskit/transpiler/passes/synthesis/unitary_synthesis.py
index a979812cd..752515c7b 100644
--- a/qiskit/transpiler/passes/synthesis/unitary_synthesis.py
+++ b/qiskit/transpiler/passes/synthesis/unitary_synthesis.py
@@ -569,15 +569,21 @@ class DefaultUnitarySynthesis(plugin.UnitarySynthesisPlugin):
         if natural_direction in {None, True} and (
             coupling_map or (target is not None and decomposer2q and not preferred_direction)
         ):
-            cmap = coupling_map
-            neighbors0 = cmap.neighbors(qubits[0])
-            zero_one = qubits[1] in neighbors0
-            neighbors1 = cmap.neighbors(qubits[1])
-            one_zero = qubits[0] in neighbors1
-            if zero_one and not one_zero:
-                preferred_direction = [0, 1]
-            if one_zero and not zero_one:
-                preferred_direction = [1, 0]
+            if coupling_map is not None:
+                cmap = coupling_map
+            else:
+                cmap = target.build_coupling_map()
+            # If we don't have a defined coupling map (either from the input)
+            # or from the target we can't check for a natural direction
+            if cmap is not None:
+                neighbors0 = cmap.neighbors(qubits[0])
+                zero_one = qubits[1] in neighbors0
+                neighbors1 = cmap.neighbors(qubits[1])
+                one_zero = qubits[0] in neighbors1
+                if zero_one and not one_zero:
+                    preferred_direction = [0, 1]
+                if one_zero and not zero_one:
+                    preferred_direction = [1, 0]
         if (
             natural_direction in {None, True}
             and preferred_direction is None
diff --git a/releasenotes/notes/shared-memory-dependency-1e32e1c55902216f.yaml b/releasenotes/notes/shared-memory-dependency-1e32e1c55902216f.yaml
new file mode 100644
index 000000000..ce92dd95b
--- /dev/null
+++ b/releasenotes/notes/shared-memory-dependency-1e32e1c55902216f.yaml
@@ -0,0 +1,21 @@
+---
+upgrade:
+  - |
+    For Python 3.7 `shared-memory38 <https://pypi.org/project/shared-memory38/>`__
+    is now a dependency. This was added as a dependency for Python 3.7 to enable
+    leveraging the shared memory constructs in the standard library of newer
+    versions of Python. If you're running on Python >= 3.8 there is no extra
+    dependency required.
+fixes:
+  - |
+    Fixed an issue with :func:`~.transpile` where in some cases providing a
+    list of basis gate strings with the ``basis_gates`` keyword argument or
+    implicitly via a :class:`~.Target` input via the ``target`` keyword
+    argument would not be interpreted correctly and result in a subset of the
+    listed gates being used for each circuit.
+  - |
+    Fixed an issue in the :class:`~.UnitarySynthesis` transpiler pass which
+    would result in an error when a :class:`~.Target` that didn't have any
+    qubit restrictions on the operations (e.g. in the case of an ideal
+    simulator target) was specified with the ``target`` keyword argument for the
+    constructor.
diff --git a/requirements.txt b/requirements.txt
index 7eedd7049..19852678a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -9,4 +9,5 @@ python-dateutil>=2.8.0
 stevedore>=3.0.0
 symengine>=0.9 ; platform_machine == 'x86_64' or platform_machine == 'aarch64' or platform_machine == 'ppc64le' or platform_machine == 'amd64' or platform_machine == 'arm64'
 tweedledum>=1.1,<2.0
+shared-memory38;python_version<'3.8'
 typing-extensions; python_version < '3.8'


Test patch:
diff --git a/test/python/compiler/test_transpiler.py b/test/python/compiler/test_transpiler.py
index d90be7257..095b6753c 100644
--- a/test/python/compiler/test_transpiler.py
+++ b/test/python/compiler/test_transpiler.py
@@ -1437,7 +1437,7 @@ class TestTranspile(QiskitTestCase):
         theta = Parameter("θ")
         phi = Parameter("ϕ")
         lam = Parameter("λ")
-        target = Target()
+        target = Target(2)
         target.add_instruction(UGate(theta, phi, lam))
         target.add_instruction(CXGate())
         target.add_instruction(Measure())
@@ -1446,13 +1446,23 @@ class TestTranspile(QiskitTestCase):
         qc = QuantumCircuit(qubit_reg, clbit_reg, name="bell")
         qc.h(qubit_reg[0])
         qc.cx(qubit_reg[0], qubit_reg[1])
-        qc.measure(qubit_reg, clbit_reg)
+        if opt_level != 3:
+            qc.measure(qubit_reg, clbit_reg)
         result = transpile(qc, target=target, optimization_level=opt_level)
-        expected = QuantumCircuit(qubit_reg, clbit_reg)
-        expected.u(np.pi / 2, 0, np.pi, qubit_reg[0])
-        expected.cx(qubit_reg[0], qubit_reg[1])
-        expected.measure(qubit_reg, clbit_reg)
-        self.assertEqual(result, expected)
+        # The Unitary synthesis optimization pass results for optimization level 3
+        # results in a different output than the other optimization levels
+        # and can differ based on fp precision. To avoid relying on a hard match
+        # do a unitary equiv
+
+        if opt_level == 3:
+            result_op = Operator.from_circuit(result)
+            self.assertTrue(result_op.equiv(qc))
+        else:
+            expected = QuantumCircuit(qubit_reg, clbit_reg)
+            expected.u(np.pi / 2, 0, np.pi, qubit_reg[0])
+            expected.cx(qubit_reg[0], qubit_reg[1])
+            expected.measure(qubit_reg, clbit_reg)
+            self.assertEqual(result, expected)
 
 
 class StreamHandlerRaiseException(StreamHandler):
diff --git a/test/python/providers/test_backend_v2.py b/test/python/providers/test_backend_v2.py
index fe12f98ae..a4625d7c3 100644
--- a/test/python/providers/test_backend_v2.py
+++ b/test/python/providers/test_backend_v2.py
@@ -188,5 +188,5 @@ class TestBackendV2(QiskitTestCase):
 
     def test_transpile_parse_inst_map(self):
         """Test that transpiler._parse_inst_map() supports BackendV2."""
-        inst_map = _parse_inst_map(inst_map=None, backend=self.backend, num_circuits=1)[0]
+        inst_map = _parse_inst_map(inst_map=None, backend=self.backend)
         self.assertIsInstance(inst_map, InstructionScheduleMap)