关于使用多个信号量的同步

Question

您好，我正在处理一项关于使用多信号量的 POSIX 线程的任务。分配的简要说明是：有 4 个不同的数据包（char/video/audio/image），每个数据包由不同的线程承载，并且我们有一个共享缓冲区。系统上可以工作的最大线程数将由用户维护作为输入。例如;如果用户输入 10，则最多可以创建 10 个线程来在给定时间内通过缓冲区传输数据包。现在让我感到困惑的是，这个缓冲区可以立即包含有限的数据包。 （例如，它最多可以包含 10 个字符数据包和 20 个视频数据包等）所以我们必须为每种数据类型设置不同的信号量。这个问题我知道如何使用信号量控制缓冲区大小，这非常简单，但无法设置使用数据包信号量的正确想法。即使我尝试了一些不同的方法，我也总是面临死锁错误。这是我的伪代码，可以更清楚地了解我的程序。

define struct packege
define semaphore list

main

initialize variables and semaphores

while threadCounter is less than MaxThreadNumber

switch(random)
case 0: create a character package
    create a thread to insert the package in buffer
case 1: create a video package
    create a thread to insert the package in buffer
case 2: create an image package
    create a thread to insert the package in buffer
case 3: create an audio package
    create a thread to insert the package in buffer

increment  threadCounter by one
end of while

create only one thread which will make the dequeue operation
end of main

producer function

for i->0 to size_of_package
    sem_wait(empty_buffer) // decrement empty_buffer semaphore by size of package

    lock_mutex
        insert item into queueu
        decrement counter of the buffer by size of package
    unlock_mutex

for i->0 to size_of_package
    sem_post(full_buffer) // increment full_buffer semaphore by size of package

end of producer function

consumer function

while TRUE // Loops forever

    lock_mutex

        if queue is not empty
            dequeue

        increment counter of the buffer size of package

    unlock_mutex

for i->0 to size_of_package // The reason why i making the sem_wait operation here is i cant make the dequeue in outer region of mutex.
    sem_wait(full_buffer)
for i->0 to size_of_package
    sem_post(empty_buffer)
end of consumer function

使用此实现程序可以正常工作。但我无法正确使用属于包线程的信号量。我可以倾听每一个建议，每一个回答都将不胜感激。

Answer 1

这不是信号量的使用方式。缓冲区的控制变量/结构应该计算缓冲区中包含的消息数量和类型。互斥锁保护缓冲区及其控制变量/结构免受不同线程的并发访问。信号量（如果使用）只是向消费者发送缓冲区状态的信号，与数据包的大小无关；它当然不会随着数据包的大小而增加！

建议您最好使用 pthread 条件变量而不是信号量。这些与 pthread 互斥锁一起使用，以保证线程之间的无竞争信号。生产者循环这样做：

• 锁定互斥体，
• 修改缓冲区等以添加新数据包，
• 向条件变量发出信号，并且
• 解锁互斥锁。

消费者循环这样做：

• 锁定互斥体，
• 处理所有缓冲数据，
• 等待条件变量。

阅读pthread_cond_init、pthread_cond_signal和pthread_cond_wait。

Answer 2

由于这是一项作业，您可能不需要读取和写入真实的数据包数据，而只需模拟它们的处理方式。

在这种情况下，问题归结为当生产者线程达到他们可以写入缓冲区的数据包限制时如何有效地阻止它们。据我所知，目前，您正在使用信号量来计算写入缓冲区的数据包的各个元素。

假设您在缓冲区中的写入是原子的，并且您只想计算数据包，而不是数据包元素。每次生产者写入数据包时，它必须使用适当的信号量向消费者发出信号，而每次消费者读取数据包时，它必须向适当的生产者发出信号。

让我强调其他几点：

• 信号量的重要属性是当它达到零时它会阻塞。比如它的初始值为10，在连续10个sem_get之后，第11个就会阻塞。
• 您有 4 种类型的数据包，每种类型对可写入缓冲区的数量都有不同的阈值。

正如我所说，生产者必须发出信号，表示它写了一个数据包，但一旦达到阈值，它也必须停止。为了实现这一点，你让它在每次发布新数据包时获取信号量，sem_get。你让消费者在每次读取数据包时执行sem_post，这与你对单个信号量版本所做的相反。但是，由于您希望生产者在阈值处停止，因此您将容量初始化为N - 1 的信号量，N 是阈值。请注意，您必须在将新数据包写入缓冲区后发出信号表明它是可用的，否则消费者可能会阻塞缓冲区。

producer<type> function

  write_packet()  // put the packet in the buffer
  sem_wait(type)    // signal a new packet is available 
  // (if there's not enough space for another packet, the producer will block here) 

end producer<type> function

consumer function

  while TRUE // Loops forever

      switch packet_available() // look if there's a new packet available
       case video:
         read_packet<video>()
         sem_post(video)
       (...)
       default: // no packet available, just wait a little
          sleep()
      end if
  end while

您仍然需要定义packet_read、packet_write 和packet_available 函数，可能使用互斥锁来限制对缓冲区的访问。