Bộ lọc có an toàn không thread

Question

Tôi có một chủ đề đang cập nhật danh sách có tên là l. Tôi có đúng trong việc nói rằng đó là thread-an toàn để làm như sau từ thread khác?

filter(lambda x: x[0] == "in", l) 

Nếu nó không đề an toàn, là điều này thì cách tiếp cận đúng:

import threading 
import time 
import Queue 

class Logger(threading.Thread): 
    def __init__(self, log): 
     super(Logger, self).__init__() 
     self.log = log 
     self.data = [] 
     self.finished = False 
     self.data_lock = threading.Lock() 

    def run(self): 
     while not self.finished: 
      try: 
       with self.data_lock: 
        self.data.append(self.log.get(block=True, timeout=0.1)) 
      except Queue.Empty: 
       pass 

    def get_data(self, cond): 
     with self.data_lock: 
      d = filter(cond, self.data)  
     return d 

    def stop(self): 
     self.finished = True 
     self.join() 
     print("Logger stopped") 

nơi phương pháp get_data(self, cond) được sử dụng để lấy một tập hợp con nhỏ của dữ liệu trong self.data trong một chủ đề an toàn cách thức.

Answer 1

Trước tiên, để trả lời câu hỏi của bạn trong tiêu đề: filter chỉ là một chức năng. Do đó, an toàn luồng của nó sẽ dựa vào cấu trúc dữ liệu mà bạn sử dụng nó.

Như đã nêu trong các nhận xét, danh sách hoạt động chính là an toàn luồng trong CPython và được bảo vệ bởi GIL, nhưng đó chỉ là chi tiết thực hiện của CPython mà bạn không nên dựa vào. Ngay cả khi bạn có thể dựa vào nó, an toàn luồng của một số thao tác của chúng có thể không có nghĩa là loại an toàn chủ đề bạn có nghĩa là:

Vấn đề là việc lặp lại chuỗi với filter nói chung không phải là hoạt động nguyên tử. Trình tự có thể được thay đổi trong khi lặp lại. Tùy thuộc vào cấu trúc dữ liệu bên dưới trình vòng lặp của bạn, điều này có thể gây ra nhiều hiệu ứng lạ hoặc ít. Một cách để khắc phục vấn đề này là bằng cách lặp qua một bản sao của chuỗi được tạo ra với một hành động nguyên tử. Cách dễ nhất để làm điều này cho chuỗi tiêu chuẩn như tuple, list, string là với các nhà điều hành lát như thế này:

filter(lambda x: x[0] == "in", l[:]) 

Ngoài này không nhất thiết phải là thread-an toàn cho dữ liệu các loại khác, có một vấn đề với mặc dù điều này : nó chỉ là một bản sao nông. Vì các phần tử của danh sách của bạn dường như giống như danh sách, một luồng khác có thể song song làm del l[1000][:] để trống một trong các danh sách bên trong (cũng được chỉ ra trong bản sao nông của bạn). Điều này sẽ làm cho biểu thức lọc của bạn không thành công với một số IndexError.

Tất cả những gì đã nói, không phải là một sự xấu hổ khi sử dụng khóa để bảo vệ quyền truy cập vào danh sách của bạn và tôi chắc chắn sẽ giới thiệu nó. Tùy thuộc vào cách dữ liệu của bạn thay đổi và cách bạn làm việc với dữ liệu được trả lại, nó thậm chí có thể là khôn ngoan để sao chép sâu các phần tử trong khi giữ khóa và trả lại các bản sao đó. Bằng cách đó bạn có thể đảm bảo rằng một khi trở lại điều kiện bộ lọc sẽ không đột ngột thay đổi cho các phần tử trả về.

Wrt. mã số Logger của bạn: Tôi không chắc chắn 100% cách bạn dự định sử dụng và nếu điều đó quan trọng đối với bạn để chạy một số chuỗi trên một hàng đợi và join chúng. Điều kỳ lạ đối với tôi là bạn không bao giờ sử dụng Queue.task_done() (giả sử rằng self.log của nó là Queue). Ngoài ra bỏ phiếu của bạn trong hàng đợi có khả năng lãng phí. Nếu bạn không cần join của thread, tôi muốn đề nghị ít nhất là biến việc mua lại khóa xung quanh:

