在单写入器多读取器线程中交换缓冲区

Question

故事

有一个编写线程，定期从某个地方收集数据（实时，但这在问题中并不重要）。有很多读者正在阅读这些数据。通常的解决方案是使用两个读写器锁和两个缓冲区，如下所示：

Writer (case 1):
acquire lock 0                        
loop
    write to current buffer
    acquire other lock
    free this lock
    swap buffers
    wait for next period

或者

Writer (case 2):
acquire lock 0                        
loop
    acquire other lock
    free this lock
    swap buffers
    write to current buffer
    wait for next period

问题

在这两种方法中，如果获取其他锁操作失败，则不会完成交换，并且写入器将覆盖其先前的内容数据（因为写入器是实时的，它不能等待读取器）所以在这种情况下，所有读取器都会丢失该数据帧。

但这并不是什么大问题，读者是我自己的代码，而且它们很短，所以使用双缓冲区，这个问题就解决了，如果有问题，我可以将其设置为三缓冲区（或更多）。

问题是我想最小化延迟。想象一下情况 1：

writer writes to buffer0                reader is reading buffer1
writer can't acquire lock1              because reader is still reading buffer1
|                                       |
|                                       reader finishes reading,
| (writer waiting for next period)      <- **this point**
|
|
writer wakes up, and again writes to buffer0

在**此时**，如果只有写入器可以在读取器完成后进行交换而不是等待下一个周期，那么理论上其他读取器可以读取 buffer0 的数据。本例中发生的情况是，仅仅因为一个读者迟到了一点，所有读者就错过了一帧数据，而这个问题本来是可以完全避免的。

情况 2 类似：

writer writes to buffer0                reader is idle
|                                       |
|                                       reader finishes reading,
| (writer waiting for next period)
|
|                                       reader starts reading buffer1
writer wakes up                         |
it can't acquire lock1                  because reader is still reading buffer1
overwrites buffer0

我尝试混合解决方案，因此编写者在写入后立即尝试交换缓冲区，如果不可能，则在下一个周期醒来后立即尝试。所以是这样的：

Writer (case 3):
acquire lock 0                        
loop
    if last buffer swap failed
        acquire other lock
        free this lock
        swap buffers
    write to current buffer
    acquire other lock
    free this lock
    swap buffers
    wait for next period

现在延迟的问题仍然存在：

writer writes to buffer0                reader is reading buffer1
writer can't acquire lock1              because reader is still reading buffer1
|                                       |
|                                       reader finishes reading,
| (writer waiting for next period)      <- **this point**
|
|
writer wakes up
swaps buffers
writes to buffer1

再次在**此时**，所有读者都可以开始读取 buffer0，这是在 buffer0 之后的短暂延迟> 已经写好了，但是还要等到作者的下一期。

问题 问题

是，我该如何处理这个问题？如果我希望编写器在所需的周期精确执行，则需要使用 RTAI 函数等待该周期，而我不能像这样执行，

Writer (case 4):
acquire lock 0                        
loop
    write to current buffer
    loop a few times or until the buffer has been swapped
        sleep a little
        acquire other lock
        free this lock
        swap buffers
    wait for next period

这会引入抖动。因为“几次”可能会比“等待下一个周期”更长，因此作者可能会错过该周期的开始。

为了更清楚起见，这就是我想要发生的事情：

writer writes to buffer0                reader is reading buffer1
|                                       |
|                                       reader finishes reading,
| (writer waiting for next period)      As soon as all readers finish reading,
|                                         the buffer is swapped
|                                       readers start reading buffer0
writer wakes up                         |
writes to buffer1

我发现了什么

我已经发现阅读-copy-update 据我所知，它会不断为缓冲区分配内存并释放它们，直到读者使用完它们为止，这对我来说是不可能的，原因有很多。第一，线程在内核空间和用户空间之间共享。其次，使用 RTAI，您无法在实时线程中分配内存（因为这样您的线程将调用 Linux 的系统调用，从而破坏实时性！（更不用说使用 Linux 自己的 RCU 实现是无用的，因为出于同样的原因）

我还考虑过有一个额外的线程以更高的频率尝试交换缓冲区，但这听起来不是一个好主意，首先，它本身需要与编写器同步，其次，嗯。我有很多这样的编写器-读取器在不同部分并行工作，并且为每个编写器提供一个额外的线程，对于所有编写器来说一个线程似乎对于与每个编写器的同步来说非常复杂。

Answer 1

您使用什么 API 来实现读写锁？您是否有定时锁，例如 pthread_rwlock_timedwrlock ？如果是，我认为这是解决您的问题的方法，就像下面的代码一样：

void *buf[2];

void
writer ()
{
  int lock = 0, next = 1;

  write_lock (lock);
  while (1)
    {
      abs_time tm = now() + period;

      fill (buf [lock]);
      if (timed_write_lock (next, tm))
        {
          unlock (lock);
          lock = next;
          next = (next + 1) & 1;
        }
      wait_period (tm);
    }
}


