windows ReleaseSemaphore 不释放信号量

ReleaseSemaphore does not release the semaphore

提问by Gabriel

(In short: main()'s WaitForSingleObject hangs in the program below).

（简而言之：main() 的 WaitForSingleObject 挂在下面的程序中）。

I'm trying to write a piece of code that dispatches threads and waits for them to finish before it resumes. Instead of creating the threads every time, which is costly, I put them to sleep. The main thread creates X threads in CREATE_SUSPENDED state.

我正在尝试编写一段代码来调度线程并等待它们在恢复之前完成。我不是每次都创建线程，这是昂贵的，而是让它们进入睡眠状态。主线程在 CREATE_SUSPENDED 状态下创建 X 个线程。

The synch is done with a semaphore with X as MaximumCount. The semaphore's counter is put down to zero and the threads are dispatched. The threds perform some silly loop and call ReleaseSemaphore before they go to sleep. Then the main thread uses WaitForSingleObject X times to be sure every thread finished its job and is sleeping. Then it loops and does it all again.

同步是使用 X 作为 MaximumCount 的信号量完成的。信号量的计数器归零，线程被分派。线程执行一些愚蠢的循环并在它们进入睡眠之前调用 ReleaseSemaphore。然后主线程使用 WaitForSingleObject X 次以确保每个线程完成其工作并处于睡眠状态。然后它循环并再次执行所有操作。

From time to time the program does not exit. When I beak the program I can see that WaitForSingleObject hangs. This means that a thread's ReleaseSemaphore did not work. Nothing is printf'ed so supposedly nothing went wrong.

有时程序不退出。当我启动程序时，我可以看到 WaitForSingleObject 挂起。这意味着线程的 ReleaseSemaphore 不起作用。没有什么是 printf'ed 所以据说没有出错。

Maybe two threads shouldn't call ReleaseSemaphore at the exact same time, but that would nullify the purpose of semaphores...

也许两个线程不应该在完全相同的时间调用 ReleaseSemaphore，但这会使信号量的目的无效......

I just don't grok it...

我只是不理解它...

Other solutions to synch threads are gratefully accepted!

感谢同步线程的其他解决方案！

#define TRY  100
#define LOOP 100

HANDLE *ids;
HANDLE semaphore;

DWORD WINAPI Count(__in LPVOID lpParameter)
{ 
 float x = 1.0f;   
 while(1)
 { 
  for (int i=1 ; i<LOOP ; i++)
   x = sqrt((float)i*x);
  while (ReleaseSemaphore(semaphore,1,NULL) == FALSE)
   printf(" ReleaseSemaphore error : %d ", GetLastError());
  SuspendThread(ids[(int) lpParameter]);
 }
 return (DWORD)(int)x;
}

int main()
{
 SYSTEM_INFO sysinfo;
 GetSystemInfo( &sysinfo );
 int numCPU = sysinfo.dwNumberOfProcessors;

 semaphore = CreateSemaphore(NULL, numCPU, numCPU, NULL);
 ids = new HANDLE[numCPU];

 for (int j=0 ; j<numCPU ; j++)
  ids[j] = CreateThread(NULL, 0, Count, (LPVOID)j, CREATE_SUSPENDED, NULL);

 for (int j=0 ; j<TRY ; j++)
 {
  for (int i=0 ; i<numCPU ; i++)
  {
   if (WaitForSingleObject(semaphore,1) == WAIT_TIMEOUT)
    printf("Timed out !!!\n");
   ResumeThread(ids[i]);  
  }
  for (int i=0 ; i<numCPU ; i++)
   WaitForSingleObject(semaphore,INFINITE);
  ReleaseSemaphore(semaphore,numCPU,NULL);
 }
 CloseHandle(semaphore);
 printf("Done\n");
 getc(stdin);
}

回答by Jerry Coffin

Instead of using a semaphore (at least directly) or having main explicitly wake up a thread to get some work done, I've always used a thread-safe queue. When main wants a worker thread to do something, it pushes a description of the job to be done onto the queue. The worker threads each just do a job, then try to pop another job from the queue, and end up suspended until there's a job in the queue for them to do:

我一直使用线程安全队列，而不是使用信号量（至少直接使用）或让 main 显式唤醒线程来完成一些工作。当 main 想要一个工作线程做某事时，它会将要完成的工作的描述推送到队列中。每个工作线程只做一个工作，然后尝试从队列中弹出另一个工作，并最终挂起，直到队列中有一个工作让他们做：

The code for the queue looks like this:

队列的代码如下所示：

#ifndef QUEUE_H_INCLUDED
#define QUEUE_H_INCLUDED

#include <windows.h>

template<class T, unsigned max = 256>
class queue { 
    HANDLE space_avail; // at least one slot empty
    HANDLE data_avail;  // at least one slot full
    CRITICAL_SECTION mutex; // protect buffer, in_pos, out_pos

    T buffer[max];
    long in_pos, out_pos;
public:
    queue() : in_pos(0), out_pos(0) { 
        space_avail = CreateSemaphore(NULL, max, max, NULL);
        data_avail = CreateSemaphore(NULL, 0, max, NULL);
        InitializeCriticalSection(&mutex);
    }

    void push(T data) { 
        WaitForSingleObject(space_avail, INFINITE);       
        EnterCriticalSection(&mutex);
        buffer[in_pos] = data;
        in_pos = (in_pos + 1) % max;
        LeaveCriticalSection(&mutex);
        ReleaseSemaphore(data_avail, 1, NULL);
    }

    T pop() { 
        WaitForSingleObject(data_avail,INFINITE);
        EnterCriticalSection(&mutex);
        T retval = buffer[out_pos];
        out_pos = (out_pos + 1) % max;
        LeaveCriticalSection(&mutex);
        ReleaseSemaphore(space_avail, 1, NULL);
        return retval;
    }

    ~queue() { 
        DeleteCriticalSection(&mutex);
        CloseHandle(data_avail);
        CloseHandle(space_avail);
    }
};

#endif

And a rough equivalent of your code in the threads to use it looks something like this. I didn't sort out exactly what your thread function was doing, but it was something with summing square roots, and apparently you're more interested in the thread synch than what the threads actually do, for the moment.

在线程中使用它的粗略等效代码看起来像这样。我没有弄清楚你的线程函数到底在做什么，但它是求和平方根的东西，显然你对线程同步比目前线程实际做什么更感兴趣。

Edit: (based on comment): If you need main()to wait for some tasks to finish, do some more work, then assign more tasks, it's generally best to handle that by putting an event (for example) into each task, and have your thread function set the events. Revised code to do that would look like this (note that the queue code isn't affected):

