Search  

---

C:/temp/src/j2k/nto/bakJuly/PCBuffer.cpp

Go to the documentation of this file.

#ifndef PCBuffer_CPP
#define PCBuffer_CPP

#include <math.h>
#include <iostream.h>

#include "PCBuffer.hpp"
#include "Basic_PThread.cpp"
#include "Mutex.cpp"
#include "RWLock.cpp"
#include "Semaphore.cpp"
#include "BoundedBuffer.cpp"
#include "Timer.cpp"
#include "TimerPulse.cpp"

#define ADD      ( 0.500000000  )
#define WAITING  (   500000000  )
#define BILLION  (  1000000000  )

#define RANDOM    250
#define TSEC      0

#define TT Timer

class Timing : public TT {
private:
  ULONG   res;
  double  current;
  double  add;
public:

 Timing( ULONG clk = 0 )
  : TT( clk, TSEC ), current( 0.0f )
  {
    res = clk;
  }

 virtual void start() 
 {
   fixprio( 30 );
   register double r = (double)res * 1000 / BILLION + TSEC;
   QBColor();
   printf( "\nTimer started with a resolution of %6.2f ms.", r );
   printf( "\nEach clock timer ticks are represented by '#'\n" );
   fflush( stdout );


   TT::start();
 }

 virtual void timedRun( int signo )
 {
    current += ADD;
    printf( "#" );
 }
 
 inline double getTime() { return current; }

};

Timing timer( WAITING );

/***********************************************************************
 ***********************************************************************
 ***                                                                 ***
 ***  message:       Notify that a message is now complete           ***
 ***                                                                 ***
 ***  slot:          Is there any available slot left ?              ***
 ***                                                                 ***
 ***  receiver:      Make sure that only one reader is accessing     ***
 ***                 the buffer at any given time.                   ***
 ***                                                                 ***
 ***  sender:        Make sure that only one writer is modifying     ***
 ***                 the buffer at any given time.                   ***
 ***                                                                 ***
 ***********************************************************************
 ***********************************************************************/

Producer::Producer( int n, int i ) 
   : Basic_PThread(), item( i ), name( n ), 
     started( 0 ), elapsed( 0.0 ), total( 0.0 ) { }
 
void Producer::run() 
   {
     QBColor();
     printf( "\nProducer %u started. ", name );
     fixprio( 8 );
     sched_yield();

     for(;;) 
     {
       if ( ++item > STEPS ) break;

       register size_t where = item % BBSIZE;

       slot[ where ]->P();  // Slot is free ?
       sender.P();          // Take write control

       started = timer.getCycles();

       boundedBuffer.enqueue( item );

       QBColor();
       printf( 
         "\nProducer %u produced component %2u in slot %2u. ",
         name, item, where
       );

       elapsed  = timer.getElapsed( started ); 
       total   += elapsed;

       elapsed *= 1000;
       printf( "\nProducer %u took %4.2f ms. ", name, elapsed );
       fflush( stdout );

       sender.V();    // Another sender can have control
       message.V();   // One message completed

     } // End of infinite loop

     QBColor( 12 );
     printf( "\nProducer %u terminates. Total time spent %6.2f ms. \n", name, total*1000 );
     QBColor();
     fflush( stdout );
     ++done;
   }
 
Consumer::Consumer( int n ) 
   : Basic_PThread(), name( n ), slotno( 0 ),
     started( 0 ), elapsed( 0.0 ), total( 0.0 ) { }

void Consumer::run() 
   {
     printf( "\nConsumer %u started. ", name );
     fixprio( 8 );
     sched_yield();

     while( item < STEPS )
     {
       message.P();                // Is there a message ?

       if ( item > STEPS ) break;  // Program's done

       receiver.P();               // Receiver take control

       started = timer.getCycles();

       ++item;

       char msg = boundedBuffer.dequeue(); // Consumed

       register size_t where = item % BBSIZE;

       printf( 
         "\nConsumer %u took message %2u from slot %2u in step %2u. ",
         name, msg, where, item
       );
       fflush( stdout );

       elapsed  = timer.getElapsed( started ); 
       total   += elapsed;

       elapsed *= 1000;
       printf( "\nConsumer %u took %4.2f ms. ", name, elapsed );
       fflush( stdout );

       slot[ where ]->V();  // Free the slot
       receiver.V();        // Release control
       sched_yield();

     } // End of infinite loop

     QBColor( 12 ); 
     printf( "\nConsumer %u terminates. Total time spent %6.2f ms. \n", name, total*1000 );
     QBColor();
     fflush( stdout );

     if ( item == STEPS ) {
       ++item; 
       message.V();  // Ask the other one to quit, if last step
     }

     ++done;
   }


int main() {
   NoColor = 1;

   slot = new Semaphore*[ BBSIZE+1 ];

   for( int k = 0; k <= BBSIZE; k++ )
   {
     char buf[80]; 
     sprintf( buf, "slot%d", k );

     int c = (9+k)%15;

     if ( c < 8 ) c += 10; 
     slot[k] = new Semaphore( 1, buf, c );
   }

   fixprio( 10 );

   Producer p1( 1 );
   Consumer c1( 1 );
   Consumer c2( 2 );

   p1.start( 10 );
   c1.start( 10 );
   c2.start( 10 );

   timer.start();
   _uint64 prgStart = timer.getCycles();

   p1.join();
   c1.join();
   c2.join();

   fixprio( 6 );
   sched_yield();

   int i = 0;
   while( done < 3 )
   {
      if ( ++i > 10 ) exit( 0 );
      printf( "M " );
      sched_yield();
   }

   QBColor( 14 );
   printf( "Total program took %6.2f ms. Last time the timer was triggered %6.2f ms \n",
           (double)timer.getElapsed( prgStart ) * 1000, timer.getTime() * 1000 );

   QBColor(); 
   printf( "\n\nDone!\n");

   delete slot;

   exit( 0 );
   return 0;
}

#endif

---