C / C++ / MFC

Rick York wrote:

I did something almost exactly like this in a different context and it worked fine. In my case the buffers were rather large and took a while to fill so it was easy to manage them but your case might be different. You may need to have more than one buffer if the processing thread is not able to always keep up and you don't want the capture thread to wait so you may want to implement your code to support N buffers and then determine what the optimum value of N is later during testing. It may be 2 or it could be 200 or it could vary widely depending on many factors. I don't know nearly enough about what you are doing to predict.

Good Luck.

BTW - I read further replies in this thread and I want to clarify one thing. My implementation has multiple producers (max of about 10) and one consumer and each producer has a double buffer. I could have had one consumer for each producer but the buffers are very big and fill so slowly that this was unnecessary. My application is not a web server but its scaling requirements are very well known to us.

Samuel Zhao wrote:

How to implement the synchronization by using CRITICAL_SECTION/EVENT or other kernal objects to get a highest performance?

#ifndef DOUBLE_BUFFER_H
#define DOUBLE_BUFFER_H

#include <afx.h>
#include <afxwin.h>

class CDoubleBuffer
{
public:
   CDoubleBuffer( unsigned int unAlloc = 0x000FFFFF );
   ~CDoubleBuffer(void);

   void Write(void* pBuf, unsigned int unBytesTo);
   void Read (void* pBuf, unsigned int unBytesFrom);

private:
   void** pAlloc;
   unsigned int unRead;
   unsigned int unWrite;
   unsigned int unSize;
};

#endif // ! defined (DOUBLE_BUFFER_H)

#include "stdafx.h"
#include "DoubleBuffer.h"

CDoubleBuffer::CDoubleBuffer( unsigned int unAlloc /* = 0x000FFFFF */ )
{
   pAlloc = NULL;
   pAlloc = (void **) ::HeapAlloc(::GetProcessHeap(), 0, 2 * sizeof(void*));
   if (!pAlloc) throw;
   pAlloc[0] = NULL;
   pAlloc[0] = (void *) ::HeapAlloc(::GetProcessHeap(), 0, unAlloc * sizeof(BYTE));
   if (!pAlloc[0]) throw;
   pAlloc[1] = NULL;
   pAlloc[1] = (void *) ::HeapAlloc(::GetProcessHeap(), 0, unAlloc * sizeof(BYTE));
   if (!pAlloc[1]) throw;
   ::FillMemory((void*) pAlloc[0], unAlloc, 0);
   ::FillMemory((void*) pAlloc[1], unAlloc, 0);

   unRead  = 0;
   unWrite = 0;
   unSize = unAlloc;
}

CDoubleBuffer::~CDoubleBuffer(void)
{
   ::HeapFree(::GetProcessHeap(), 0, pAlloc[0]);
   ::HeapFree(::GetProcessHeap(), 0, pAlloc[1]);
   ::HeapFree(::GetProcessHeap(), 0, pAlloc);
}

void CDoubleBuffer::Write(void* pBuf, unsigned int unBytesTo)
{
   unsigned int unTryWrite = (unWrite++)%2;
   while( unRead == unTryWrite ) { ::Sleep(10); }
   ::MoveMemory( pAlloc[unTryWrite], pBuf, __min(unBytesTo,unSize) );
   unWrite = unTryWrite;
}

void CDoubleBuffer::Read(void* pBuf, unsigned int unBytesFrom)
{
   while( unRead == unWrite ) { ::Sleep(10); }
   unsigned int unTryRead = (unRead++)%2;
   ::MoveMemory(pBuf, pAlloc[unTryRead], __min(unBytesFrom,unSize));
   unRead = unTryRead;
}

// Light Weight Circular Buffer with elementary locking

#ifndef CBUFFER_H
#define CBUFFER_H

#if _MSC_VER > 1000
#pragma warning (disable: 4786)
#pragma warning (disable: 4748)
#pragma warning (disable: 4103)
#endif /* _MSC_VER > 1000 */

#include <afx.h>
#include <afxwin.h>

// Default number of elements in buffer
#define CBUFFER_NELEM		      0x0000FFFF

// The Circular Buffer errors
#define CBUFFER_ERRCODE_OK               0
#define CBUFFER_ERRCODE_NO_MEMORY        1
#define CBUFFER_ERRCODE_READ_ERROR       2
#define CBUFFER_ERRCODE_WRITE_ERROR      3
#define CBUFFER_ERRCODE_NO_DATA          4

#define CBUFFER_NULL_PTR                 0xFFFFFFFF

