จัดการทรัพยากรการคำนวณและข้อมูลของ Qiskit Serverless

เวอร์ชันของแพ็คเกจ

โค้ดในหน้านี้ถูกพัฒนาโดยใช้ข้อกำหนดต่อไปนี้ แนะนำให้ใช้เวอร์ชันเหล่านี้หรือใหม่กว่า

qiskit[all]~=2.0.0
qiskit-ibm-runtime~=0.37.0
qiskit-serverless~=0.22.0

ด้วย Qiskit Serverless คุณสามารถจัดการการคำนวณและข้อมูลใน Qiskit pattern รวมถึง CPU, QPU และตัวเร่งการคำนวณอื่น ๆ

ตั้งค่าสถานะแบบละเอียด

Serverless workload มีหลายขั้นตอนในกระบวนการทำงาน ตามค่าเริ่มต้น สถานะต่อไปนี้สามารถดูได้ด้วย job.status():

• QUEUED: workload อยู่ในคิวรอทรัพยากร classical
• INITIALIZING: workload กำลังถูกตั้งค่า
• RUNNING: workload กำลังรันอยู่บนทรัพยากร classical
• DONE: workload เสร็จสมบูรณ์แล้ว

คุณยังสามารถตั้งค่าสถานะแบบกำหนดเองที่อธิบายขั้นตอนของ workflow เฉพาะได้ดังนี้

# Added by doQumentation — required packages for this notebook
!pip install -q qiskit qiskit-ibm-runtime qiskit-serverless

# This cell is hidden from users, it just creates a new folder
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)

%%writefile ./source_files/status_example.py

from qiskit_serverless import update_status, Job

## If your function has a mapping stage, particularly application functions, you can set the status to "RUNNING: MAPPING" as follows:
update_status(Job.MAPPING)

## While handling transpilation, error suppression, and so forth, you can set the status to "RUNNING: OPTIMIZING_FOR_HARDWARE":
update_status(Job.OPTIMIZING_HARDWARE)

## After you submit jobs to Qiskit Runtime, the underlying quantum job will be queued. You can set status to "RUNNING: WAITING_FOR_QPU":
update_status(Job.WAITING_QPU)

## When the Qiskit Runtime job starts running on the QPU, set the following status "RUNNING: EXECUTING_QPU":
update_status(Job.EXECUTING_QPU)

## Once QPU is completed and post-processing has begun, set the status "RUNNING: POST_PROCESSING":
update_status(Job.POST_PROCESSING)

Writing ./source_files/status_example.py

หลังจาก workload เสร็จสมบูรณ์ (ด้วย save_result()) สถานะนี้จะถูกอัปเดตเป็น DONE โดยอัตโนมัติ

Workflow แบบขนาน

สำหรับงาน classical ที่สามารถประมวลผลแบบขนานได้ ให้ใช้ decorator @distribute_task เพื่อกำหนดข้อกำหนดการคำนวณที่จำเป็นในการทำงาน เริ่มต้นด้วยการเรียกดูตัวอย่าง transpile_remote.py จากหัวข้อ เขียนโปรแกรม Qiskit Serverless แรกของคุณ ด้วยโค้ดต่อไปนี้

โค้ดต่อไปนี้ต้องการให้คุณบันทึก credentials ไว้แล้ว

%%writefile ./source_files/transpile_remote.py

from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime import QiskitRuntimeService
from qiskit_serverless import distribute_task

service = QiskitRuntimeService()

@distribute_task(target={"cpu": 1})
def transpile_remote(circuit, optimization_level, backend):
    """Transpiles an abstract circuit (or list of circuits) into an ISA circuit for a given backend."""
    pass_manager = generate_preset_pass_manager(
        optimization_level=optimization_level,
        backend=service.backend(backend)
    )
    isa_circuit = pass_manager.run(circuit)
    return isa_circuit

Writing ./source_files/transpile_remote.py

