创建一个将阻塞的函数，直到它被超过 n/2 个线程(伪代码)调用

Question

有n线程。我正在尝试实现一个函数(伪代码)，如果它被线程调用，它将直接阻塞。每个线程都将被阻塞，如果该函数被超过n/2个线程调用，则该函数将停止阻塞线程。如果调用该函数的线程超过n/2，则该函数将不再阻止其他线程，而是立即返回。

我是这样做的，但我不确定我是否正确完成了最后一部分，如果超过n/2个线程调用它，函数将立即返回？ ：S

(伪代码是高度赞赏的，因为这样我就有更好的机会理解它:)！

int n = total amount of threads
sem waiter = 0
sem mutex = 1
int counter = 0
function void barrier()
int x
P(mutex)
if counter > n / 2 then
V(mutex)
for x = 0; x <= n / 2; x++;
V(waiter)
end for
end if
else
counter++
V(mutex)
P(waiter)
end else
end function

Answer 1

您描述的是非重置障碍。 Pthreads 有一个屏障实现，但它是重置的种类。

要实现你对 pthreads 的期望，你需要一个互斥锁加一个条件变量和一个共享计数器。 进入函数的线程锁定互斥锁并检查计数器。 如果没有足够的其他线程到达，则它会等待 CV，否则它会广播给它以唤醒所有等待的线程。 如果您愿意，您可以将其设置为提示广播规模的线程。 例：

struct my_barrier {
pthread_mutex_t barrier_mutex;
pthread_cond_t barrier_cv;
int threads_to_await;
};
void barrier(struct my_barrier *b) {
pthread_mutex_lock(&b->barrier_mutex);
if (b->threads_to_await > 0) {
if (--b->threads_to_await == 0) {
pthread_cond_broadcast(&b->barrier_cv);
} else {
do {
pthread_cond_wait(&b->barrier_cv, &b->barrier_mutex);
} while (b->threads_to_await);
}
}
pthread_mutex_unlock(&b->barrier_mutex);
}

更新：伪代码

或者，由于伪代码表示对你很重要，所以伪代码语言中的同样的事情类似于问题中使用的语言：

int n = total amount of threads
mutex m
condition_variable cv
int to_wait_for = n / 2
function void barrier()
lock(mutex)
if to_wait_for == 1 then
to_wait_for = 0
broadcast(cv)
else if to_wait_for > 1 then
to_wait_for = to_wait_for - 1
wait(cv)
end if
unlock(mutex)
end function

这比伪代码的级别略高，因为它不假定互斥锁是作为信号量实现的。(对于您标记的 pthreads，您需要一个 pthreads 互斥锁，而不是信号量，才能与 pthreads 条件变量一起使用)。 它还省略了真实 C 代码的细节，这些代码处理等待条件变量和初始化互斥体和 cv 的杂散唤醒。 此外，它呈现变量就好像它们都是全局变量一样——这样的函数在实践中可以这样实现，但它的形式很差。

另请注意，它假定 pthreads 是条件变量的语义：等待 cv 将暂时释放互斥锁，允许其他线程锁定它，但等待 cv 的线程将在自身通过等待之前重新获取互斥锁。

Answer 2

我在回答中做出的一些假设：

• P(...)类似于sem_wait(...)
• V(...)类似于sem_post(...)
• 障碍无法重置

我

---

不确定我是否正确完成了最后一部分，如果超过n/2个线程调用它，该函数将立即返回

伪代码在大多数情况下应该可以正常工作，但早期返回/退出条件可以得到显着改善。

一些问题(但没什么大不了的)：

• 第一次满足条件counter > n / 2时，waiter信号灯发出信号(即V(...))(n / 2) + 1次(因为它是从0到n / 2包含)，而不是n / 2(这也是当时counter的价值)。
• 首次满足counter > n / 2之后的每个后续调用也将发出信号(即V(...))waiter信号量又(n / 2) + 1次。相反，它应该尽早返回，而不是重新发出信号。

这些可以通过一些小的调整来解决。

int n = total count of threads
sem mutex = 1;
sem waiter = 0;
int counter = 0;
bool released = FALSE;
function void barrier() {
P(mutex)
// instead of the `released` flag, could be replaced with the condition `counter > n / 2 + 1`
if released then
// ensure the mutex is released prior to returning
V(mutex)
return
end if
if counter > n / 2 then
// more than n/2 threads have tried to wait, mark barrier as released
released = TRUE
// mutex can be released at this point, as any thread acquiring `mutex` after will see that `release` is TRUE and early return
V(mutex)
// release all blocked threads; counter is guaranteed to never be incremeneted again
int x
for x = 0; x < counter; x++
V(waiter)
end for
else
counter++
V(mutex)
P(waiter)
end else
}