void
reader ()
{
  int lock = 0;
  while (1)
    {
      reade_lock (lock);
      process (buf [lock]);
      unlock (lock);
      lock = (lock + 1) & 1;
    }
}

这里发生的情况是，对于编写者来说，等待锁定还是等待下一个周期并不重要，只要它是一定要在下一个经期到来之前醒来。绝对超时可以确保这一点。

Answer 2

这不正是三重缓冲要解决的问题吗？所以你有 3 个缓冲区，我们称它们为 write1、write2 和 read。写入线程在写入 write1 和 write2 之间交替，确保它们永远不会阻塞，并且最后一个完整帧始终可用。然后，在读取线程中，在某个适当的点（例如，在读取帧之前或之后），读取缓冲区将与可用的写入缓冲区进行翻转。

虽然这将确保写入者永远不会阻塞（只需翻转两个指针就可以非常快速地完成缓冲区翻转，甚至可能使用 CAS 原子而不是锁），但仍然存在读取器必须等待其他读取器完成的问题与翻转之前的读缓冲区。我想这可以通过一个读缓冲区池来稍微解决 RCU 式的问题，其中可以翻转可用的缓冲区。

Answer 3

• 使用队列（先进先出链表）
• 实时写入器总是追加（入队）到队列末尾
• 读取器总是从队列开头移除（出队）
• 如果队列为空，读取器将阻塞

编辑以避免动态分配

我可能会使用循环队列......
我会使用内置的 __sync 原子操作。
http://gcc.gnu.org /onlinedocs/gcc-4.1.0/gcc/Atomic-Builtins.html#Atomic-Builtins

• 循环队列（FIFO 二维数组）
• • 例如：字节[][]数组=新字节[MAX_SIZE][BUFFER_SIZE];
• • 开始和结束索引指针
• Writer 覆盖 Array[End][] 处的缓冲区
• • 如果 Start 完全循环结束，Writer 可以递增 Start
• 如果Reader 从 Array[Start][] 获取缓冲区
• • 如果 Start == End，则读取器会阻塞

Answer 4

如果您不希望写入者等待，也许它不应该获取其他人可能持有的锁。不过，我会让它执行某种同步，以确保它写入的内容确实被写出 - 通常，大多数同步调用将导致执行内存刷新或屏障指令，但详细信息将取决于内存模型你的CPU和你的线程包的实现。

我会看看是否有任何其他同步原语更适合事情，但如果推送到了紧要关头，我会让编写器锁定并解锁其他人从未使用过的锁。

然后，读者必须做好偶尔会错过一些内容的准备，并且必须能够发现自己何时错过了一些内容。我会将有效性标志和长序列计数与每个缓冲区相关联，并让编写者执行类似“清除有效性标志、递增序列计数、同步、写入缓冲区、递增序列计数、设置有效性标志、同步”之类的操作。如果读取器读取序列计数、同步、看到有效性标志为真、读出数据、同步并重新读取相同的序列计数，那么也许有希望它没有得到乱码数据。

如果您要这样做，我会进行详尽的测试。在我看来，这似乎是合理的，但它可能不适用于从编译器到内存模型的所有内容的特定实现。

另一种想法或检查此方法的一种方法是向缓冲区添加校验和并最后写入。

另请参阅有关无锁算法的搜索，例如 http://www.rossbencina.com/code/lockfree

为此，您可能需要一种方式让作者向熟睡的读者发出信号。您可以为此使用 Posix 信号量 - 例如，让读者要求编写者在特定信号量达到给定序列号或缓冲区有效时调用 sem_post()。

Answer 5

另一种选择是坚持锁定，但确保读取器不会长时间持有锁。读者可以通过在持有锁时不执行任何其他操作，而是从写入者的缓冲区复制数据来保持持有锁所花费的时间较短且可预测。唯一的问题是，低优先级读取器可能会在写入过程中被更高优先级任务打断，解决方法是 http://en.wikipedia.org/wiki/Priority_ceiling_protocol。

鉴于此，如果写入线程具有高优先级，则每个缓冲区要完成的最坏情况工作是写入线程填充缓冲区，以及每个读取线程将数据从该缓冲区复制到另一个缓冲区。如果您可以在每个周期中负担得起，那么写入器线程和一定量的读取器数据复制将始终完成，而处理其复制的数据的读取器可能会或可能不会完成其工作。如果他们不这样做，他们就会落后，并且当他们下次抓住锁并环顾四周以查看他们想要复制哪个缓冲区时会注意到这一点。

FWIW，我阅读实时代码的经验（当需要证明错误存在时，而不是在我们的代码中）是，它是令人难以置信的和故意简单的，布局非常清晰，并且不一定比它更有效需要满足其最后期限，因此如果您负担得起的话，为了使锁定能够直接工作而进行一些显然毫无意义的数据复制可能是一笔不错的交易。