ในตัวอย่างนี้ คุณได้ตกแต่งฟังก์ชัน transpile_remote() ด้วย @distribute_task(target={"cpu": 1}) เมื่อรัน จะสร้าง worker task แบบขนานแบบ asynchronous พร้อม CPU core เดียว และคืนค่าเป็น reference สำหรับติดตาม worker หากต้องการดึงผลลัพธ์ ให้ส่ง reference ไปยังฟังก์ชัน get() เราสามารถใช้สิ่งนี้เพื่อรัน task แบบขนานหลายรายการ:

%%writefile --append ./source_files/transpile_remote.py

from time import time
from qiskit_serverless import get, get_arguments, save_result, update_status, Job

# Get arguments
arguments = get_arguments()
circuit = arguments.get("circuit")
optimization_level = arguments.get("optimization_level")
backend = arguments.get("backend")

Appending to ./source_files/transpile_remote.py

%%writefile --append ./source_files/transpile_remote.py
# Start distributed transpilation

update_status(Job.OPTIMIZING_HARDWARE)

start_time = time()
transpile_worker_references = [
    transpile_remote(circuit, optimization_level, backend)
    for circuit in arguments.get("circuit_list")
]

transpiled_circuits = get(transpile_worker_references)
end_time = time()

Appending to ./source_files/transpile_remote.py

%%writefile --append ./source_files/transpile_remote.py
# Save result, with metadata

result = {
    "circuits": transpiled_circuits,
    "metadata": {
        "resource_usage": {
            "RUNNING: OPTIMIZING_FOR_HARDWARE": {
                "CPU_TIME": end_time - start_time,
                "QPU_TIME": 0,
            },
        }
    },
}

save_result(result)

Appending to ./source_files/transpile_remote.py

# This cell is hidden from users.
# It uploads the serverless program and checks it runs.

def test_serverless_job(title, entrypoint):
    # Import in function to stop them interfering with user-facing code
    from qiskit.circuit.random import random_circuit
    from qiskit_serverless import IBMServerlessClient, QiskitFunction
    import time
    import uuid

    title += "_" + uuid.uuid4().hex[:8]
    serverless = IBMServerlessClient()
    transpile_remote_demo = QiskitFunction(
        title=title,
        entrypoint=entrypoint,
        working_dir="./source_files/",
    )
    serverless.upload(transpile_remote_demo)
    job = serverless.get(title).run(
        circuit=random_circuit(3, 3),
        circuit_list=[random_circuit(3, 3) for _ in range(3)],
        backend="ibm_torino",
        optimization_level=1,
    )
    for retry in range(25):
        time.sleep(5)
        status = job.status()
        if status == "DONE":
            print("Job completed successfully")
            return
        if status not in [
            "QUEUED",
            "INITIALIZING",
            "RUNNING",
            "RUNNING: OPTIMIZING_FOR_HARDWARE",
            "DONE",
        ]:
            raise Exception(
                f"Unexpected job status '{status}'.\nHere's the logs:\n"
                + job.logs()
            )
        print(f"Waiting for job (status '{status}')")
    raise Exception("Job did not complete in time")

test_serverless_job(
    title="transpile_remote_serverless_test", entrypoint="transpile_remote.py"
)

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'RUNNING')

Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')

Job completed successfully

สำรวจการกำหนดค่า task ที่แตกต่างกัน

คุณสามารถจัดสรร CPU, GPU และหน่วยความจำสำหรับ task ได้อย่างยืดหยุ่นผ่าน @distribute_task() สำหรับ Qiskit Serverless บน IBM Quantum® Platform แต่ละโปรแกรมมี CPU core 16 core และ RAM 32 GB ซึ่งสามารถจัดสรรแบบไดนามิกตามความต้องการ

CPU core สามารถจัดสรรเป็น core เต็มหรือแบบเศษส่วนได้ตามที่แสดงด้านล่าง

