一个线程维护多个信号量的问题

Question

问题描述：1. 建立了多个buffer，每个buffer都有独自的CRITICAL_SECTION变量，一个读计数semaphore，一个写计数semaphore。分别初始化它们。2. 建立了一个读线程，一个写线程。写线程随机往这几个buffer中写数，读线程轮询这些buffer，若有数据则读出。现在的问题是，写线程往其中一个buffer中写一个数时，所有buffer的写计数semaphore都跟着加1。比如，最开始我往buffer1中写一个数，往buffer2中写一个数，正常情况下只有buffer1和buffer2的写计数semaphore是1，其他的都是0。但实际上所有buffer的写计数semaphore都变成了2。刚开始我以为是初始化时这几个buffer的写计数semaphore指向了同一个HANDLE，可是打印出它们的地址，发现并不是同一个。于是我没辙了，特来向各位大侠求救！

Answer 1

1. 建立了多个buffer，每个buffer都有独自的CRITICAL_SECTION变量，一个读计数semaphore，一个写计数 semaphore。分别初始化它们。
是每个都有一个semaphore?还是公用了一个?如果是公用一个,就是出现你的问题,如果不是,拿你代码,看看还有,你只有一个写线程,用什么CRITICAL_SECTION锁定呢?多线程写操作才使用这个东西的.
再有,使用CRITICAL_SECTION的情况是,多线程写一个缓冲区,也就是只有一个buffer(集合).

Answer 2

谢谢各位。整理一下代码发上来。好像不止这几个buffer的写计数semaphore指向了同一个HANDLE，读计数semaphore也同样。所以读写操作全乱套了。
（之前有个POSIX版本运行在linux上是可以的，但是移植到win32上就出问题了 = =|||）/* buffer的结构（我给它起名为MSG_QUEUE，因为实现的功能类似消息队列） */
typedef struct _MSG_QUEUE
{
    CRITICAL_SECTION cs;  
    HANDLE           sem_empty;  /* 读计数semaphore */
    HANDLE           sem_full;   /* 写计数semaphore */
    U32              ptr_wr;     /* 写指针 */
    U32              ptr_rd;     /* 读指针 */
    MSG_UNIT         unit[MSG_QUEUE_LEN]; /* MSG_QUEUE_LEN = 8 */
} MSG_QUEUE;/* 初始化msgq */
BOOL msgq_init(MSG_QUEUE * msgq)
{
    U32 i = 0;
    assert(msgq != NULL);
    InitializeCriticalSection(&msgq->cs);    if((msgq->sem_full =
        CreateSemaphore(NULL, 0, MSG_QUEUE_LEN, "SEM_FULL")) == NULL)
    {
        printf("[msgq_init] sem_full init error\n");
        return FALSE;
    }
    printf("[msgq_init]&msgq->sem_full = %d\n", &msgq->sem_full);    if((msgq->sem_empty =
        CreateSemaphore(NULL, MSG_QUEUE_LEN, MSG_QUEUE_LEN, "SEM_EMPTY")) == NULL)
    {
        printf("[msgq_init] sem_empty init error\n");
        return FALSE;
    }    msgq->ptr_rd = 0;
    msgq->ptr_wr = 0;    for(; i < MSG_QUEUE_LEN; i++)
    {
        msgq->unit[i].entity_id = -1;
        msgq->unit[i].event = -1;
    }    return TRUE;
}/* msgq写操作 */
BOOL msg_snd(MSG_QUEUE * msgq, MSG_UNIT * snd_unit)
{
    LONG lpPC;    if(WaitForSingleObject(msgq->sem_empty, INFINITE) != WAIT_OBJECT_0)
    {
        printf("[msg_snd] WaitForSingleObject ERROR\n");
        return FALSE;
    }
    EnterCriticalSection(&msgq->cs);     msgq->unit[msgq->ptr_wr++] = *snd_unit;
    if(msgq->ptr_wr == MSG_QUEUE_LEN)
    {
        msgq->ptr_wr = 0;
    }    LeaveCriticalSection(&msgq->cs);
    if(!ReleaseSemaphore(msgq->sem_full, 1, &lpPC)) 
    {
        printf("[msg_snd] ReleaseSemaphore ERROR\n");
        return FALSE;
    }
    printf("[msg_snd]sem_full value: %d, &sem_full: %d \n", 
           lpPC, &msgq->sem_full);    return TRUE;
}/* msgq读操作(非阻塞) */
BOOL msg_rcv_nonblk(MSG_QUEUE * msgq, MSG_UNIT * rcv_unit)
{
    LONG lpPC;
    DWORD res = WaitForSingleObject(msgq->sem_full, 0); 
    if(res == WAIT_TIMEOUT) /* queue is empty, return without blocking */
    {
        printf("[msg_rcv_nonblk] msg queue empty\n");
        return FALSE;
    }
    else if(res == WAIT_FAILED)
    {
        printf("[msg_rcv_nonblk] WaitForSingleObject ERROR\n");
        return FALSE;
    }
    EnterCriticalSection(&msgq->cs);    *rcv_unit = msgq->unit[msgq->ptr_rd++];
    if(msgq->ptr_rd == MSG_QUEUE_LEN)
    {
        msgq->ptr_rd = 0;
    }    LeaveCriticalSection(&msgq->cs); 
    if(!ReleaseSemaphore(msgq->sem_empty, 1, &lpPC)) 
    {
        printf("[msg_rcv_nonblk] ReleaseSemaphore ERROR\n");
    }    printf("[msg_rcv_nonblk] sem_empty value: %d, &sem_empty: %d \n",
           lpPC, &msgq->sem_empty);    return TRUE;
}/* 写线程 - 往LOGICAL_CH_MAX_NUM个msgqueue中各发1条消息 */
DWORD WINAPI sender_func(LPVOID lpThreadParm)
{
    MSG_QUEUE * msgq = (MSG_QUEUE *)lpThreadParm;
    MSG_UNIT msg_s;
    U32 i;    msg_s.event = 1;
    msg_s.entity_id = 2;
    msg_s.cont[0] = 3;    for(i = 0; i < LOGICAL_CH_MAX_NUM; i++)
    {
        msg_snd(&msgqueue[i], &msg_s);
        msg_s.event++;
        msg_s.entity_id++;
    }    while(1);    _endthreadex(0);
}/* 读线程 - 等写线程发完消息后，只读msgqueue[0]的消息。
   ——结果读出来8条，参见后面的打印结果 */
