c - 如何同步以下场景

Question

这是我正在尝试做的一个最小的例子。现在，如果运行此示例，您会发现该示例需要 10 秒多一点的时间才能完成。整个过程应该少于 2 秒。问题是存在竞争条件。循环时间太长，SetEvent 在 WaitForSingle 对象获取之前发生。如果事件可以触发，但 WaitForSingleObject 仍然知道它以某种方式触发，那就太好了。这里发生的情况是生成数据，这可能需要很长时间。然后，该数据通过网络发送，这可能需要更长的时间。因此，我想对要发送的数据进行排队，然后以我的快乐方式，同时另一个线程拾取数据并将其发送出去。这样我就可以无限排队，直到我无事可做，然后线程加入发送线程，直到它完成发送所有网络数据。

#include <stdio.h>
#include <windows.h>
#include <unistd.h>

#define PIPE_NAME "\\\\.\\pipe\\testpipe"

void copy_protocol_buffer(struct protocol_buffer *in, struct protocol_buffer *out);
DWORD PipeClientStartSendBufferThread(struct client_pipe_settings *pipe_settings, LPDWORD lpThreadId);
DWORD InitializeClientPipeSettings(struct client_pipe_settings *pipe_settings);
void initialize_protocol_buffer(struct protocol_buffer *protocol_buffer);

struct protocol_buffer {

    size_t length;
    size_t location;
    int data_type;
    char *data;
    struct protocol_buffer *next;
};

struct client_pipe_settings {
    HANDLE hPipe;
    LPCTSTR name;
    DWORD pipe_timeout;
    HANDLE write_event;
    struct protocol_buffer *fifo;
    HANDLE send_thread;
};


DWORD WINAPI PipeClientSendThread(LPVOID client_pipe_settings_object) {

    struct client_pipe_settings *pipe_settings = (struct client_pipe_settings *)client_pipe_settings_object;

    struct protocol_buffer *buf = NULL;

    while(1) {
        WaitForSingleObject(pipe_settings->write_event, 10000);

        if (buf == NULL) {
            buf = pipe_settings->fifo;
        } else {
            struct protocol_buffer *fifo_protocol_buffer = buf->next;
            free(buf);
            buf = fifo_protocol_buffer;

            if(buf->length == 0)
                //signal to quit
                return 0;
        }

        //Send data over the network
        Sleep(500);
    }

    return 0;
}

DWORD PipeQueueBuffer(struct client_pipe_settings *pipe_settings, struct protocol_buffer *buf)
{
    struct protocol_buffer *out_protocol_buffer = (struct protocol_buffer *)malloc(sizeof *buf);

    if(out_protocol_buffer == NULL)
        exit(1);

    copy_protocol_buffer(buf, out_protocol_buffer);

    if (pipe_settings->fifo == NULL) {
        pipe_settings->fifo = out_protocol_buffer;
    }
    else
    {
        struct protocol_buffer *last_protocol_buffer = pipe_settings->fifo;
        while(last_protocol_buffer->next != NULL)
        {
            last_protocol_buffer = last_protocol_buffer->next;
        }
        last_protocol_buffer->next = out_protocol_buffer;
    }

    if(!SetEvent(pipe_settings->write_event))
        return GetLastError();

    return ERROR_SUCCESS;
}



int main(void) {
    struct client_pipe_settings pipe_settings;
    InitializeClientPipeSettings(&pipe_settings);

    DWORD dwThreadId = 0;

    PipeClientStartSendBufferThread(&pipe_settings, &dwThreadId);

    //Generate data which could take a while
    Sleep(1000);
    struct protocol_buffer buf;

    initialize_protocol_buffer(&buf);

    buf.length = 5;
    buf.data = (char *)malloc(5);
    buf.data[0] = 'b';
    buf.data[1] = 'l';
    buf.data[2] = 'a';
    buf.data[3] = 'h';
    buf.data[4] = '\0';

    PipeQueueBuffer(&pipe_settings, &buf);

    Sleep(100);
    PipeQueueBuffer(&pipe_settings, &buf);

    buf.length = 0;

    PipeQueueBuffer(&pipe_settings, &buf);

    WaitForSingleObject(pipe_settings.send_thread, 100000);

}

DWORD InitializeClientPipeSettings(struct client_pipe_settings *pipe_settings)
{
    pipe_settings->write_event = CreateEvent(NULL, 0, 0, NULL);
    if(pipe_settings->write_event == NULL)
        return GetLastError();

    pipe_settings->hPipe = INVALID_HANDLE_VALUE;
    pipe_settings->fifo = NULL;

    pipe_settings->send_thread = NULL;


    return ERROR_SUCCESS;
}

DWORD PipeClientStartSendBufferThread(struct client_pipe_settings *pipe_settings, LPDWORD lpThreadId)
{
    HANDLE h = CreateThread(NULL, 0, PipeClientSendThread, pipe_settings, 0, lpThreadId);

    if(h == NULL)
        return GetLastError();

    pipe_settings->send_thread = h;

    return ERROR_SUCCESS;
}

void copy_protocol_buffer(struct protocol_buffer *in, struct protocol_buffer *out) {
    out->data_type = in->data_type;
    out->length = in->length;
    out->location = in->location;
    out->next = in->next;

    out->data = (char*)malloc(in->length);
    if (out->data == NULL) {
        exit(1);
    }

    memcpy(out->data, in->data, in->length);
}

void initialize_protocol_buffer(struct protocol_buffer *protocol_buffer)
{
    protocol_buffer->data = NULL;
    protocol_buffer->length = 0;
    protocol_buffer->location = 0;
    protocol_buffer->next = NULL;
    protocol_buffer->data_type = 0;
}

Answer 1

你的机制根本就是错误的。这与提前出现的 SetEvent 无关。

如果设置了事件，则可能会设置“多次”。 PipeClientSendThread 应该等待该事件，如果设置了该事件，它应该发送已到达队列的所有元素。您将 3 个元素编码到队列中，但事件设置一次，线程运行并一次仅发送一个元素，然后仅在达到超时时发送下一个元素......

你还有一个大问题。您的队列必须受到临界区或互斥锁的保护。您修改并循环队列中的元素，同时另一个线程也在读取和修改队列。

使用关键部分和 std::queue...这也将帮助您摆脱内存释放/malloc 的东西。