Parallel workflows with QiskitFunction

In this document, we will learn how to run distributed workflows inside a function. In this case, we will compute the quasi-probability distribution in parallel for a list of quantum circuits.

Let’s take a look at the function file ./source_files/function_with_parallel_workflow.py.

from qiskit_serverless import get_arguments, save_result, distribute_task, get

from qiskit import QuantumCircuit
from qiskit.primitives import Sampler


@distribute_task()
def distributed_sample(circuit: QuantumCircuit):
    """Distributed task that returns quasi distribution for given circuit."""
    return Sampler().run(circuit).result().quasi_dists[0]


arguments = get_arguments()
circuits = arguments.get("circuits")


# run distributed tasks as async function
# we get task references as a return type
sample_task_references = [
    distributed_sample(circuit)
    for circuit in circuits
]

# now we need to collect results from task references
results = get(sample_task_references)

save_result({
    "results": results
})

There are a lot of new concepts introduced in this Qiskit Function, so let’s go over them in more detail:

The distribute_task decorator converts a function into a distributed task. This means that the function will be executed on compute resources asynchronously and in parallel to the main context of the Qiskit Function.

When you call a converted function, it will return a reference to the function execution instead of the result. In order to get the result back, you need to call the get function on the function reference. get will wait until the function is finished and return the result of the function execution.

In the Qiskit Function above, we have applied the distribute_task decorator to the distributed_sample function. This function takes a QuantumCircuit as input and returns the quasi distribution for that circuit.

After we have defined the distributed_sample function, we read the circuits from the Qiskit Function arguments using the get_arguments function. We then call the distributed_sample function for each of the circuits, which creates a reference to each of the function executions.

These function executions will run in parallel on compute resources, and we get task references as the return type. We store these task references in the sample_task_references list.

After we have created the task references for each of the function executions, we need to collect the results from these tasks. We do this by calling the get function on the list of task references, which waits until all the tasks have completed and returns the results.

Once we have the results, we can save them using the save_result function.

Essentially, this Qiskit Function reads the circuits from the Qiskit Function arguments, executes the distributed_sample function on each circuit in parallel, collects the results from the function executions, and saves the results.

⚠ This provider is set up with default credentials to a test cluster intended to run on your machine. For information on setting up infrastructure on your local machine, check out the guide on local infrastructure setup.

from qiskit_serverless import ServerlessClient
import os

client = ServerlessClient(
    token=os.environ.get("GATEWAY_TOKEN", "awesome_token"),
    host=os.environ.get("GATEWAY_HOST", "http://localhost:8000"),
    # If you are using the kubernetes approach the URL must be http://localhost
)

client

<gateway-client>

Let’s create a list of random circuits which we will be passed as arguments to the function.

from qiskit.circuit.random import random_circuit

circuits = [random_circuit(2, 2) for _ in range(3)]
[circuit.measure_all() for circuit in circuits]
circuits

[<qiskit.circuit.quantumcircuit.QuantumCircuit at 0x16c07c390>,
 <qiskit.circuit.quantumcircuit.QuantumCircuit at 0x16c054350>,
 <qiskit.circuit.quantumcircuit.QuantumCircuit at 0x16c067ed0>]

Run the function as usual, but pass the circuits in as a keyword argument, circuits.

from qiskit_serverless import QiskitFunction

function = QiskitFunction(
    title="function-with-parallel-workflow",
    entrypoint="function_with_parallel_workflow.py",
    working_dir="./source_files/",
)

client.upload(function)

QiskitFunction(function-with-parallel-workflow)

parallel_function = client.get("function-with-parallel-workflow")
job = parallel_function.run(circuits=circuits)
job

<Job | 8eedfd5f-a903-49f1-9147-ee2140f67e55>

job.status()

'DONE'

job.result()

{'results': [{'00': 1024}, {'00': 1024}, {'01': 1002, '00': 22}]}