编辑：（基于评论）：如果您需要main()等待某些任务完成，做更多工作，然后分配更多任务，通常最好通过将事件（例如）放入每个任务中来处理，并让您的线程函数设置事件。修改后的代码如下所示（注意队列代码不受影响）：

#include "queue.hpp"

#include <iostream>
#include <process.h>
#include <math.h>
#include <vector>

struct task { 
    int val;
    HANDLE e;

    task() : e(CreateEvent(NULL, 0, 0, NULL)) { }
    task(int i) : val(i), e(CreateEvent(NULL, 0, 0, NULL)) {}
};

void process(void *p) { 
    queue<task> &q = *static_cast<queue<task> *>(p);

    task t;
    while ( -1 != (t=q.pop()).val) {
        std::cout << t.val << "\n";
        SetEvent(t.e);
    }
}

int main() { 
    queue<task> jobs;

    enum { thread_count = 4 };
    enum { task_count = 10 };

    std::vector<HANDLE> threads;
    std::vector<HANDLE> events;

    std::cout << "Creating thread pool" << std::endl;
    for (int t=0; t<thread_count; ++t)
        threads.push_back((HANDLE)_beginthread(process, 0, &jobs));
    std::cout << "Thread pool Waiting" << std::endl;

    std::cout << "First round of tasks" << std::endl;

    for (int i=0; i<task_count; ++i) {
        task t(i+1);
        events.push_back(t.e);
        jobs.push(t);
    }

    WaitForMultipleObjects(events.size(), &events[0], TRUE, INFINITE);

    events.clear();

    std::cout << "Second round of tasks" << std::endl;

    for (int i=0; i<task_count; ++i) {
        task t(i+20);
        events.push_back(t.e);
        jobs.push(t);
    }

    WaitForMultipleObjects(events.size(), &events[0], true, INFINITE);
    events.clear();

    for (int j=0; j<thread_count; ++j)
        jobs.push(-1);

    WaitForMultipleObjects(threads.size(), &threads[0], TRUE, INFINITE);

    return 0;
}

