cửa sổ C++ thread chờ đợi trên hàng đợi đẩy dữ liệu

Question

Chương trình của tôi được thiết lập như sau:
Có một lớp xếp hàng an toàn chủ đề, một luồng đẩy dữ liệu vào nó trong khi đang ngồi trong vòng lặp vô hạn và dữ liệu luồng thứ hai xuất hiện nó trong khi ngồi trong một vòng lặp vô hạn. Tôi đang cố gắng nghĩ về cách sử dụng các sự kiện cửa sổ hoặc một số cơ chế khác để tạo luồng_1 (bên dưới), đợi trong vòng lặp vô hạn và chỉ lặp lại khi độ sâu hàng đợi lớn hơn hoặc bằng 1.

class thread-safe_Queue 
{ 
public: 
    push(); 
    pop(); 
}; 

DWORD thread_1() 
{ 
while(1) 
{ 
    // wait for thread-safe queue to have data on it 
    // pop data off 
    // process data 
} 
} 

DWORD thread_2() 
{ 
while(1) 
{ 
    // when data becomes available, push data onto thread-safe queue 
} 
} 

Answer 1

Làm thế nào về điều này (tôi giả sử bạn đã quen thuộc với cơ chế sự kiện).

1.

thread_safe_Queue::push(something) 
{ 
// lock the queue 
... 
// push object 
// Signal the event 
SetEvent(notification); 

// unlock the queue 
} 

2.

thread_safe_Queue::pop(something) 
{ 
WaitForSingleObject(notification); 
// lock the queue 
... 
// get object 
// reset the event 
if (queue is empty) 
    ResetEvent(notification); 

// unlock the queue 
} 

3. thread_1 chỉ cố gắng để bật các đối tượng và xử lý nó. Khi một cái gì đó được đẩy, sự kiện được kích hoạt, do đó, pop có thể được gọi thành công. Nếu không, nó sẽ đợi bên trong pop. Trên thực tế, bạn có thể sử dụng các đối tượng đồng bộ hóa khác như mutexes hoặc các phần quan trọng thay vì các sự kiện trong trường hợp này.

CẬP NHẬT. sự kiện bên ngoài: thread 1:

void thread_1() 
    { 
    while(1) 
    { 
    WaitForSingleObject(notification); 
    if (!pop(object)) // pop should return if there are any objects left in queue 
     SetEvent(notification);  
    } 
    } 

thread_2

void thread_2() 
    { 
    while(1) 
    { 
    // push the object and than signal event 
    ResetEvent(notification) 
    } 
    } 

Answer 2

Bạn có thể sử dụng các sự kiện được đặt tên. Mỗi thread sẽ gọi CreateEvent truyền trong cùng một tên. Sau đó sử dụng WaitForMultipleObjects để đợi sự kiện liên quan đến hàng đợi hoặc sự kiện chương trình kết thúc. Chủ đề pop sẽ đợi các sự kiện queue_has_data và end_program. Chủ đề đẩy sẽ đợi các sự kiện data_available và end_program và đặt sự kiện queue_has_data khi nó đặt thứ gì đó vào hàng đợi.

Answer 3

Tôi nghĩ rằng điều này có thể thực hiện thủ thuật. Sự kiện lớp bắt nguồn và quá tải hàm Process().

#include <process.h> // Along with all the normal windows includes 

//********************************************* 
using namespace os; 

Mutex globalQueueMutex; 

class QueueReader : public Event 
{ 
public: 
    virtual void Process() 
    { 
     // Lock the queue 
     Locker l(globalQueueMutex); 
     // pop data off 
     // process data 
     return; // queue will automatically unlock 
    } 
}; 

QueueReader myQueueReader; 

//********************************************* 
// The queue writer would have functions like : 
void StartQueueReader() 
{ 
    Thread(QueueReader::StartEventHandler, &myQueueReader); 
} 
void WriteToQueue() 
{ 
    Locker l(globalQueueMutex); 
    // write to the queue 
    myQueueReader.SignalProcess(); // tell reader to wake up 
} 
// When want to shutdown 
void Shutdown() 
{ 
    myQueueReader.SignalShutdown(); 
} 

Đây là các lớp thực hiện phép thuật.

