Synchronization of multiple reader and writer processes using Win32 APIs

• Download demo project - 52.3 Kb
• Download source - 5.15 Kb

Introduction

This source code demonstrates how to use Win32 events to solve the multiple readers and writers problem. On compilation of this source code, you will get a console application that can act as both a reader and a writer of the shared memory. Multiple instances of the same executable should be executed for testing. Only one write operation is allowed at a time. If there is already a write operation going on, all other write and read operations will get blocked. However, multiple read operations are allowed. If read operations are underway, a write operation will wait for all read operations to get over before proceeding. This synchronization mechanism ensures proper synchronization of readers and writers, and helps in prevention of data corruption at all times.

Background

I was asked to design a solution for the multiple readers and writers problem, in a recent interview. I described at an abstract level what I had in mind. But the more that I thought on this, the more I was convinced that I should implement it to flesh out the ideas that I had in mind. Hence, this source code.

Using the code

This source code has a shared memory which the readers and writers are trying to use. The synchronization is done using events. I have tried to add comments at the right places to make understanding of this source code easier.

Although, this code is in C++, it can easily be converted to C. This program doesn't use any object oriented concept or design.

The code includes <tchar.h> and uses generic functions at many places. However, this code was compiled and tested with ASCII builds only. The code may not compile or work for Unicode builds.

Following is the initialization function which initializes the events and the shared memory:

bool Initialize()
{
     //Trying to be defensive
     for (int ii = 0; ii < MAX_READ_PROCESSES_ALLOWED; ii++)
     {
          g_hReadEvent[ii] = NULL;
     }
     
     TCHAR szBuffer[32];  //big enough to hold the name
     
     //Creating Read events
     for (ii = 0; ii < MAX_READ_PROCESSES_ALLOWED; ii++)
     {
          _stprintf(szBuffer, _T("%s %d"), 
                             g_szReadEventName, ii);
          
          //Creates or opens event depending, 
          //on whether already exists or not
          g_hReadEvent[ii] = CreateEvent( 
               NULL,        //default security
               false,       //auto reset
               true,        //default state signaled
               szBuffer);
          
          if (NULL == g_hReadEvent[ii])
          {
               return false;
          }
     }
     
     //Write Event
     //Creates or opens event, depending 
     //on whether already exists or not
     g_hWriteEvent = CreateEvent( 
          NULL,             //default security
          false,            //auto reset
          true,             //default state signaled
          g_szWriteEventName);
     
     if (NULL == g_hWriteEvent)
     {
          return false;
     }
     
     //Shared Memory Stuff
     //Creates or opens shared memory, 
     //depending on whether already exists or not
     g_hSharedMemory = CreateFileMapping(
          INVALID_HANDLE_VALUE,    // use paging file
          NULL,                    // default security 
          PAGE_READWRITE,          // read/write access
          0,                       // max. object size 
          MAX_SH_MEM_SIZE,         // buffer size  
          g_szShareMemoryName);    // name of mapping object
     
     if (NULL == g_hSharedMemory || 
         INVALID_HANDLE_VALUE == g_hSharedMemory) 
     { 
          cout << "Error occured while" 
                  " creating file mapping object :" 
               << GetLastError() << "\n";
          return false;
     }
     
     g_pBuffer = 
          (LPTSTR) MapViewOfFile(g_hSharedMemory, // handle to map object
          FILE_MAP_ALL_ACCESS, // read/write permission
          0,                   
          0,                   
          MAX_SH_MEM_SIZE);           
     
     if (NULL == g_pBuffer) 
     { 
          cout << "Error occured while mapping" 
                  " view of the file :" << GetLastError() 
               << "\n";
          return false;
     }
     
     return true;
}

Following is the read function which blocks itself on write and read events:

