我需要的 pthread 線程集結(jié)點功能，使用同一集結(jié)點的線程將通過 rend_wait 函數(shù)等待，當(dāng)集結(jié)點到達(dá)指定數(shù)量的線程后同時激發(fā)繼續(xù)執(zhí)行。使用 pthread 的 mutex 和 cond 超輕量實現(xiàn)。下面 rend.h 是集結(jié)點實現(xiàn)，rendezvous.c 是測試應(yīng)用。

    /*
 * rend.h
 *
 *  Created on: 2009-11-14
 *      Author: liuzy (lzy.dev@gmail.com)
 */

#ifndef REND_H_
#define REND_H_

#include <pthread.h>
#include <assert.h>

struct rend_t {
	volatile int count;
	pthread_mutex_t count_lock;
	pthread_cond_t ready;
};

#define DECLARE_REND(name, count) /
	struct rend_t name = {(count), PTHREAD_MUTEX_INITIALIZER, PTHREAD_COND_INITIALIZER}

int rend_init(struct rend_t* prend, int count) {
	int ret = 0;

	assert(prend);

	prend->count = count;

	if ((ret = pthread_mutex_init(&prend->count_lock, NULL)))
		return ret;

	if ((ret = pthread_cond_init(&prend->ready, NULL)))
		return ret;

	return EXIT_SUCCESS;
}

int rend_wait(struct rend_t* prend) {
	int ret = 0;

	assert(prend);

	if ((ret = pthread_mutex_lock(&prend->count_lock)))
		return ret;

	/* check count value is ready to weak up block code */
	if (prend->count == 1) {
		if ((ret = pthread_cond_broadcast(&prend->ready)))
			return ret;

		if ((ret = pthread_mutex_unlock(&prend->count_lock)))
			return ret;
	} else {
		prend->count--;

		ret = pthread_cond_wait(&prend->ready, &prend->count_lock);
		prend->count++;

		if (ret) {
			pthread_mutex_unlock(&prend->count_lock);
			return ret;
		}

		if ((ret = pthread_mutex_unlock(&prend->count_lock)))
			return ret;
	}

	return EXIT_SUCCESS;
}

int rend_free(struct rend_t* prend) {
	int ret = 0;

	assert(prend);

	prend->count = 0;

	if ((ret = pthread_mutex_destroy(&prend->count_lock)))
		return ret;

	if ((ret = pthread_cond_destroy(&prend->ready)))
		return ret;

	return EXIT_SUCCESS;
}

#endif /* REND_H_ */

  

rend 使用更簡單：

1. 定義/初始化 rend_t 集結(jié)點對象。DECLARE_REND 宏用于靜態(tài)定義，rend_init 函數(shù)可以對動態(tài)創(chuàng)建的集結(jié)點結(jié)構(gòu)初始化；
2. pthread 線程通過調(diào)用 rend_wait 函數(shù) P/V 集結(jié)狀態(tài)。集結(jié)關(guān)系的線程要 P/V 在同一個 rend_t 集結(jié)對象上；
3. 釋放集結(jié)對象，rend_free 函數(shù)。

以上函數(shù)都是成功返回 0，出錯返回 errno 值（非 0）。

    /*
 ==============================
 Name        : rendezvous.c
 Author      : liuzy (lzy.dev@gmail.com)
 Version     : 0.1
==============================
 */

#include <stdio.h>
#include <stdlib.h>
#include <stdarg.h>		/* va_list */
#include <unistd.h>
#include <string.h>
#include <errno.h>		/* errno */
#include <syslog.h>		/* for syslog(2) and level */
#include <pthread.h>

#include "rend.h"

static int daemon_proc = 0;	/* for syslog in err_doit */

#define	MAXLINE 4096		/* max text line length */

void err_doit(int errnoflag, int level, const char* fmt, va_list ap) {

	char buf[MAXLINE + 1] = { 0 };
	int errno_save = errno, n = 0;

#ifdef HAVE_VSNPRINTF
	vsnprintf(buf, MAXLINE, fmt, ap);
#else
	vsprintf(buf, fmt, ap);
#endif	/* HAVE_VSNPRINTF */

	n = strlen(buf);
	if (errnoflag)
		snprintf(buf + n, MAXLINE - n, ": %s", strerror(errno_save));
	strcat(buf, "/n");

	if (daemon_proc) {
		syslog(level, "%s", buf);
	} else {
		fflush(stdout);
		fputs(buf, stderr);
		fflush(stderr);
	}

	return;
}

void err_msg(const char* fmt, ...) {
	va_list ap;

	va_start(ap, fmt);
	err_doit(0, LOG_INFO, fmt, ap);
	va_end(ap);

	return;
}

void err_sys(const char* fmt, ...) {
	va_list ap;

	va_start(ap, fmt);
	err_doit(1, LOG_ERR, fmt, ap);
	va_end(ap);

	exit(EXIT_FAILURE);
}

#define THREAD_COUNT 100	/* rendezvous test thread workers */

struct worker_arg {
	int worker_id;
	struct rend_t* prend;
};

static void* pthread_worker(void* arg) {
	struct worker_arg* parg = (struct worker_arg*) arg;

	err_msg("worker #%d running.", (int) parg->worker_id);

	srand(parg->worker_id * 2);
	sleep(rand() % 5);

	rend_wait(parg->prend);	/* workers rendezvous */

	err_msg("worker #%d exiting.", (int) parg->worker_id);

	return EXIT_SUCCESS;
}

int main(void) {
	int idx = 0;
	void* exitcode = NULL;

	pthread_t thds[THREAD_COUNT];
	struct worker_arg arg[THREAD_COUNT];
	DECLARE_REND(rend, THREAD_COUNT);

	err_msg("workers creating.");

	for (idx = 0; idx < THREAD_COUNT; idx++) {
		arg[idx].prend = &rend;
		arg[idx].worker_id = idx;

		if (pthread_create(thds + idx, NULL, pthread_worker, (void*) &arg[idx]))
			err_sys("worker #%d create error.", idx);
	}

	puts("workers exiting.");

	for (idx = 0; idx < THREAD_COUNT; idx++)
		if (pthread_join(thds[idx], &exitcode) || (exitcode != EXIT_SUCCESS))
			err_msg("worker #%d exit error.", idx);

	err_msg("all done. exit 0.");

	rend_free(&rend);

	return EXIT_SUCCESS;
}

  

看了下 semaphore os syscall 及其 infrastructure，也許以后還需要進(jìn)程間（非 pthread）集結(jié)時用得上。kernel 實現(xiàn)的超強啊，呵呵～

// 2009.11.17 14:34 添加 ////

有幾點不還不理解：

1. “futex 如果說是一個用戶空間的整數(shù)值，那怎么被多個進(jìn)程共享？Futex 系統(tǒng)調(diào)用在 kernel 態(tài)怎么操作該值并仲裁競爭？這是那種直接映射到 userspace 的 kernel 地址么。 這個需要程序間通過 mmap 在共享段中訪問，與 futex 沒什么關(guān)系。
2. 這個“robust futex”機(jī)制指的應(yīng)該就是 SVRx 傳統(tǒng) sem IPC 里的 SEM_UNDO flag 吧？

一篇不錯的文章，引發(fā)對 glibc nptl 實現(xiàn)源碼的探索：

關(guān)于信號量與線程互斥鎖的區(qū)別與實現(xiàn)

超輕量 pthread 集結(jié)點實現(xiàn)