Re: WaitForSingleObject

From:

"Larry" <dontmewithme@got.it>

Newsgroups:

microsoft.public.vc.language

Date:

Thu, 28 Jan 2010 22:03:09 +0100

Message-ID:

<4b61fb8c$0$1138$4fafbaef@reader1.news.tin.it>

"Mihajlo Cvetanovi??" <mcvetanovic@gmail> ha scritto nel messaggio
news:ufaFTmEoKHA.1552@TK2MSFTNGP04.phx.gbl...

First, you need only one critical section, and its name should relate to
the data it guards, not to the entities that use it:

CRITICAL_SECTION csMySharedData;

Your current solution does not prevent one consumer and one producer to
access the same data at the same time.

Second, the life of critical section should be the same as the useful life
of data it's supposed to guard. You're safe to call DeleteCriticalSection
only when you know for sure that no one will access data anymore. This is
tricky to implement. You must wait for all spawned threads to finish, and
delete CS after that.

Perfect! I totally got that now. Also, I have made some changes to the
previous code so that it should make more sense (it is working!)

/*
*
* Streaming Server v1.1 by THEARTOFWEB Software
*
*/

#include <iostream>
#include <string>
#include <map>
#include <algorithm>
#include <process.h>
#include <cstdlib>
#include <ctime>
#include "socket.h"
#include <boost/circular_buffer.hpp>
using namespace std;
using namespace boost;

const string CRLF = "\r\n";
const int numbuff = 3;

unsigned int __stdcall Consumer(void* sock);
unsigned int __stdcall Producer(void*);

void getDateTime(char * szTime);

enum buffer_status
{
BUFF_DONE = 1,
BUFF_EMPTY = 0
};

struct buffer
{
unsigned char data[1024];
int bytesRecorded;
int flag;
buffer(const unsigned char * data_, const int bytesRecorded_, const int
flag_) :
  bytesRecorded(bytesRecorded_), flag(flag_)
  {
   copy(data_, data_ + bytesRecorded_, data);
  }
};

struct circular
{
circular_buffer<buffer> cb;
};

// Global maps

map<int, circular> users;
CRITICAL_SECTION users_mutex;

map<int, HANDLE> eventi;
CRITICAL_SECTION eventi_mutex;

int main()
{
// Initialize all critical sections
InitializeCriticalSection(&users_mutex);
InitializeCriticalSection(&eventi_mutex);

// Launch Producer Thread
unsigned int prodRet;
_beginthreadex(0,0,Producer,NULL,0,&prodRet);
if(prodRet)
  cout << "Launched Producer Thread!" << endl;

// Server.
// Set up server (port: 8000, maxconn: 10)
//
SocketServer sockIn(8000, 10);

while(1)
{
  // ...wait for incoming connections...
  Socket* s = sockIn.Accept();

  // Spawn a new Consumer Thread each
  // time a client connects.
  unsigned int sockRet;
  _beginthreadex(0,0,Consumer,s,0,&sockRet);
  if(sockRet)
   cout << "Spawned a new thread!" << endl;

}

DeleteCriticalSection(&users_mutex);
DeleteCriticalSection(&eventi_mutex);

sockIn.Close();

return EXIT_SUCCESS;
}

// Consumer Thread
unsigned int __stdcall Consumer(void* sock)
{
Socket* s = (Socket*) sock;

s->SendBytes("Hello World!" + CRLF);

int threadid = (int)GetCurrentThreadId();

// Create Event && Push it in the event map
HANDLE hevent = CreateEvent(NULL,FALSE,FALSE,NULL);

EnterCriticalSection(&eventi_mutex);
eventi.insert(make_pair(threadid,hevent));
LeaveCriticalSection(&eventi_mutex);

// Prepare && Add circular buffer to the map
circular c;
c.cb.set_capacity(numbuff);

for(int i = 0; i<numbuff; i++)
{
  c.cb.push_back(buffer(NULL,0,BUFF_EMPTY));
}

EnterCriticalSection(&users_mutex);
users.insert(make_pair(threadid, c));
LeaveCriticalSection(&users_mutex);

//
// Read data from the buffer
// and send it to the client
//
// When using push_back the oldest
// element in the circular buffer
// will be in the index 0
//

Sleep(500);

while(1)
{
  // CALLBACK EVENT
  WaitForSingleObject(eventi[threadid], INFINITE);

  if(users[threadid].cb.at(0).flag == BUFF_DONE)
  {
   string line = (char*)users[threadid].cb.at(0).data;
   int ret = s->SendBytes(line + CRLF);
   if(SOCKET_ERROR == ret)
    break;
  }
}

// Close & remove event from event map
CloseHandle(eventi[threadid]);

EnterCriticalSection(&eventi_mutex);
eventi.erase(threadid);
LeaveCriticalSection(&eventi_mutex);

// Remove buffer from the users map
EnterCriticalSection(&users_mutex);
users.erase(threadid);
LeaveCriticalSection(&users_mutex);

// Say bye to the client
s->SendBytes("Bye bye!" + CRLF);

// Disconnect client
cout << "Closing thread..." << endl;

s->Close();
delete s;
return 0;
}

// Producer Thread
unsigned int __stdcall Producer(void*)
{
while(1)
{
  Sleep(1000);
  char szTime[30]; getDateTime(szTime);
  map<int, circular>::iterator uit;

  EnterCriticalSection(&users_mutex);
  EnterCriticalSection(&eventi_mutex);

  for(uit=users.begin(); uit!=users.end(); ++uit)
  {
   users[uit->first].cb.push_back(buffer((unsigned char*)szTime, 30,
BUFF_DONE));

   if(eventi[uit->first])
    SetEvent(eventi[uit->first]);

   cout << "Producer is writing to: " << uit->first << endl;
  }

  LeaveCriticalSection(&eventi_mutex);
  LeaveCriticalSection(&users_mutex);
}
return 0;
}

void getDateTime(char * szTime)
{
time_t rawtime = time(NULL);
struct tm timeinfo;
gmtime_s(&timeinfo, &rawtime);
strftime(szTime, 30, "%a, %d %b %Y %X GMT", &timeinfo);
}

thanks