Memproses DAG dengan async Python dan graphlib

Baru-baru ini saya menjumpai modul yang menarik dalam pustaka standard tanpa dasar Python: graphlib. Jika anda tidak pernah menggunakannya sebelum ini, ia adalah utiliti kecil yang telah ditambahkan dalam Python 3.9 dan hanya melaksanakan satu kelas: TopologicalSorter.

Namanya jelas -- ini ialah kelas untuk pengisihan topologi graf. Tetapi saya tidak fikir ia pada asalnya ditulis dengan hanya menyusun dalam fikiran, kerana ia mempunyai API yang agak samar, tetapi sangat berguna, seperti kaedah prepare() atau is_active(). Contoh dalam dokumentasi ini membayangkan motivasi di sebaliknya:

topological_sorter = TopologicalSorter()

# Add nodes to 'topological_sorter'...

topological_sorter.prepare()
while topological_sorter.is_active():
    for node in topological_sorter.get_ready():
        # Worker threads or processes take nodes to work on off the
        # 'task_queue' queue.
        task_queue.put(node)

    # When the work for a node is done, workers put the node in
    # 'finalized_tasks_queue' so we can get more nodes to work on.
    # The definition of 'is_active()' guarantees that, at this point, at
    # least one node has been placed on 'task_queue' that hasn't yet
    # been passed to 'done()', so this blocking 'get()' must (eventually)
    # succeed.  After calling 'done()', we loop back to call 'get_ready()'
    # again, so put newly freed nodes on 'task_queue' as soon as
    # logically possible.
    node = finalized_tasks_queue.get()
    topological_sorter.done(node)

Jadi graphlib bukan modul hanya untuk mengisih graf, ia juga utiliti untuk menjalankan graf tugas dalam susunan topologi, yang berguna jika beban kerja anda mempunyai tugas bergantung pada hasil tugasan lain. Graf ialah cara terbaik untuk memodelkan masalah ini dan susunan topologi ialah cara anda memastikan tugasan diproses dalam susunan yang betul.

Satu perkara yang hilang daripada dokumen ialah contoh asyncio, yang ternyata agak mudah untuk ditulis. Memandangkan dengan asyncio anda tidak perlu berurusan dengan keselamatan benang, anda boleh bertahan tanpa menggunakan baris gilir untuk menyegerakkan utas atau sebarang kerumitan tambahan lain seperti itu.

Kami akan mentakrifkan fungsi pelawat nod async ringkas:

async def visit(node: str, sorter: TopologicalSorter):
    print(f"processing node {node}")
    sorter.done(node)

Dalam dunia nyata ini boleh menjadi sekompleks yang anda mahukan, selagi anda melakukan kerja terikat I/O jadi nikmatilah faedah asyncio. Perkara yang penting ialah memanggil sorter.done(nod) di penghujung fungsi untuk memberitahu contoh TopologicalSorter bahawa kita telah selesai dengan nod ini dan kita boleh maju ke seterusnya.

Kemudian kami pasangkan fungsi lawatan ke dalam larian tersusun topologi kami:

sorter = TopologicalSorter(graph)

sorter.prepare()

while sorter.is_active():
    node_group = sorter.get_ready()

    if not node_group:
        # no nodes are ready yet, so we sleep for a bit
        await asyncio.sleep(0.25)
    else:
        tasks = set()
        for node in node_group:
            task = asyncio.create_task(visit(node, sorter))
            tasks.add(task)
            task.add_done_callback(tasks.discard)

Kod sumber penuh skrip yang berfungsi boleh didapati dalam intipati ini.

Satu aspek pelik graphlib ialah format graf yang diterima oleh TopologicalSorter sebagai hujah -- ia berada dalam susunan terbalik daripada perwakilan graf biasa anda. Cth. jika anda mempunyai graf seperti ini A -> B -> C, biasanya anda akan mewakilinya seperti ini:

graph = {
  "A": ["B"],
  "B": ["C"],
  "C": [],
}

tetapi TopologicalSorter mahu graf ini dalam arah tepi diterbalikkan:

Jika hujah graf pilihan disediakan, ia mestilah kamus yang mewakili graf akiklik terarah dengan kekuncinya ialah nod dan nilainya boleh diulang daripada semua pendahulu nod tersebut dalam graf

Jadi cara yang betul untuk mewakili A -> B -> C untuk TopologicalSorter ialah ini:

graph = {
  "C": ["B"],
  "B": ["A"],
  "A": [],
}

Maklumat lanjut dan perdebatan yang agak hangat tentang perkara ini boleh didapati di sini: https://bugs.python.org/issue46071.

Selamat mengekod!

Atas ialah kandungan terperinci Memproses DAG dengan async Python dan graphlib. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!