🚀 go-pugleaf

Colin B Maharaj wrote:

>How to use...(actual code is below)...

Nice code.
I'd be very tempted to replace tmps with
buffer[rear].s which would eliminate the
memcpy2, speeding up the critical section.
It would also eliminate what appears to be a bug:

Thread 1 does GetElement() which copies
data into tmps and then copies tmps into
the target buffer. While Thread 1 is
copying tmps twice (in a critical section!),
Thread 2 fills tmps (outside a critical section)
to issue a SetElement().

I think "read_ buffer" and "write_buffer"
property wrappers would help.
(following is without properties)

{Alternative would be read_bufer and write_buffer
pointers which are set to point into buffers[].)

>void __fastcall TDataModule1::UDPServer1UDPRead(TObject *Sender,
>       TStream *AData, TIdSocketHandle *ABinding)
>{
>   if (Buff1) // does our object exists?
>   {
     int len =  AData->Size; // in case Size changes
// Will len > PACKETSIZE ever happen ?
     *Buff1->write_size() = 0;// existing data is obsolete
     AData->ReadBuffer(Buff1->write_buffer(), len);
     *Buff1->write_size() = len;// data is available
     Buff1->Add();
>   }
>}

char* TCircBuff::write_buffer()
{	return buffer[rear].s;
}
int* TCircBuff::write_size()
{	return &buffer[rear].l;
}

// Single-writer version
//void TCircBuff::SetElement(unsigned char *ss, int ll)
void TCircBuff::Add()
>{
   // two fewer buffers available this way
   // [front] is being consumed, [rear] is being written
   // if we use the multi-consumer below, then -1 is fine
   if (count < TOTALPACKETS - 2 )
>   {
      EnterCriticalSection(&GlobalCriticalSection);
	// might need to list count as volatile.
>     count++;
      LeaveCriticalSection(&GlobalCriticalSection);
>     rear++;
>     if (rear == TOTALPACKETS)
>     {
>       rear = 0;
>     }
      //write_buffer = buffer[rear].s;
>   }
>}

// Single-consumer version
// Note difference in ss parameter
void TCircBuff::GetElement(unsigned char **ss, int & ll)
>{
>   if( count>0 )
>   {
      //read_buffer = buffer[front].s;
      *ss = buffer[front].s;
      ll = buffer[front].l;
      EnterCriticalSection(&GlobalCriticalSection);
>     count--;
      LeaveCriticalSection(&GlobalCriticalSection);
>     front++;
>     if (front == TOTALPACKETS) front=0;
>   }
>   else
>   {
     *ss = 0;
     ll=0;
>   }
>}

One of the goals of circular buffers is to eliminate
temporary buffers. The data is entered and consumed
directly to/from the internal buffers.
It is also a goal that buffer[front] != buffer[rear]
so that there is no collision requiring a critical
section on the buffers in the first place.

Critical sections should be short.
Since tmps is available to the application outside
a critical section, placing the tmps copy code inside
a critical section is meaningless.

Ok, so if you have multiple consumer threads, then
you need to take the data out of the buffer during
the "pop". Just don't use the vulnerable tmps.

// Multi-consumer version
void TCircBuff::GetElement(unsigned char *ss, int & ll)
>{
>   EnterCriticalSection(&GlobalCriticalSection);
>   if (count>0)
>   {
      memcpy2(ss, buffer[front].s, PACKETSIZE);
      ll = buffer[front].l;
>     count--;
>     front++;
>     if (front == TOTALPACKETS) front=0;
      LeaveCriticalSection(&GlobalCriticalSection);
>   }
>   else
>   {
     LeaveCriticalSection(&GlobalCriticalSection);
     ZeroMemory(ss, PACKETSIZE);
     ll=0;
>   }
>}