หน่วยความจำจัดสรรเป็นจำนวน bytes โดยจำไว้ว่ามี 1024 bytes ในหนึ่ง kilobyte, 1024 kilobytes ในหนึ่ง megabyte และ 1024 megabytes ในหนึ่ง gigabyte หากต้องการจัดสรรหน่วยความจำ 2 GB ให้กับ worker คุณต้องจัดสรร "mem": 2 * 1024 * 1024 * 1024

%%writefile --append ./source_files/transpile_remote.py

@distribute_task(target={
    "cpu": 16,
    "mem": 2 * 1024 * 1024 * 1024
})
def transpile_remote(circuit, optimization_level, backend):
    return None

Appending to ./source_files/transpile_remote.py

# This cell is hidden from users.
# It checks the distributed program works.
test_serverless_job(
    title="transpile_remote_serverless_test", entrypoint="transpile_remote.py"
)

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'QUEUED')

Waiting for job (status 'RUNNING')

Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')

Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')

Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')

Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')

Job completed successfully

จัดการข้อมูลในโปรแกรมของคุณ

Qiskit Serverless ช่วยให้คุณจัดการไฟล์ในไดเรกทอรี /data ใน program ทั้งหมดของคุณ ซึ่งมีข้อจำกัดหลายประการ:

• รองรับเฉพาะไฟล์ tar และ h5 เท่านั้น
• เป็นเพียงการจัดเก็บ /data แบบแบน ไม่สามารถมีไดเรกทอรีย่อย /data/folder/ ได้

ต่อไปนี้แสดงวิธีอัปโหลดไฟล์ ตรวจสอบให้แน่ใจว่าคุณได้ authenticate กับ Qiskit Serverless ด้วยบัญชี IBM Quantum ของคุณแล้ว (ดูคำแนะนำที่ Deploy to IBM Quantum Platform)

import tarfile
from qiskit_serverless import IBMServerlessClient

# Create a tar
filename = "transpile_demo.tar"
file = tarfile.open(filename, "w")
file.add("./source_files/transpile_remote.py")
file.close()

# Get a reference to a QiskitFunction
serverless = IBMServerlessClient()
transpile_remote_demo = next(
    program
    for program in serverless.list()
    if program.title == "transpile_remote_serverless"
)

# Upload the tar to Serverless data directory
serverless.file_upload(file=filename, function=transpile_remote_demo)

'{"message":"/usr/src/app/media/5e1f442128cdf60018496a04/transpile_demo.tar"}'

จากนั้น คุณสามารถแสดงรายการไฟล์ทั้งหมดในไดเรกทอรี data ของคุณ ข้อมูลนี้เข้าถึงได้จากทุก program

serverless.files(function=transpile_remote_demo)

['classifier_name.pkl.tar', 'output.json.tar', 'transpile_demo.tar']

ซึ่งสามารถทำได้จาก program โดยใช้ file_download() เพื่อดาวน์โหลดไฟล์ไปยัง program environment และคลาย tar

%%writefile ./source_files/extract_tarfile.py

import tarfile
from qiskit_serverless import IBMServerlessClient

serverless = IBMServerlessClient(token="<YOUR_API_KEY>") # Use the 44-character API_KEY you created and saved from the IBM Quantum Platform Home dashboard
files = serverless.files()
demo_file = files[0]
downloaded_tar = serverless.file_download(demo_file)

with tarfile.open(downloaded_tar, 'r') as tar:
    tar.extractall()

ณ จุดนี้ program ของคุณสามารถโต้ตอบกับไฟล์ได้เหมือนกับการทดลองในเครื่อง file_upload(), file_download() และ file_delete() สามารถเรียกใช้จากการทดลองในเครื่องหรือ program ที่อัปโหลดแล้ว เพื่อการจัดการข้อมูลที่สอดคล้องและยืดหยุ่น

ขั้นตอนถัดไป

คำแนะนำ

• ดูตัวอย่างเต็มรูปแบบที่ ย้ายโค้ดที่มีอยู่ไปยัง Qiskit Serverless
• อ่านบทความวิจัยที่นักวิจัยใช้ Qiskit Serverless และ quantum-centric supercomputing เพื่อสำรวจเคมีควอนตัม