void ReadAndPrint()
{
     cout << "Trying to read and print the shared memory...\n";
     
     //Wait to make sure all writing 
     //operations are done and data is in sync
     bool bContinue = true;
     
     while(bContinue)  //We want to block until we get to read
     {
          cout << "Waiting for write operation to complete...\n";
          
          DWORD dwWaitResult = WaitForSingleObject(g_hWriteEvent, INFINITE );
          
          if (WAIT_OBJECT_0 == dwWaitResult)
          {
               bool bEventFound = false;
               
               for (int ii = 0; ii < MAX_READ_PROCESSES_ALLOWED; ii++)
               {
                    DWORD dwWaitResult = 
                      WaitForSingleObject(g_hReadEvent[ii], WAIT_TIME_OUT);
                    
                    if (WAIT_OBJECT_0 == dwWaitResult)
                    //The state of the specified object is signaled
                    {
                         bEventFound = true;
                         
                         cout << "Setting the Write Event...\n";
                         SetEvent(g_hWriteEvent);
                         
                         //Reading the shared memory
                         cout << "Shared Memory: " 
                              << g_pBuffer << "\n";
                         
                         cout << "Setting the Read Event...\n";
                         SetEvent(g_hReadEvent[ii]);
                         
                         bContinue = false;
                         break; //get out of the for loop
                    }
                    else
                    {
                         continue;
                         //The time-out interval elapsed, 
                         //and the object's state is nonsignaled.
                    }
               } // for
               
               if (false == bEventFound)
               {
                    cout << "Setting the Write Event...\n";
                    SetEvent(g_hWriteEvent);
                    
                    //SwitchToThread();
                    Sleep(WAIT_TIME_OUT);
               }
          }
          else
          {
               cout << "Error occured while waiting :" 
                    << GetLastError() << "\n";
          }
          
     } //while
}

Similarly, the write function uses events to synchronize:

void WriteAndPrint()
{
     cout << "Trying to write and print the shared memory...\n";
     
     cout << "Waiting for write operation to complete...\n";
     
     //Wait to make sure all writing operations are done and data is in sync
     if (WAIT_OBJECT_0 == WaitForSingleObject(g_hWriteEvent, INFINITE ))
     {
          cout << "Waiting for all read operations to complete...\n";
          
          DWORD dwWaitResult = WaitForMultipleObjects( 
               MAX_READ_PROCESSES_ALLOWED,   // number of handles in array
               g_hReadEvent,  // array of read-event handles
               TRUE,         // wait until all are signaled
               INFINITE);    // indefinite wait
          
          if (WAIT_OBJECT_0 == dwWaitResult)
          {
               cout << "Enter a string (without spaces): ";
               
               cin >> g_pBuffer;
               
               //Reading the shared memory
               cout << "Shared Memory: " << g_pBuffer << "\n";
          }
          else
          {
               cout << "Error occured while waiting :" 
                    << GetLastError() << "\n";
          }
          
          cout << "Setting the Write Event...\n";
          SetEvent(g_hWriteEvent);
          
          cout << "Setting the Read Events...\n";
          for (int ii = 0; ii < MAX_READ_PROCESSES_ALLOWED; ii++)
          {
               SetEvent(g_hReadEvent[ii]);
          }
     }
     else
     {
          cout << "Error occured while waiting :" 
               << GetLastError() << "\n";
     }
}

The de-initialization function is shown next. This function takes care of the de-initialization of the shared memory and events.

void DeInitialize()
{
     for (int ii = 0; ii < MAX_READ_PROCESSES_ALLOWED; ii++)
     {
          CloseHandle(g_hReadEvent[ii]);
     }
     
     CloseHandle(g_hWriteEvent);
     
     UnmapViewOfFile(g_pBuffer);
     
     CloseHandle(g_hSharedMemory);
}

Points of Interest

If you are a newbie to events, make sure you understand the difference between manual and auto-reset events. I have used auto-reset events, they are very handy to use.

History

• Feb 2^nd 2006 - Removed a wrong statement - CloseHandle(g_hReadEvent); from Initialize().