Search  

---

C:/temp/src/j2k/nto/cs5.cpp

Go to the documentation of this file.

#ifndef __J2K__CS5_CPP__
#define __J2K__CS5_CPP__

#include <j2k/nto/cs_def.hpp>
#include <j2k/nto/Server.hpp>

/*** Client Side of the code ***/
class Consumer 
  : public Basic_PThread,
    public MsgContainer
{
protected:
  int idx;
public:
  Consumer( NameSpace* n0, int idx0 = 0 ) 
    : Basic_PThread(), 
      MsgContainer( n0, client ),
      idx( idx0 )
  { 
    message->setRequest( MQI_CONSUMER );
    message->setIndex( idx );
  }

  inline void setPThread( pthread_t tid )
  {
    printf( "Consumer PID[%d] \n", (int)tid );
    fflush( stdout );

    message->setPThread( tid );
  }

  virtual ~Consumer() { }

  /*** Consumer Side of the code ***/
  virtual void run()
  {
    printf( "Consumer %d run \n", idx );
    fflush( stdout );

    connect();

    printf( "Consumer %d opened a connection \n", idx );
    fflush( stdout );
    n->waitForThreads();

    /* Do whatever work you wanted with server connection */
    register int i = 0;
     
    for ( i = 0; i < 5; i++ ) 
    {
      sched_yield();
      printf( "Consumer Semaphore P \n" );
//    m.P();

        register int data = 1000 * ( idx + 1 ) + i;
        printf("Consumer %d sending %d \n", idx, data );
        fflush( stdout );

        message->setNumber( data );

        if ( message->send( fd, replyMsg ) < 0 )
        {
          perror( "Couldn't send Message ! \n" );
        }
  
        register int status = replyMsg->getStatus();

        printf( "Status[%d] \n", status );
        fflush( stdout );

        if ( replyMsg->isSuicide() )
        {
          printf( "Consumer %d SUICIDE ! \n", idx );
          fflush( stdout );
          i = 100;
        }

        if ( replyMsg->isFull() && replyMsg->isGet() )
        {
          printf( "Consumer %d retrieved %d \n", idx, replyMsg->getNumber() );
          fflush( stdout );
        }

        printf( "Consumer loop end for %d \n", data );
        fflush( stdout );

    } // End of for

    printf( "Consumer disconnect \n" );
    disconnect();
  }
};

/*** Producer Side of the code ***/
class Producer 
  : public Basic_PThread, 
    public MsgContainer
{
protected:
  int idx;
public:
  Producer( NameSpace* n0, int idx0 = 0 )
    : Basic_PThread(),
      MsgContainer( n0, client ),
      idx( idx0 )
  { 
    message->setRequest( MQI_PRODUCER );
    message->setIndex( idx );
  }

  inline void setPThread( pthread_t tid )
  {
    printf( "Producer PID[%d] \n", (int)tid );
    fflush( stdout );

    message->setPThread( tid );
  }

  virtual ~Producer() { }

  /*** Producer Side of the code ***/
  virtual void run()
  {
    printf( "Producer %d run \n", idx );
    fflush( stdout );

    connect();

    printf( "Producer %d opened a connection \n", idx );
    fflush( stdout );
    n->waitForThreads();

    /* Do whatever work you wanted with server connection */
    register int i = 0;
     
    for ( i = 0; i < 5; i++ ) 
    {
      sched_yield();
      printf( "Producer Semaphore P \n" );
//    m.P();

        register int data = 200 * (idx + 1) + i;
        printf("Producer %d sending %d \n", idx, data );
        fflush( stdout );

        message->setNumber( data );

        if ( message->send( fd, replyMsg ) < 0 )
        {
          perror( "Couldn't send Message ! \n" );
        }

        if ( replyMsg->isSuicide() )
        {
          printf( "Producer %d SUICIDE ! \n", idx );
          fflush( stdout );
          i = 100;
        }

        printf( "Producer loop end for %d \n", data );
        fflush( stdout );

    } // End of for

    if ( replyMsg->isEmpty() )
    {
      printf( "Producer %d was consumed by %d \n", idx, replyMsg->getNumber() );
      fflush( stdout );
    }

    printf( "Producer disconnect \n" );
    disconnect();
  }
};

int main( int argc, char **argv ) 
{
  const int nbConsumers = 2;
  const int nbProducers = 1;
  const int nbServers   = 1;

  NameSpace* n = new NameSpace( "coen320_fred", nbConsumers, nbProducers, nbServers  );
  Server s( n );

  Producer* p[ nbProducers ];
  Consumer* c[ nbConsumers ];

  s.start();
  setprio( s.getPID(), 20 );

  n->waitForServer();

  setprio( pthread_self(), 19 );

  register int i = 0;
  for( i = 0; i < nbProducers; i++ )
  {
    p[i] = new Producer( n, i+100 );
    p[i]->start();
    p[i]->setPThread( p[i]->getPID() );
    setprio( p[i]->getPID(), 18 );
  }

  for( i = 0; i < nbConsumers; i++ )
  {
    c[i] = new Consumer( n, i+100 );
    c[i]->start();
    c[i]->setPThread( c[i]->getPID() );
    setprio( c[i]->getPID(), 18 );
  }

  n->waitForThreads();

//  pthread_join( c.getPID(), NULL );

  n->waitForClient();

  printf( "Main Done \n" );
  fflush( stdout );

  // Shut down the READ-Blocked server
  printf( "Shut down the READ-Blocked server... \n" );
  fflush( stdout );

  pthread_kill( s.getPID(), SIGKILL );

  if ( n != NULL ) delete n;

  return 0;
}

#endif // End of CS4.cpp

---