Hacktoberfest 2026：メンテナが10月に向けて印を付けた、オープンで初心者向けの issue。 Hacktoberfest の issue を見る

Concurrent kv_put(tag=...) on one key loses tag updates

オープン
  • 何か問題がありますか？
#188 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
30/100
issue の種類
バグ
明瞭さ
明確に書かれている
活発さ
活発
技術スタック
python
領域
backend, data

調査の方向性

The race sits between the client-side merge in BatchMeta.update_custom_meta and the controller write in DataPartitionStatus.set_custom_meta, reached through kv_put and kv_batch_put. Start by running the threaded repro in the issue to confirm the lost tags, then trace how the controller's request loop applies custom_meta. Done means the repro reports no lost tags and the sequential case still passes; the fix itself is a design choice the maintainers should weigh in on.

索引モデルが issue の本文から書いたものです。

説明

Problem

Concurrent tag updates on the same key lose updates. Two writers that set different tag names on one key can leave it with only one of them.

kv_put(key, partition_id, tag={...}) (and kv_batch_put(..., tags=[...]), plus the async versions) is a read-modify-write split between client and controller:

  1. The client fetches the key's metadata, including its current tags.
  2. It merges the new tag into that copy locally (BatchMeta.update_custom_meta).
  3. It sends the whole merged dict back, and the controller stores it with custom_meta.update(...) (DataPartitionStatus.set_custom_meta), which replaces each sample's dict.

When two writers interleave between steps 1 and 3, the second one writes a dict that was read before the first one's tag landed, so the first tag is dropped.

Repro (on main)

import threading
import torch
import transfer_queue as tq

tq.init()
P = "tag_race"
keys = [f"k{i}" for i in range(300)]
for key in keys:
    tq.kv_put(key, P, fields={"x": torch.tensor([0])})
barriers = [threading.Barrier(2) for _ in keys]

def writer(tag_name):
    for key, barrier in zip(keys, barriers):
        barrier.wait()
        tq.kv_put(key, P, tag={tag_name: 1})

threads = [threading.Thread(target=writer, args=(name,)) for name in ("a", "b")]
for t in threads:
    t.start()
for t in threads:
    t.join()
tags = tq.kv_list(P)[P]
print(sum(not {"a", "b"} <= set(tags[k]) for k in keys))  # 300: every key lost one tag

Run sequentially, both tags survive.

Why it matters

Retry or bookkeeping logic that keeps state in tags (for example an attempt id written by one worker while another writes a score) silently loses one of the writes. Callers can work around it today by holding a lock around the update (the KV locks in #186), but a tag-only update should be safe on its own.

Suggested fix

Merge per key in the controller instead of on the client: send only the new tag entries, and have the controller apply them to the sample's existing dict (custom_meta.setdefault(idx, {}).update(new_tags)) in its single request loop. That makes each tag update atomic without any lock.

Found while reviewing #186.

主要言語
Python
スター
168
フォーク
56
平均マージ
4日 1時間
マージ済み PR（30日）
15

環境構築

このプロジェクトには開発コンテナ、Dockerfile、コントリビューションガイドがありません。まず README を読み、一般的な手順ははじめてのコントリビューションガイドを参照してください。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

Ascend/TransferQueue のほかの issue

Ascend/TransferQueue の issue をすべて見る

似ている issue

Python の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。