Python: 'thread._local object не имеет атрибута' todo '

Question

Я в настоящее время программирую на основе paton datagram-Server, используя потоки и все такое.

У меня возникла следующая проблема: я использую несколько потоков распределения для распределения входящих пакетов на разные потоки обработки. Внутри потоков обработки я использую threading.local() для отслеживания локальных переменных потока.

В настоящее время я тестирую, как мой сервер реагирует во время высокой нагрузки (2000 пакетов в ~ 2 секунды), и натолкнулся на странное поведение локального объекта().

Похоже, что это прекрасно работает на некоторое время, а затем, в какой-то момент, он бросает исключение:

Exception in thread 192.168.1.102: # <-- This is the Processing Thread 
Traceback (most recent call last): 
    File "/opt/lib/python2.7/threading.py", line 552, in __bootstrap_inner 
    self.run() 
    File "/volume1/home/Max/Python/MyThread/pyProcThread.py", line 37, in run 
    print self.loc.father 
AttributeError: 'thread._local' object has no attribute 'father' 

# The following three lines are debug 
# This is the Allocation thread that has called the Processing thread 
<Thread(192.168.1.102, started 1106023568)> 
# This confirms that the queue it tries to access exists 
<Queue.Queue instance at 0x40662b48> 
# This is the Processing thread, which has stopped executing on exception 
<Thread(192.168.1.102, stopped 1106023568)> 

Exception in thread pyAlloc-0: # <-- This is the Allocation thread 
Traceback (most recent call last): 
    File "/opt/lib/python2.7/threading.py", line 552, in __bootstrap_inner 
    self.run() 
    File "/volume1/home/Max/Python/MyThread/pyAllocThread.py", line 60, in run 
    foundThread.addTask(str(data)) 
    File "/volume1/home/Max/Python/MyThread/pyProcThread.py", line 58, in addTask 
    print self.loc.todo 
AttributeError: 'thread._local' object has no attribute 'todo' 

Часть резьбообрабатывающие

# Imports 
import threading 
import time 
import Queue 

class Thread(threading.Thread): 
    def __init__(self,pThread): 
     threading.Thread.__init__(self) 
     self.loc = threading.local() 
     self.loc.todo = Queue.Queue() 
     self.loc.father = pThread 
     print "Konstruktor ausgefuehrt" 
     print self.loc.todo 

    def run(self): 
     self.loc.run = True; 
     print self.loc.father 
     while (self.loc.run): 
     try: 
      task = self.loc.todo.get(True, 1) 
      print "processing..." 
      if(task == self): 
       self.loc.run=False 
      else: 
       print task 
     except Queue.Empty: 
      pass 
     self.loc.father.threadTerminated(self) 
     print self.name, "terminating..." 

    def addTask(self, pTask): 
     print self 
     print self.loc.todo 
     self.loc.todo.put(pTask) 

и выделение Тема:

import threading 
import pyProcThread # My processing Thread 
import Queue 
import time 
class Thread(threading.Thread): 
    # Lock-Objects 
    threadListLock = threading.Lock() 
    waitListLock = threading.Lock() 

    alive = True; 

    taskQueue = Queue.Queue() 
    # Lists 
    # List of all running threads 
    threads = [] 

    def threadExists(self,pIP): 
     """Checks if there is already a thread with the given Name""" 
     for x in self.threads: 
      if x.name == pIP: 
       return x 
     return None 

    def threadTerminated(self,pThread): 
     """Called when a Processing Thread terminates""" 
     with self.threadListLock: 
      self.threads.remove(pThread) 
      print "Thread removed" 

    def threadRegistered(self,pThread): 
     """Registers a new Thread""" 
     self.threads.append(pThread) 

    def killThread(self): 
     self.alive = False 

    def run(self): 
     while(self.alive): 
      # print "Verarbeite Nachricht ", self.Message 
      # print "Von ", self.IP 
      try: 
       data, addtemp = self.taskQueue.get(True, 1) 
       addr, _junk = addtemp 
       with self.threadListLock: 
        foundThread=self.threadExists(str(addr)) 
        # print "Thread " + self.name + " verarbeitet " + data 
        if (foundThread!=None): 
         #print "recycling thread" 
         foundThread.addTask(str(data)) 
        else: 
         print "running new Thread" 
         t = pyProcThread.Thread(self) 
         t.name = str(addr) 
         t.addTask(str(data)) 
         t.start() 
         self.threadRegistered(t) 
       self.taskQueue.task_done() 
      except Queue.Empty: 
       pass 
     print self.name, "terminating..." 
     with self.threadListLock: 
      for thread in self.threads: 
       thread.addTask(thread) 

полный выход, в том числе все отладки принт:

running new Thread 
Konstruktor ausgefuehrt 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, initial)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
Exception in thread 192.168.1.102: 
Traceback (most recent call last): 
    File "/opt/lib/python2.7/threading.py", line 552, in __bootstrap_inner 
    self.run() 
    File "/volume1/home/Max/Python/MyThread/pyProcThread.py", line 37, in run 
    print self.loc.father 
AttributeError: 'thread._local' object has no attribute 'father' 

<Thread(192.168.1.102, started 1106023568)> 
<Queue.Queue instance at 0x40662b48> 
<Thread(192.168.1.102, stopped 1106023568)> 
Exception in thread pyAlloc-0: 
Traceback (most recent call last): 
    File "/opt/lib/python2.7/threading.py", line 552, in __bootstrap_inner 
    self.run() 
    File "/volume1/home/Max/Python/MyThread/pyAllocThread.py", line 60, in run 
    foundThread.addTask(str(data)) 
    File "/volume1/home/Max/Python/MyThread/pyProcThread.py", line 58, in addTask 
    print self.loc.todo 
AttributeError: 'thread._local' object has no attribute 'todo' 

Terminating main 
192.168.1.102 
DEINEMUDDA sent to 192.168.1.102 : 8082 
pyAlloc-1 terminating... 
<Thread(192.168.1.102, stopped 1106023568)> 
<Queue.Queue instance at 0x40662b48> 

Как вы можете видеть в журнале, все, кажется, работает хорошо на некоторое время, хотя он никогда не достигает print "processing в основной функции обработки потока.

Я искал в Интернете и StackOverflow для подобных проблем, но не смог найти. Заранее спасибо за любую помощь, и извините мой стиль кодирования, я только программирую Python в течение нескольких дней и все еще изучая канаты с некоторыми из его функций.

EDIT: Конечно, для этого сервера больше. Существует основной поток, который принимает пакеты и отправляет их в поток распределения, а также пучок других вещей в фоновом режиме. Кроме того, сервер еще далек от завершения, но я хочу, чтобы работа получала, прежде чем я начну с других вещей.

Answer 1

threading.local - класс, который представляет данные на основе потока. Локальными данными по потоку являются данные, значения которых являются конкретными потоками и, следовательно, доступны для local thread only (поток, который создал объект local()). Другими словами, только поток, который устанавливает значение, видит значение.

Поскольку вы установили значение loc.parent в методе обработки потока, но выполняемое внутри нити распределителя, локаторы потоков потоков процесса будут доступны только в потоке распределителя. Таким образом, он не будет работать в потоке обработки run(), поскольку он будет выполняться разными потоками (а не распределителем).