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

Đang mở
  • Có gì không ổn?
#188 0 bình luận 0 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
30/100
Loại issue
Lỗi
Độ rõ ràng
Đặc tả rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
python
Lĩnh vực
backend, data

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
Python
Star
169
Fork
56
Merge trung bình
1 ngày 17 giờ
Pull request đã merge (30 ngày)
15

Chuẩn bị môi trường

Dự án này không cung cấp dev container, Dockerfile hay hướng dẫn đóng góp, nên bạn cần tự thiết lập môi trường: hãy bắt đầu từ README và xem hướng dẫn đóng góp lần đầu của chúng tôi để biết các bước chung.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của Ascend/TransferQueue

Tất cả issue của Ascend/TransferQueue

Issue tương tự

Thêm issue về Python

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.