class CCircBuffer
{
// Constructors, Destructor
public:
   CCircBuffer(UINT32 unElements = CBUFFER_NELEM);
   ~CCircBuffer(void);

// Operations
public:
   // Read  the data
   UINT32 GetTheData(void*& lpData);
   // Write the data
   UINT32 SetTheData(void*  lpData);
   // Increment the read pointer
   UINT32 IncTheReadPtr(void);
   // Increment the write pointer
   UINT32 IncTheWritePtr(void);
   // Check if there is any space for reading data
   bool IsReadyForRead(void);
   // Check if there is any space for writing new data
   bool IsReadyForWrite(void);
   // Get the last known value for the read pointer
   UINT32 GetLastReadPtr(void);
   // Get the last known value for the write pointer
   UINT32 GetLastWritePtr(void);

// Attributes
private:
   // The Circular Buffer of void pointers
   void**  TheCircularBuffer;
   SIZE_T unElems;
   // The Circular Buffer pointers
   UINT32 unReadPtr, unWritePtr;
   // The Circular Buffer last error
   UINT32 unLastErrorCode;
};

#endif /* ! defined(CBUFFER_H) */

#include "StdAfx.h"
#include "CircBuffer.h"

CCircBuffer::CCircBuffer(UINT32 unElements /* = CBUFFER_NELEM */)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Allocate the Buffer
   unElems = (SIZE_T) unElements;
   if ( unElems >= CBUFFER_NULL_PTR ) // max size
      unElems = CBUFFER_NULL_PTR - 1;
   TheCircularBuffer = (void **) HeapAlloc(GetProcessHeap(), 0, unElems * sizeof(void*));
   // Cannot allocate memory
   if(!TheCircularBuffer)
   {
      unLastErrorCode = CBUFFER_ERRCODE_NO_MEMORY;
   }
   else
   {
      SecureZeroMemory(TheCircularBuffer, sizeof(TheCircularBuffer));
   }
   // Set Pointers
   unReadPtr = unWritePtr = 0;
}

CCircBuffer::~CCircBuffer(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   BOOL bResult = HeapFree(GetProcessHeap(), 0, TheCircularBuffer);
   if( bResult != TRUE )
   {
      unLastErrorCode = CBUFFER_ERRCODE_NO_MEMORY;
   }
}

// Operations

// Read the data
UINT32 CCircBuffer::GetTheData(void*& lpData)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Check if there is any data to be read in the buffer
   if( ! IsReadyForRead() )
   {
      // No data to be read in buffer
      unLastErrorCode = CBUFFER_ERRCODE_NO_DATA;
      return CBUFFER_NULL_PTR;
   }
   // Get the data
   lpData = (void*) TheCircularBuffer[unReadPtr];
   // Increment the read pointer
   unReadPtr++;
   if( unReadPtr >= unElems )
      unReadPtr = 0;	// Restart
   return unReadPtr;
}

// Write the data
UINT32 CCircBuffer::SetTheData(void* lpData)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Check if there is any way of writing the data
   if( ! IsReadyForWrite() )
   {
      // No way to write data in Buffer
      unLastErrorCode = CBUFFER_ERRCODE_NO_DATA;
      return CBUFFER_NULL_PTR;
   }
   // Set the data
   TheCircularBuffer[unWritePtr] = (void*) lpData;
   // Increment the write pointer
   unWritePtr++;
   if( unWritePtr >= unElems )
      unWritePtr = 0; // Restart
   return unWritePtr;
}

// Increment the read pointer
UINT32 CCircBuffer::IncTheReadPtr(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Check if there is any data to be read in the buffer
   if ( ! IsReadyForRead() )
   {
      // No data available to be read
      unLastErrorCode = CBUFFER_ERRCODE_NO_DATA;
      return CBUFFER_NULL_PTR;
   }
   // Increment the read pointer
   unReadPtr++;
   if( unReadPtr >= unElems )
      unReadPtr = 0; // restart
   return unReadPtr;
}

// Increment the write pointer
UINT32 CCircBuffer::IncTheWritePtr(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Checks if there is any way of writing the data
   if( ! IsReadyForWrite() )
   {
      // No way to write data
      unLastErrorCode = CBUFFER_ERRCODE_NO_DATA;
      return CBUFFER_NULL_PTR;
   }
   // Increment the write pointer
   unWritePtr++;
   if( unWritePtr >= unElems )
      unWritePtr = 0; // restart
   return unWritePtr;
}