回答by stmax

the problem happens in the following case:

问题发生在以下情况：

the main thread resumes the worker threads:

主线程恢复工作线程：

  for (int i=0 ; i<numCPU ; i++)
  {
   if (WaitForSingleObject(semaphore,1) == WAIT_TIMEOUT)
    printf("Timed out !!!\n");
   ResumeThread(ids[i]);  
  }

the worker threads do their work and release the semaphore:

工作线程完成它们的工作并释放信号量：

  for (int i=1 ; i<LOOP ; i++)
   x = sqrt((float)i*x);
  while (ReleaseSemaphore(semaphore,1,NULL) == FALSE)

the main thread waits for all worker threads and resets the semaphore:

主线程等待所有工作线程并重置信号量：

  for (int i=0 ; i<numCPU ; i++)
   WaitForSingleObject(semaphore,INFINITE);
  ReleaseSemaphore(semaphore,numCPU,NULL);

the main thread goes into the next round, trying to resume the worker threads (note that the worker threads haven't event suspended themselves yet! this is where the problem starts... you are trying to resume threads that aren't necessarily suspended yet):

主线程进入下一轮，尝试恢复工作线程（请注意，工作线程还没有自己挂起事件！这是问题开始的地方......您正在尝试恢复不一定挂起的线程然而）：

  for (int i=0 ; i<numCPU ; i++)
  {
   if (WaitForSingleObject(semaphore,1) == WAIT_TIMEOUT)
    printf("Timed out !!!\n");
   ResumeThread(ids[i]);  
  }

finally the worker threads suspend themselves (although they should already start the next round):

最后工作线程挂起自己（虽然他们应该已经开始下一轮）：

  SuspendThread(ids[(int) lpParameter]);

and the main thread waits forever since all workers are suspended now:

并且主线程永远等待，因为所有工人现在都被挂起：

  for (int i=0 ; i<numCPU ; i++)
   WaitForSingleObject(semaphore,INFINITE);

here's a link that shows how to correctly solve producer/consumer problems:

这是一个显示如何正确解决生产者/消费者问题的链接：

http://en.wikipedia.org/wiki/Producer-consumer_problem

http://en.wikipedia.org/wiki/Producer-consumer_problem

also i think critical sectionsare much faster than semaphores and mutexes. they're also easier to understand in most cases (imo).

我还认为关键部分比信号量和互斥锁快得多。在大多数情况下，它们也更容易理解（imo）。

回答by Hans Passant

I don't understand the code, but the threading sync is definitely bad. You assume that threads will call SuspendThread() in a certain order. A succeeded WaitForSingleObject() call doesn't tell you whichthread called ReleaseSemaphore(). You'll thus call ReleaseThread() on a thread that wasn't suspended. This quickly deadlocks the program.

我不明白代码，但线程同步肯定很糟糕。您假设线程将按特定顺序调用 SuspendThread()。成功的 WaitForSingleObject() 调用不会告诉您哪个线程调用了 ReleaseSemaphore()。因此，您将在未挂起的线程上调用 ReleaseThread()。这很快就会使程序死锁。

Another bad assumption is that a thread already called SuspendThread after the WFSO returned. Usually yes, not always. The thread could be pre-empted right after the RS call. You'll again call ReleaseThread() on a thread that wasn't suspended. That one usually takes a day or so to deadlock your program.

另一个错误的假设是在 WFSO 返回后线程已经调用了 SuspendThread。通常是的，并非总是如此。线程可以在 RS 调用后立即被抢占。您将再次在未挂起的线程上调用 ReleaseThread()。这通常需要一天左右的时间来使您的程序陷入僵局。

And I think there's one ReleaseSemaphore call too many. Trying to unwedge it, no doubt.

我认为一个 ReleaseSemaphore 调用太多了。毫无疑问，试图解开它。

You cannot control threading with Suspend/ReleaseThread(), don't try.

您无法使用 Suspend/ReleaseThread() 控制线程，不要尝试。

回答by Dingo

The problem is that you are waiting more often than you are signaling.

问题是您等待的次数比发出信号的次数要多。

The for (int j=0 ; j<TRY ; j++)loop waits eight times for the semaphore, while the four threads will only signal once each and the loop itself signals it once. The first time through the loop, this is not an issue of because the semaphore is given an initial count of four. The second and each subsequent time, you are waiting for too many signals. This is mitigated by the fact that on the first four waits you limit the time and don't retry on error. So sometimes it may work and sometimes your wait will hang.

