在Python中使用条件和事件在两个线程之间交替进行多线程处理

Question

我试图编写一个程序，用它在两个线程之间交替使用，thread1和thread2。棘手的是，首先开始执行的线程必须是thread1。 这是我目前掌握的代码：

Class Client:
#member variables
def sendFile(self,cv1,lock1):

        sent=0;
        while (i<self.size):
            message = self.data[i:1024+i]
            cv1.acquire()
            BadNet.transmit(self.clientSocket,message,self.serverIP,self.serverPort)
            cv1.notify() 
            cv1.release()

            i = i+1024
            sent+=1
            lock1.wait()

        print "File sent successfully !"   
        self.clientSocket.close()

    def receiveAck(self,cv1,lock2):
        i=0
        while (1):
            lock1.clear()
            cv1.acquire()
            cv1.wait()
            print "\nentered ack !\n"
            self.ack, serverAddress = self.clientSocket.recvfrom(self.buf)

            cv1.release()
            lock1.set()


if __name__ == "__main__":
    lock1 = Event()
    cv1 = Condition()
    cv2= Condition()
    client = Client();
    client.readFile();

    thread1 = Thread(target = client.sendFile, args=[cv1,lock1])
    thread2 = Thread(target = client.receiveAck, args=[cv1,lock1])

    thread1.start()
    thread2.start()
    thread1.join()
    thread2.join()

我当前面临的问题是，最初程序在两个线程之间交替（由控制台上的输出确认）。但是经过任意次数的迭代（通常在20到80次之间），程序就会挂起，不再执行进一步的迭代。在

Answer 1

同步至少有两个问题。在

首先，你用错了cv1。您的接收线程必须在其cv周围循环，检查条件并每次调用wait。否则，您只需将cv用作断开事件+锁定组合。你没有这样的循环。更重要的是，你甚至没有等待的条件。在

第二，你用错了lock1。接收线程设置事件，然后立即将其清除。但不能保证发送线程已经进入等待状态。（上一个问题的竞争使这个问题变得更严重，但即使你解决了它，它仍然是一个问题。）在多核机器上，它通常会及时到达那里，但在线程编程中“通常”甚至比从来没有更糟糕。因此，最终发送线程将在接收线程完成清除后到达等待，因此它将永远等待。同时，接收线程将等待发送线程的通知，这是永远不会发生的。所以你陷入僵局了。在

为了将来的参考，在每个阻塞操作（尤其是sync操作）前后添加print语句将使调试变得非常困难：您将看到接收线程的最后一条消息是“receive waiting on cv1”，而send线程的最后一条消息是“send waiting on lock1”，并且死锁在哪里很明显。在

---

不管怎样，我甚至不确定“修复”一个没有条件的简历，或者一个你试图作为简历使用的事件意味着什么，所以我将展示如何用两个简历写一些合理的东西。在这种情况下，我们可以使用一个标志，我们来回翻转作为两个CV的条件。在

在此期间，我将修复一些使代码不可测试的其他问题（例如，i从未初始化），并包括调试信息，以及为使此成为完整示例而必须填写的内容，但否则，我将尝试保留您的结构和不相关的问题（如Client是一个旧样式的类）。在

class Client:
    def __init__(self):
        self.clientSocket = socket(AF_INET, SOCK_DGRAM)
        self.serverIP = '127.0.0.1'
        self.serverPort = 11111
        self.buf = 4
        self.waitack = False

    def readFile(self):
        self.data = ', '.join(map(str, range(100000)))
        self.size = len(self.data)

    #member variables
    def sendFile(self,cv1,lock1):
        i = 0
        sent=0
        while (i<self.size):
            message = self.data[i:1024+i]
            print "s cv1 acquire"
            with cv1:
                print "s sendto"
                self.clientSocket.sendto(message, (self.serverIP, self.serverPort))
                self.waitack = True
                print "s cv1 notify"
                cv1.notify() 

            i = i+1024
            sent+=1

            print "s cv2 acquire"
            with cv2:
                print "s cv2 wait"
                while self.waitack:
                    cv2.wait()

        print "File sent successfully !"   
        self.clientSocket.close()

    def receiveAck(self,cv1,lock2):
        i=0
        while (1):
            print "r cv1 acquire"
            with cv1:
                while not self.waitack:
                    print "r cv1 wait"
                    cv1.wait()
            print "r recvfrom"
            self.ack, serverAddress = self.clientSocket.recvfrom(self.buf)
            i += 1
            print self.ack, i            

            print "r cv2 acquire"
            with cv2:
                self.waitack = False
                print "r cv2 notify"
                cv2.notify()

这里有一个测试服务器：

^{pr2}$

启动服务器，启动客户端，服务器将计数为672，客户端将计数为673（因为您的代码以1为基础），其中673对平衡消息和一个“文件发送成功！”最后。（当然，客户端将永远挂起，因为receiveAck无法完成，而服务器则是因为我将其作为一个无限循环编写的。）