queue/test_blocking_q.cpp

   1 // ============================================================================
   2 /// @file  test_blocking_q.cpp
   3 /// @brief Benchmark blocking queue
   4 // ============================================================================
   5
   6
   7 #include <iostream>
   8 #include <glib.h>   // GTimeVal + g_get_current_time
   9 #include <omp.h>    // parallel processing support in gcc
  10 #include "g_blocking_queue.h"
  11
  12 #ifndef N_PRODUCERS
  13 #define N_PRODUCERS         1
  14 #endif
  15
  16 #ifndef N_CONSUMERS
  17 #define N_CONSUMERS         1
  18 #endif
  19
  20 #ifndef N_ITERATIONS
  21 #define N_ITERATIONS 10000000
  22 #endif
  23
  24 #ifndef QUEUE_SIZE
  25 #define QUEUE_SIZE       1000
  26 #endif
  27
  28 void TestBlockingQueue()
  29 {
  30     BlockingQueue<int> theQueue(QUEUE_SIZE);
  31     GTimeVal iniTimestamp;
  32     GTimeVal endTimestamp;
  33
  34     std::cout << "=== Start of testing blocking queue ===" << std::endl;
  35     g_get_current_time(&iniTimestamp);
  36     #pragma omp parallel shared(theQueue) num_threads (2)
  37     {
  38         if (omp_get_thread_num() == 0)
  39         {
  40             if (!omp_get_nested())
  41             {
  42                 std::cerr << "WARNING: Nested parallel regions not supported. Working threads might have unexpected behaviour" << std::endl;
  43                 std::cerr << "Are you running with \"OMP_NESTED=TRUE\"??" << std::endl;
  44             }
  45         }
  46
  47         #pragma omp sections //nowait
  48         {
  49             #pragma omp section
  50             {
  51                 // producer section
  52                 #pragma omp parallel shared(theQueue) num_threads (N_PRODUCERS)
  53                 {
  54                     int i;
  55                     #pragma omp for schedule(static) private(i) nowait
  56                     for (i = 0 ; i < N_ITERATIONS ; i++)
  57                     {
  58                         while(!theQueue.Push(i))
  59                         {
  60                             // queue full
  61                         }
  62                     }
  63                 }
  64             }
  65
  66             #pragma omp section
  67             {
  68                 // consumer section
  69                 #pragma omp parallel shared(theQueue) num_threads (N_CONSUMERS)
  70                 {
  71                     int i;
  72                     int result;
  73                     #pragma omp for schedule(static) private(i, result) nowait
  74                     for (i = 0 ; i < N_ITERATIONS ; i++)
  75                     {
  76                         // this call will block if the queue is empty until
  77                         // some other thread pushes something into it
  78                         theQueue.Pop(result);
  79                     }
  80                 }
  81             }
  82         }
  83
  84     } // #pragma omp parallel
  85
  86     g_get_current_time(&endTimestamp);
  87
  88     // calculate elapsed time
  89     GTimeVal elapsedTime;
  90     if (endTimestamp.tv_usec >= iniTimestamp.tv_usec)
  91     {
  92         elapsedTime.tv_sec  = endTimestamp.tv_sec  - iniTimestamp.tv_sec;
  93         elapsedTime.tv_usec = endTimestamp.tv_usec - iniTimestamp.tv_usec;
  94     }
  95     else
  96     {
  97         elapsedTime.tv_sec  = endTimestamp.tv_sec  - iniTimestamp.tv_sec - 1;
  98         elapsedTime.tv_usec = G_USEC_PER_SEC + endTimestamp.tv_usec - iniTimestamp.tv_usec;
  99     }
 100
 101     std::cout << "Elapsed: " << elapsedTime.tv_sec << "." << elapsedTime.tv_usec << std::endl;
 102     std::cout << "=== End of testing blocking queue ===" << std::endl;
 103 }
 104
 105 int main(int /*argc*/, char** /*argv*/)
 106 {
 107     TestBlockingQueue();
 108     std::cout << "Done!!!" << std::endl;
 109
 110         return 0;
 111 }
 112