该for (int j=0 ; j<TRY ; j++)循环等待信号量的八倍，而四个线程只会一次信号，每循环本身信号一次。第一次循环时，这不是问题，因为信号量的初始计数为 4。第二次和以后的每一次，您都在等待太多的信号。这可以通过在前四个等待中限制时间并且不重试错误这一事实而得到缓解。所以有时它可能会起作用，有时你的等待会挂起。

I think the following (untested) changes will help.

我认为以下（未经测试的）更改会有所帮助。

Initialize the semaphore to zero count:

将信号量初始化为零计数：

semaphore = CreateSemaphore(NULL, 0, numCPU, NULL);

Get rid of the wait in the thread resumption loop (i.e. remove the following):

摆脱线程恢复循环中的等待（即删除以下内容）：

   if (WaitForSingleObject(semaphore,1) == WAIT_TIMEOUT)  
      printf("Timed out !!!\n");  

Remove the extraneous signal from the end of the try loop (i.e. remove the following):

删除 try 循环末尾的无关信号（即删除以下内容）：

ReleaseSemaphore(semaphore,numCPU,NULL);

回答by Gabriel

Here is a practical solution.

这是一个实用的解决方案。

I wanted my main program to use threads (then using more than one core) to munch jobs and wait for all the threads to complete before resuming and doing other stuff. I did not want to let the threads die and create new ones because that's slow. In my question, I was trying to do that by suspending the threads, which seemed natural. But as nobugz pointed out, "Thou canst control threading with Suspend/ReleaseThread()".

我希望我的主程序使用线程（然后使用多个内核）来处理工作并等待所有线程完成，然后再继续执行其他操作。我不想让线程死掉并创建新线程，因为这很慢。在我的问题中，我试图通过暂停线程来做到这一点，这似乎很自然。但正如 nobugz 指出的那样，“您可以使用 Suspend/ReleaseThread() 控制线程”。

The solution involves semaphores like the one I was using to control the threads. Actually one more semaphore is used to control the main thread. Now I have one semaphore per thread to control the threads and one semaphore to control the main.

解决方案涉及信号量，就像我用来控制线程的信号量。实际上还有一个信号量用于控制主线程。现在我每个线程有一个信号量来控制线程，一个信号量来控制主线程。

Here is the solution:

这是解决方案：

#include <windows.h>
#include <stdio.h>
#include <math.h>
#include <process.h>

#define TRY  500000
#define LOOP 100

HANDLE *ids;
HANDLE *semaphores;
HANDLE allThreadsSemaphore;

DWORD WINAPI Count(__in LPVOID lpParameter)
{   
    float x = 1.0f;         
    while(1)
    {   
        WaitForSingleObject(semaphores[(int)lpParameter],INFINITE);
        for (int i=1 ; i<LOOP ; i++)
            x = sqrt((float)i*x+rand());
        ReleaseSemaphore(allThreadsSemaphore,1,NULL);
    }
    return (DWORD)(int)x;
}

int main()
{
    SYSTEM_INFO sysinfo;
    GetSystemInfo( &sysinfo );
    int numCPU = sysinfo.dwNumberOfProcessors;

    ids = new HANDLE[numCPU];
    semaphores = new HANDLE[numCPU]; 

    for (int j=0 ; j<numCPU ; j++)
    {
        ids[j] = CreateThread(NULL, 0, Count, (LPVOID)j, NULL, NULL);
        // Threads blocked until main releases them one by one
        semaphores[j] = CreateSemaphore(NULL, 0, 1, NULL);
    }
    // Blocks main until threads finish
    allThreadsSemaphore = CreateSemaphore(NULL, 0, numCPU, NULL);

    for (int j=0 ; j<TRY ; j++)
    {
        for (int i=0 ; i<numCPU ; i++) // Let numCPU threads do their jobs
            ReleaseSemaphore(semaphores[i],1,NULL);
        for (int i=0 ; i<numCPU ; i++) // wait for numCPU threads to finish
            WaitForSingleObject(allThreadsSemaphore,INFINITE);
    }
    for (int j=0 ; j<numCPU ; j++)
        CloseHandle(semaphores[j]);
    CloseHandle(allThreadsSemaphore);
    printf("Done\n");
    getc(stdin);
}