class Logger(threading.Thread): 
    def __init__(self, log): 
     super(Logger, self).__init__() 
     self.daemon = True 
     self.log = log 
     self.data = [] 
     self.data_lock = threading.Lock() 

    def run(self): 
     while True: 
      l = self.log.get() # thread will sleep here indefinitely 
      with self.data_lock: 
       self.data.append(l) 
      self.log.task_done() 

    def get_data(self, cond): 
     with self.data_lock: 
      d = filter(cond, self.data) 
      # maybe deepcopy d here 
     return d 

Bên ngoài bạn vẫn có thể làm log.join() để đảm bảo rằng tất cả các yếu tố của log hàng đợi được xử lý.

Answer 2

Nếu một chủ đề ghi vào danh sách và một chuỗi khác đọc danh sách đó, thì cả hai phải được đồng bộ hóa. Nó không quan trọng cho khía cạnh đó cho dù người đọc sử dụng filter(), một chỉ mục hoặc lặp lại hoặc liệu người viết có sử dụng append() hoặc bất kỳ phương pháp nào khác không.

Trong mã của bạn, bạn đạt được đồng bộ hóa cần thiết bằng cách sử dụng threading.Lock. Vì bạn chỉ truy cập danh sách trong ngữ cảnh của with self.data_lock, các quyền truy cập sẽ loại trừ lẫn nhau.

Tóm lại, mã của bạn chính xác là chính xác liên quan đến việc xử lý danh sách giữa các chuỗi. Nhưng:

• Bạn truy cập self.finished mà không cần khóa, có vấn đề. Việc chỉ định cho thành viên đó sẽ thay đổi self, tức là ánh xạ đối tượng cho các thành viên theo dõi, vì vậy việc này sẽ được đồng bộ hóa. Có hiệu quả, điều này sẽ không bị tổn thương, bởi vì True và False là hằng số toàn cầu, ở mức tồi tệ nhất, bạn sẽ có một sự chậm trễ ngắn giữa cài đặt trạng thái trong một chuỗi và xem trạng thái ở trạng thái khác. Nó vẫn còn xấu, bởi vì nó là hình thành thói quen.
• Theo quy tắc, khi bạn sử dụng khóa, luôn là tài liệu đối tượng khóa này bảo vệ. Ngoài ra, tài liệu mà đối tượng được truy cập theo chủ đề nào. Thực tế là self.finished được chia sẻ và yêu cầu đồng bộ hóa sẽ rõ ràng. Ngoài ra, tạo sự khác biệt trực quan giữa các chức năng công cộng và dữ liệu cũng như dữ liệu riêng tư (bắt đầu bằng số _underscore, xem PEP 8) giúp theo dõi điều này. Nó cũng giúp người đọc khác.
• Một vấn đề tương tự là baseclass của bạn. Nói chung, kế thừa từ threading.Thread là một ý tưởng tồi. Thay vào đó, bao gồm một thể hiện của lớp chuỗi và cung cấp cho nó một hàm như self._main_loop để chạy. Lý do là bạn nói rằng Logger của bạn là một Thread và rằng tất cả các thành viên công cộng của bảng xếp hạng cũng là thành viên công khai trong lớp của bạn, có lẽ là giao diện rộng hơn nhiều so với những gì bạn dự định.
• Bạn không bao giờ nên khóa bằng khóa được giữ. Trong mã của bạn, bạn chặn trong self.log.get(block=True, timeout=0.1) bằng khóa trên mutex. Trong thời gian đó, ngay cả khi không có gì thực sự xảy ra, không có chủ đề nào khác có thể gọi và hoàn thành một cuộc gọi đến get_data(). Thực sự chỉ là một cửa sổ nhỏ giữa việc mở khóa mutex và khóa lại nó khi người gọi get_data() không không phải đợi, điều này rất tồi tệ cho hiệu suất. Tôi thậm chí có thể tưởng tượng rằng câu hỏi của bạn được thúc đẩy bởi hiệu suất thực sự xấu này gây ra. Thay vào đó, hãy gọi log.get(..) mà không cần khóa, không cần thiết. Sau đó, với khóa được giữ, hãy thêm dữ liệu vào self.data và kiểm tra self.finished.