And just to be complete:
(AddTmp and tmps don't exist for multi-writer)

// Multi-writer version
void TCircBuff::SetElement(unsigned char *ss, int ll)
>{
/*	FudgeFactor
	= 2 for single-single
	= 1 for mixed
	=0 for multi/multi
*/
>  EnterCriticalSection(&GlobalCriticalSection);
   if (count < TOTALPACKETS /* - FudgeFactor */ )
>  {
>     memcpy2(buffer[rear].s, ss,PACKETSIZE);
>     buffer[rear].l = ll;
>     count++;
>     rear++;
>     if (rear == TOTALPACKETS)
>     {
>       rear = 0;
>     }
   }else{
	//throw EDroppedPacket
>  };
>  LeaveCriticalSection(&GlobalCriticalSection);
>}

In summary, the multi-writer/multi-reader is the general-purpose/cover all bases version,
and as such, is slower with all those memcpy functions.
This overhead might be aleviated by giving each thread a buffer and just swapping
pointers.

// Multi-writer pointer version
// (again, **ss)
void TCircBuff::SetElement(unsigned char **ss, int ll)
{
   EnterCriticalSection(&GlobalCriticalSection);
   if (count < TOTALPACKETS )
   {
	swap( buffer[rear].s, *ss );//might be wrong
	// places the thread's buffer into our array,
	// and gives the thread our empty buffer
>     buffer[rear].l = ll;
>     count++;
>     rear++;
>     if (rear == TOTALPACKETS)
>     {
>       rear = 0;
>     }
   }else{
	//throw EDroppedPacket
>  };
>  LeaveCriticalSection(&GlobalCriticalSection);
>}



{original code not snipped}
>#include "CircBuff.h"
>TCircBuff *Buff1 = new TCircBuff; // Declaration....
>
>//---------------------------------------------------------------------------
>// In Server Code - UDP Reader
>//---------------------------------------------------------------------------
>
>void __fastcall TDataModule1::UDPServer1UDPRead(TObject *Sender,
>       TStream *AData, TIdSocketHandle *ABinding)
>{
>   if (Buff1) // does our object exists?
>   {
>     Buff1->tmpl = AData->Size;
>     AData->ReadBuffer(Buff1->tmps, Buff1->tmpl);
>     Buff1->AddTmp();
>   }
>}
>//---------------------------------------------------------------------------
>// In Consumer Code, e.g. A Form with a Timer and a Memo
>//---------------------------------------------------------------------------
>
>void __fastcall TForm1::Timer1Timer(TObject *Sender)
>{
>   unsigned char s[4096]; // you may want to declare this in a heap...
>   int len;
>   Timer1->Enabled = false;
>   if (Buff1->Count>0)
>   {
>      Memo1->Lines->Add("----------------------");
>      Buff1->GetElement(s, len);
>      AnsiString B = s;
>      Memo1->Lines->Add(B); // place in a memo or something...
>   }
>   Timer1->Enabled = true;
>}
>==================================CIRCBUFF.H================================
>
>#ifndef CircBuffH
>#define CircBuffH
>
>// You may need to change these accordingly
>
>#define TOTALPACKETS 512
>#define PACKETSIZE   4096
>struct packetstruct
>{
>    unsigned char *s;
>    int l;
>};
>
>class TCircBuff
>{
>private:
>    volatile int front, rear, count;
>    packetstruct * buffer;
>
>    CRITICAL_SECTION GlobalCriticalSection;
>    void memcpy2(unsigned char *d, unsigned char *s, int l);
>
>public:
>   __fastcall TCircBuff();
>   __fastcall ~TCircBuff();
>
>   char *tmps;
>   int  tmpl;
>
>   void GetElement(unsigned char *ss, int & ll);
>   void SetElement(unsigned char *ss, int ll);
>   int  FGetCount(void) { return count; };
>   void AddTmp();
>
>   __property int Count = { read=FGetCount  };
>};
>#endif
>
>=============================end-of-CIRCBUFF.H===============================
>
>================================CIRCBUFF.CPP=================================
>
>#include <vcl.h>
>#pragma hdrstop
>
>#include "CircBuff.h"
>
>//---------------------------------------------------------------------------
>#pragma package(smart_init)
>//---------------------------------------------------------------------------
>
>__fastcall TCircBuff::TCircBuff()
>{
>    buffer = new packetstruct [TOTALPACKETS];
>    tmps = new unsigned char [PACKETSIZE];
>    for (int i=0; i<TOTALPACKETS; i++)
>    {
>       buffer[i].s = new unsigned char [PACKETSIZE];
>       buffer[i].l = 0;
>    }
>    front=0;
>    rear=0;
>    count=0;
>    InitializeCriticalSection(&GlobalCriticalSection);
>}
>
>//---------------------------------------------------------------------------
>
>__fastcall TCircBuff::~TCircBuff()
>{
>    for (int i=0; i<TOTALPACKETS; i++)
>    {
>       delete []buffer[i].s;
>    }
>   delete []buffer;
>   delete []tmps;
>   DeleteCriticalSection(&GlobalCriticalSection);
>}
>
>//---------------------------------------------------------------------------
>
>void TCircBuff::GetElement(unsigned char *ss, int & ll)
>{
>   EnterCriticalSection(&GlobalCriticalSection);
>   if (count>0)
>   {
>     memcpy2(tmps, buffer[front].s, PACKETSIZE);
>     tmpl = buffer[front].l;
>     count--;
>     front++;
>     if (front == TOTALPACKETS) front=0;
>   }
>   else
>   {
>     ZeroMemory(tmps, PACKETSIZE);
>     tmpl=0;
>   }
>   memcpy2(ss, tmps, PACKETSIZE);
>   ll = tmpl;
>   LeaveCriticalSection(&GlobalCriticalSection);
>}
>
>//---------------------------------------------------------------------------
>
>void TCircBuff::SetElement(unsigned char *ss, int ll)
>{
>   EnterCriticalSection(&GlobalCriticalSection);
>   memcpy2(buffer[rear].s, ss,PACKETSIZE);
>   buffer[rear].l = ll;
>   if (count < TOTALPACKETS)
>   {
>     count++;
>     rear++;
>     if (rear == TOTALPACKETS)
>     {
>       rear = 0;
>     }
>   }
>   LeaveCriticalSection(&GlobalCriticalSection);
>}
>
>//---------------------------------------------------------------------------
>
>void TCircBuff::AddTmp()
>{
>   SetElement(tmps, tmpl);
>}
>
>//---------------------------------------------------------------------------
>
>void TCircBuff::memcpy2(unsigned char *d, unsigned char *s, int l)
>{
>    if (!d) return;
>    if (!s) return;
>    while (l--) *d++ = *s++;
>}
>
>//---------------------------------------------------------------------------
>
>==============================end-of-CIRCBUFF.CPP============================