DWORD WINAPI receiver_func(LPVOID lpThreadParm)
{
    MSG_QUEUE * msgq = (MSG_QUEUE *)lpThreadParm;
    MSG_UNIT msg_r;    while(1)
    {
        Sleep(100);
        if(msg_rcv_nonblk(&msgqueue[0], &msg_r))
            printf("[receiver_func] msg: %d %d %d\n", 
                   msg_r.event, msg_r.entity_id, msg_r.cont[0]);
    }    _endthreadex(0);
}int main()
{
    MSG_QUEUE msgqueue[LOGICAL_CH_MAX_NUM]; /* LOGICAL_CH_MAX_NUM = 8 */
    HANDLE hd_thread_sender, hd_thread_receiver;
    DWORD res;
    U32 i;    for(i = 0; i < LOGICAL_CH_MAX_NUM; i++)
    {
        msgq_init(&msgqueue[i]); /* 分别初始化LOGICAL_CH_MAX_NUM个msgqueue */
    }    hd_thread_sender = (HANDLE)_beginthreadex
                                (NULL, 0, sender_func, (LPVOID)msgqueue, 0, NULL);
    hd_thread_receiver = (HANDLE)_beginthreadex
                                (NULL, 0, receiver_func, (LPVOID)msgqueue, 0, NULL);    res = WaitForSingleObject(hd_thread_sender, INFINITE) ||
          WaitForSingleObject(hd_thread_receiver, INFINITE);
    if(res != WAIT_OBJECT_0)
        printf("[main] threads return ERROR\n");    system("PAUSE");
    return 0;
}
打印结果：
[msgq_init]&msgq->sem_full = 4298204
[msgq_init]&msgq->sem_full = 4298388
[msgq_init]&msgq->sem_full = 4298572
[msgq_init]&msgq->sem_full = 4298756
[msgq_init]&msgq->sem_full = 4298940
[msgq_init]&msgq->sem_full = 4299124
[msgq_init]&msgq->sem_full = 4299308
[msgq_init]&msgq->sem_full = 4299492
[msg_snd]sem_full value: 0, &sem_full: 4298204
[msg_snd]sem_full value: 1, &sem_full: 4298388
[msg_snd]sem_full value: 2, &sem_full: 4298572
[msg_snd]sem_full value: 3, &sem_full: 4298756
[msg_snd]sem_full value: 4, &sem_full: 4298940
[msg_snd]sem_full value: 5, &sem_full: 4299124
[msg_snd]sem_full value: 6, &sem_full: 4299308
[msg_snd]sem_full value: 7, &sem_full: 4299492
[msg_rcv_nonblk] sem_empty value: 0, &sem_empty: 4298200
[receiver_func] msg: 1 2 3
[msg_rcv_nonblk] sem_empty value: 1, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] sem_empty value: 2, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] sem_empty value: 3, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] sem_empty value: 4, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] sem_empty value: 5, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] sem_empty value: 6, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] sem_empty value: 7, &sem_empty: 4298200
[receiver_func] msg: 255 255 0
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
[msg_rcv_nonblk] msg queue empty
...

Answer 3

是每个buffer都有一个semaphore。我以为只要可能同时对临界区有读写操作，就要用CRITICAL_SECTION保护呢。比如说，如果写线程正写了一半，读线程就读出来了，或者读线程读了一半，写线程就去写，会不会出问题？

Answer 4

建立多个buffer，每个buffer都有独自的CRITICAL_SECTION变量，一个读计数semaphore，一个写计数 semaphore。分别初始化它们

Answer 5

ls，我又看了几遍，确实是分别初始化的啊。    for(i = 0; i < LOGICAL_CH_MAX_NUM; i++)
    {
        msgq_init(&msgqueue[i]); /* 分别初始化LOGICAL_CH_MAX_NUM个msgqueue */
    }

Answer 6

如果你直接把
HANDLE           sem_empty;  /* 读计数semaphore */
HANDLE           sem_full;   /* 写计数semaphore */
两个改成unsignd long行不？

Answer 7

静下心来看了本win32编程的书，终于把bug捉出来了！果然是个很低级的错误：
msgq->sem_full = CreateSemaphore(NULL, 0, MSG_QUEUE_LEN, "SEM_FULL");
msgq->sem_empty = CreateSemaphore(NULL, MSG_QUEUE_LEN, MSG_QUEUE_LEN, "SEM_EMPTY");
注意其中的“SEM_FULL”和“SEM_EMPTY”。在多次调用初始化函数时给不同semaphore命了相同的名字，所以系统认为这是共享handle，所以它们虽然handle地址不同，但实际上是同一个semaphore！当时移植时直接抓来API函数就往里套，压根没关参数的意义，果然浮躁是大忌啊。可以结贴了，于是这40分可以奖给自己么？呵呵。