namespace os { 

// ********************************************************************** 
/// Windows implementation to spawn a thread. 
static uintptr_t Thread (void (*StartAddress)(void *), void *ArgList) 
{ 
    return _beginthread(StartAddress, 0, ArgList); 
} 

// ********************************************************************** 
/// Windows implementation of a critical section. 
class Mutex 
{ 
public: 
    // Initialize section on construction 
    Mutex() { InitializeCriticalSection(&cs_); } 
    // Delete section on destruction 
    ~Mutex() { DeleteCriticalSection(&cs_); } 
    // Lock it 
    void lock() { EnterCriticalSection(&cs_); } 
    // Unlock it 
    void unlock() { LeaveCriticalSection(&cs_); } 

private: 
    CRITICAL_SECTION cs_; 
}; // class Mutex 

/// Locks/Unlocks a mutex 
class Locker 
{ 
public: 
    // Lock the mutex on construction 
    Locker(Mutex& mutex): mutex_(mutex) { mutex_.lock(); } 
    // Unlock on destruction 
    ~Locker() { mutex_.unlock(); } 
private: 
    Mutex& mutex_; 
}; // class Locker 

// ********************************************************************** 
// Windows implementation of event handler 
#define ProcessEvent hEvents[0] 
#define SetTimerEvent hEvents[1] 
#define ShutdownEvent hEvents[2] 

/// Windows implementation of events 
class Event 
{ 
    /// Flag set when shutdown is complete 
    bool Shutdown; 
    /// Max time to wait for events 
    DWORD Timer; 
    /// The three events - process, reset timer, and shutdown 
    HANDLE hEvents[3]; 

public: 
    /// Timeout is disabled by default and Events assigned 
    Event(DWORD timer = INFINITE) : Timer(timer) 
    { 
    Shutdown = false; 
    ProcessEvent = CreateEvent(NULL,TRUE,FALSE,NULL); 
    SetTimerEvent = CreateEvent(NULL,TRUE,FALSE,NULL); 
    ShutdownEvent = CreateEvent(NULL,TRUE,FALSE,NULL); 
    } 

    /// Close the event handles 
    virtual ~Event() 
    { 
    CloseHandle(ProcessEvent); 
    CloseHandle(SetTimerEvent); 
    CloseHandle(ShutdownEvent); 
    } 

    /// os::Thread calls this to start the Event handler 
    static void StartEventHandler(void *pMyInstance) 
    { ((Event *)pMyInstance)->EventHandler(); } 
    /// Call here to Change/Reset the timeout timer 
    void ResetTimer(DWORD timer) { Timer = timer; SetEvent(SetTimerEvent); } 
    /// Set the signal to shutdown the worker thread processing events 
    void SignalShutdown() { SetEvent(ShutdownEvent); while (!Shutdown) Sleep(30);} 
    /// Set the signal to run the process 
    void SignalProcess() { SetEvent(ProcessEvent); } 

protected: 
    /// Overload in derived class to process events with worker thread 
    virtual void Process(){} 
    /// Override to process timeout- return true to terminate thread 
    virtual bool Timeout(){ return true;} 

    /// Monitor thread events 
    void EventHandler() 
    { 
    DWORD WaitEvents; 
    while (!Shutdown) 
    { 
     // Wait here, looking to be signaled what to do next 
     WaitEvents = WaitForMultipleObjects(3, hEvents, FALSE, Timer); 

     switch (WaitEvents) 
     { 
     // Process event - process event then reset for the next one 
     case WAIT_OBJECT_0 + 0: 
      Process(); 
      ResetEvent(ProcessEvent); 
      break; 

     // Change timer event - see ResetTimer(DWORD timer) 
     case WAIT_OBJECT_0 + 1: 
      ResetEvent(SetTimerEvent); 
      continue; 

     // Shutdown requested so exit this thread 
     case WAIT_OBJECT_0 + 2: 
      Shutdown = true; 
      break; 

     // Timed out waiting for an event 
     case WAIT_TIMEOUT: 
      Shutdown = Timeout(); 
      break; 

     // Failed - should never happen 
     case WAIT_FAILED: 
      break; 

     default: 
      break; 
     } 
    } 
    } 


}; 

} // namespace os