// Check if there is any way of reading the data
bool CCircBuffer::IsReadyForRead(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Check if there is any way of reading the data
   // The test pointer
   UINT32 unCheckReadPtr = unReadPtr;
   if( unCheckReadPtr >= unElems )
      unCheckReadPtr = 0;	// Restart
   if( unCheckReadPtr == unWritePtr )
   {
      // No way to read data
      unLastErrorCode = CBUFFER_ERRCODE_NO_DATA;
      return false;
   }
   return true;
}

// Check if there is any space for writing new data
bool CCircBuffer::IsReadyForWrite(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   // Checks if there is any way of writing the data
   // Increment the test pointer
   UINT32 unCheckWritePtr = unWritePtr;
   unCheckWritePtr++;
   if( unCheckWritePtr >= unElems )
      unCheckWritePtr = 0;	// Restart
   if( unCheckWritePtr == unReadPtr )
   {
      // No way to write data
      unLastErrorCode = CBUFFER_ERRCODE_NO_DATA;
      return false;
   }
   return true;
}

// Get the last known value for the read pointer
UINT32 CCircBuffer::GetLastReadPtr(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   return unReadPtr;
}

// Get the last known value for the write pointer
UINT32 CCircBuffer::GetLastWritePtr(void)
{
   unLastErrorCode = CBUFFER_ERRCODE_OK;
   return unWritePtr;
}

// Increment the read pointer
UINT32 IncTheReadPtr(void);
// Increment the write pointer
UINT32 IncTheWritePtr(void);

federico.strati wrote:

#ifndef DOUBLE_BUFFER_H
#define DOUBLE_BUFFER_H

#include <afx.h>
#include <afxwin.h>

class CDoubleBuffer
{
public:
CDoubleBuffer( unsigned int unAlloc = 0x000FFFFF );
~CDoubleBuffer(void);

void Write(void* pBuf, unsigned int unBytesTo);
void Read (void* pBuf, unsigned int unBytesFrom);

private:
void** pAlloc;
unsigned int unRead;
unsigned int unWrite;
unsigned int unSize;
};

#endif // ! defined (DOUBLE_BUFFER_H)

volatile unsigned int unRead;   <br />
volatile unsigned int unWrite;

<br />
void CDoubleBuffer::Write(void* pBuf, unsigned int unBytesTo)<br />
{   <br />
//unsigned int unTryWrite = (unWrite++)%2;<br />
unsigned int unTryWrite = (unWrite+1)%2;<br />
while( unRead == unTryWrite ) { ::Sleep(10); }   <br />
::MoveMemory( pAlloc[unTryWrite], pBuf, __min(unBytesTo,unSize) );   <br />
unWrite = unTryWrite;<br />
}<br />
<br />
void CDoubleBuffer::Read(void* pBuf, unsigned int unBytesFrom)<br />
{   <br />
while( unRead == unWrite ) { ::Sleep(10); }   <br />
//unsigned int unTryRead = (unRead++)%2;<br />
unsigned int unTryRead = (unRead+1)%2;   <br />
::MoveMemory(pBuf, pAlloc[unTryRead], __min(unBytesFrom,unSize));   <br />
unRead = unTryRead;<br />
}<br />

federico.strati wrote:

Note:

1. it is necessary to declare the variables volatile on some systems
to prevent using cached values, you'll be fine by declaring all the
integer indexes unRead/unWrite etc... as volatile

2. the classes are thread safe for the single situation where you have a single
producer (writer) and a single consumer (reader) in different
threads. It is NOT safe for multiple consumers / multiple producers.

3. The modifs you've done are good even if don't understand why
you changed the way you increment the pointers
it is not important to increment pointers in an atomic manner
as there will be only two threads: one writer and one reader.
If you want atomicity then you shall use InterlockedIncrement
/ Decrement functions. But it is a wasted effort here.

Hope that helps

Cheers
Federico

void CDoubleBuffer::Write(void* pBuf, unsigned int unBytesTo)
{
   unsigned int unTryWrite = (unWrite+1)%2;
   while( unRead == unTryWrite ) { ::Sleep(10); }
   // use current unWrite !
   ::CopyMemory( pAlloc[unWrite], pBuf, __min(unBytesTo,unSize) );
   unWrite = unTryWrite;
}

void CDoubleBuffer::Read(void* pBuf, unsigned int unBytesFrom)
{
   while( unRead == unWrite ) { ::Sleep(10); }
   // use current unRead !
   ::CopyMemory(pBuf, pAlloc[unRead], __min(unBytesFrom,unSize));
   unRead = (unRead+1)%2;
}

TCPMem wrote:

What's the best way to find newline in char[]