1 /*

     2 * Copyright (c) 2010 Nokia Corporation and/or its subsidiary(-ies).

     3 * All rights reserved.

     4 * This component and the accompanying materials are made available

     5 * under the terms of "Eclipse Public License v1.0"

     6 * which accompanies this distribution, and is available

     7 * at the URL "http://www.eclipse.org/legal/epl-v10.html".

     8 *

     9 * Initial Contributors:

    10 * Nokia Corporation - initial contribution.

    11 *

    12 * Contributors:

    13 *

    14 * Description: 

    15 *

    16 */

    17 

    18 #include <unistd.h>

    19 #include <stdio.h>

    20 #include <stdlib.h>

    21 #include <string.h>

    22 

    23 #include <list>

    24 

    25 #include "itk.h"

    26 

    27 #include "cpixsynctools.h"

    28 #include "cpixsyncqueue.h"

    29 

    30 

    31 #define LENIENT_STR "#IGNORE "

    32 

    33 #define MSG(fmt, args...)  printf(fmt, ##args); fflush(stdout);

    34 

    35 

    36 typedef Cpt::SyncQueue<int>    IntQueue;

    37 

    38 

    39 struct ProducerParam

    40 {

    41     IntQueue                 * queue_;

    42     int                        maxCount_;

    43     int                        maxSleepMs_;

    44     Itk::TestMgr             * testMgr_;

    45 };

    46 

    47 

    48 void * producer(void * param)

    49 {

    50     struct ProducerParam

    51         const * p = reinterpret_cast<ProducerParam const*>(param);

    52 

    53     int

    54         ms = rand();

    55     ms = ms % p->maxSleepMs_;

    56 

    57     enum 

    58     { 

    59         MSEC_PER_SEC = 1000,

    60         NSEC_PER_MSEC = 1000000

    61     };

    62 

    63     struct timespec 

    64         tim;

    65     tim.tv_sec = ms / MSEC_PER_SEC;

    66     tim.tv_nsec = ms % MSEC_PER_SEC * NSEC_PER_MSEC;

    67             

    68     MSG("[B:%d] ",

    69         ms);

    70 

    71     try

    72         {

    73             bool

    74                 goOn = true;

    75 

    76             for (int i = 0; 

    77                  i <= p->maxCount_ && goOn; 

    78                  ++i)

    79                 {

    80                     

    81                     nanosleep(&tim,

    82                               NULL);

    83                     goOn = p->queue_->put(i);

    84                     MSG(">%d ",

    85                            i);

    86                     ITK_DBGMSG(p->testMgr_, ":");

    87                 }

    88         }

    89     catch (Cpt::SyncExc & exc)

    90         {

    91             // we should not get exceptions

    92             MSG("\nSync exception in producer ");

    93         }

    94 

    95     MSG("\n" LENIENT_STR "[E] ");

    96 

    97     return NULL;

    98 }

    99 

   100 

   101 /**

   102  * A simple integer threads can add to in a synchronized way.

   103  */

   104 class SyncSum : public Cpt::SyncEntity

   105 {

   106     int        sum_;

   107 public:

   108     SyncSum()

   109         : sum_(0)

   110     {

   111         ;

   112     }

   113 

   114     

   115     int sum()

   116     {

   117         { // SYNC

   118             Cpt::SyncRegion

   119                 sr(*this);

   120             

   121             return sum_;

   122 

   123         } // SYNC

   124     }

   125 

   126 

   127     void inc(int delta)

   128     {

   129         { // SYNC

   130             Cpt::SyncRegion

   131                 sr(*this);

   132             

   133             sum_ += delta;

   134 

   135         } // SYNC

   136     }

   137 };

   138 

   139 

   140 struct ConsumerParam

   141 {

   142     IntQueue                 * queue_;

   143     SyncSum                  * sum_;

   144     Itk::TestMgr             * testMgr_;

   145 };

   146 

   147 

   148 void * consumer(void * param)

   149 {

   150     struct ConsumerParam

   151         const * p = reinterpret_cast<ConsumerParam const *>(param);

   152 

   153     MSG("(B) ");

   154 

   155     bool

   156         x = false;

   157 

   158     try

   159         {

   160             bool

   161                 goOn = true;

   162 

   163             while (goOn)

   164                 {

   165                     MSG(x ? "xB " : ". ");

   166                     int

   167                         i;

   168                     goOn = p->queue_->get(&i);

   169                     if (goOn)

   170                         {

   171                             p->sum_->inc(i);

   172                             MSG("<%d ",

   173                                    i);

   174                         }

   175                     else

   176                         {

   177                             MSG("\n#IGNORE !X ");

   178                             x = true;

   179                         }

   180                     MSG(x ? "xA " : ". ");

   181                     ITK_DBGMSG(p->testMgr_, ".");

   182                 }

   183         }

   184     catch (Cpt::SyncExc & exc)

   185         {

   186             // we should not get exceptions

   187             MSG("\nSync exception in consumer ");

   188         }

   189 

   190     MSG("\n" LENIENT_STR "(E) ");

   191 

   192     return NULL;

   193 }

   194 

   195 

   196 

   197 void testMultiThreadSession(Itk::TestMgr * testMgr,

   198                             int            count,

   199                             int            mainSleepSec)

   200 {

   201     IntQueue

   202         queue(5);

   203     SyncSum

   204         sum;

   205     

   206     struct ProducerParam

   207         producerParam = {

   208         &queue,

   209         10,  // max count

   210         1100,    // max sleep millisecs

   211         testMgr

   212     };

   213 

   214     struct ConsumerParam

   215         consumerParam = {

   216         &queue,

   217         &sum,

   218         testMgr

   219     };

   220 

   221     std::list<pthread_t>

   222         threadHndls;

   223 

   224     int

   225         result;

   226 

   227     MSG("Creating %d producers and %d consumers\n",

   228         count,

   229         count);

   230 

   231     MSG(LENIENT_STR);

   232 

   233     for (int i = 0; i < count; ++i)

   234         {

   235             pthread_t

   236                 threadHndl;

   237 

   238             result = pthread_create(&threadHndl,

   239                                     NULL,

   240                                     &producer,

   241                                     &producerParam);

   242             ITK_ASSERT(testMgr,

   243                        result == 0,

   244                        "Could not create producer thread %d",

   245                        2*i);

   246             

   247             threadHndls.push_back(threadHndl);

   248 

   249             result = pthread_create(&threadHndl,

   250                                     NULL,

   251                                     &consumer,

   252                                     &consumerParam);

   253 

   254             ITK_ASSERT(testMgr,

   255                        result == 0,

   256                        "Could not create consumer thread %d",

   257                        2*i + 1);

   258 

   259             threadHndls.push_back(threadHndl);

   260         }

   261 

   262     result = sleep(mainSleepSec);

   263 

   264     MSG("forcing stop ");

   265             

   266     queue.stopProcessing();

   267             

   268     MSG("forced stop ");

   269 

   270     MSG("joining threads ");

   271 

   272     int

   273         jIdx = 0;

   274 

   275     std::list<pthread_t>::iterator

   276         i = threadHndls.begin(),

   277         end = threadHndls.end();

   278     for (; i != end; ++i)

   279         {

   280             void

   281                 * retVal;

   282 

   283             MSG("j:%d ",

   284                 jIdx);

   285 

   286             result = pthread_join(*i,

   287                                   &retVal);

   288             ITK_EXPECT(testMgr,

   289                        result == 0,

   290                        "Failed to join %s thread %d",

   291                        ((jIdx % 2) == 0 ? "producer" : "consumer"),

   292                        jIdx);

   293 

   294             ++jIdx;

   295         }

   296 

   297     MSG("\nJoined all workers\n");

   298     MSG("\n" LENIENT_STR "SUM: %d\n",

   299         sum.sum());

   300     MSG("\n session done.");

   301 }

   302 

   303 

   304 

   305 void testMTEarlyStop1(Itk::TestMgr * testMgr)

   306 {

   307     testMultiThreadSession(testMgr,

   308                            1,     // count

   309                            2);     // sleep

   310 }

   311 

   312 

   313 void testMTEarlyStop2(Itk::TestMgr * testMgr)

   314 {

   315     testMultiThreadSession(testMgr,

   316                            2,     // count

   317                            2);     // sleep

   318 }

   319 

   320 

   321 void testMTEarlyStop5(Itk::TestMgr * testMgr)

   322 {

   323     testMultiThreadSession(testMgr,

   324                            5,     // count

   325                            2);     // sleep

   326 }

   327 

   328 

   329 void testMTLateStop1(Itk::TestMgr * testMgr)

   330 {

   331     testMultiThreadSession(testMgr,

   332                            1,     // count

   333                            12);     // sleep

   334 }

   335 

   336 

   337 void testMTLateStop2(Itk::TestMgr * testMgr)

   338 {

   339     testMultiThreadSession(testMgr,

   340                            2,     // count

   341                            12);    // sleep

   342 }

   343 

   344 

   345 void testMTLateStop5(Itk::TestMgr * testMgr)

   346 {

   347     testMultiThreadSession(testMgr,

   348                            5,     // count

   349                            12);     // sleep

   350 }

   351 

   352 

   353 

   354 

   355 

   356 Itk::TesterBase * CreateSyncQueueTests()

   357 {

   358     using namespace Itk;

   359 

   360     SuiteTester

   361         * syncQueueTests = new SuiteTester("syncqueue");

   362 

   363 #define TEST "earlyStop1"

   364     syncQueueTests->add(TEST,

   365                         testMTEarlyStop1,

   366                         TEST,

   367                         LENIENT_STR);

   368 #undef TEST

   369 

   370 #define TEST "earlyStop2"

   371     syncQueueTests->add(TEST,

   372                         testMTEarlyStop2,

   373                         TEST,

   374                         LENIENT_STR);

   375 #undef TEST

   376 

   377 #define TEST "earlyStop5"

   378     syncQueueTests->add(TEST,

   379                         testMTEarlyStop5,

   380                         TEST,

   381                         LENIENT_STR);

   382 #undef TEST

   383 

   384 #define TEST "lateStop1"

   385     syncQueueTests->add(TEST,

   386                         testMTLateStop1,

   387                         TEST,

   388                         LENIENT_STR);

   389 #undef TEST

   390 

   391 #define TEST "lateStop2"

   392     syncQueueTests->add(TEST,

   393                         testMTLateStop2,

   394                         TEST,

   395                         LENIENT_STR);

   396 #undef TEST

   397 

   398 #define TEST "lateStop5"

   399     syncQueueTests->add(TEST,

   400                         testMTLateStop5,

   401                         TEST,

   402                         LENIENT_STR);

   403 #undef TEST

   404 

   405     // ... add more tests to suite

   406     

   407     return syncQueueTests;

   408     

   409 }