4

私は現在、スレッドとそのすべてを使用して、Python ベースのデータグラム サーバーをプログラミングしています。

次の問題が発生しました。複数の割り当てスレッドを使用して、受信パッケージを異なる処理スレッドに割り当てています。Processing Threads 内では、threading.local() を使用してスレッドローカル変数を追跡しています。

現在、サーバーが高負荷 (約 2 秒で 2000 パケット) のときにどのように反応するかをテストしており、local()-Object の奇妙な動作に遭遇しました。

しばらくは問題なく動作しているように見えますが、ある時点で例外がスローされます。

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、処理スレッドのメイン関数に到達することはありませんが、しばらくの間はすべて正常に動作しているように見えます。

Web と StackOverflow で同様の問題を検索しましたが、何も見つかりませんでした。事前に助けてくれてありがとう、そして私のコーディングスタイルを許してください、私は数日間Pythonをプログラミングしているだけで、まだその機能のいくつかを使ってロープを学んでいます.

編集: もちろん、このサーバーにはこれ以上のものがあります。パケットを受信して​​割り当てスレッドに送信するメイン スレッドと、バックグラウンドでのその他の処理が多数あります。さらに、サーバーはまだ完成していませんが、他のことを始める前に受信を機能させたいと思っています。

4

1 に答える 1

7

threading.localスレッドローカル データを表すクラスです。スレッド ローカル データは、値がスレッド固有であり、ローカルスレッド (local()オブジェクトを作成したスレッド) でのみ使用できるデータです。つまり、値を設定したスレッドだけが値を認識します。

loc.parent処理スレッドのメソッドで値を設定します__init__が、アロケーター スレッド内で実行されるため、処理スレッドのスレッド ローカルはアロケーター スレッドでのみ使用できます。そのためrun()、別のスレッド (アロケーターではなく) によって実行されるため、処理スレッドでは機能しません。

于 2012-02-25T18:33:07.620 に答える