/*

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

* All rights reserved.

* This component and the accompanying materials are made available

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

* which accompanies this distribution, and is available

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

*

* Initial Contributors:

* Nokia Corporation - initial contribution.

*

* Contributors:

*

* Description: 

*

*/

#include <unistd.h>

#include <stdio.h>

#include <stdlib.h>

#include <string.h>

#include <list>

#include "itk.h"

#include "cpixsynctools.h"

#include "cpixsyncqueue.h"

#define LENIENT_STR "#IGNORE "

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

typedef Cpt::SyncQueue<int>    IntQueue;

struct ProducerParam

{

    IntQueue                 * queue_;

    int                        maxCount_;

    int                        maxSleepMs_;

    Itk::TestMgr             * testMgr_;

};

void * producer(void * param)

{

    struct ProducerParam

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

    int

        ms = rand();

    ms = ms % p->maxSleepMs_;

    enum 

    { 

        MSEC_PER_SEC = 1000,

        NSEC_PER_MSEC = 1000000

    };

    struct timespec 

        tim;

    tim.tv_sec = ms / MSEC_PER_SEC;

    tim.tv_nsec = ms % MSEC_PER_SEC * NSEC_PER_MSEC;

    MSG("[B:%d] ",

        ms);

    try

        {

            bool

                goOn = true;

            for (int i = 0; 

                 i <= p->maxCount_ && goOn; 

                 ++i)

                {

                    nanosleep(&tim,

                              NULL);

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

                    MSG(">%d ",

                           i);

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

                }

        }

    catch (Cpt::SyncExc & exc)

        {

            // we should not get exceptions

            MSG("\nSync exception in producer ");

        }

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

    return NULL;

}

/**

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

 */

class SyncSum : public Cpt::SyncEntity

{

    int        sum_;

public:

    SyncSum()

        : sum_(0)

    {

        ;

    }

    int sum()

    {

        { // SYNC

            Cpt::SyncRegion

                sr(*this);

            return sum_;

        } // SYNC

    }

    void inc(int delta)

    {

        { // SYNC

            Cpt::SyncRegion

                sr(*this);

            sum_ += delta;

        } // SYNC

    }

};

struct ConsumerParam

{

    IntQueue                 * queue_;

    SyncSum                  * sum_;

    Itk::TestMgr             * testMgr_;

};

void * consumer(void * param)

{

    struct ConsumerParam

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

    MSG("(B) ");

    bool

        x = false;

    try

        {

            bool

                goOn = true;

            while (goOn)

                {

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

                    int

                        i;

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

                    if (goOn)

                        {

                            p->sum_->inc(i);

                            MSG("<%d ",

                                   i);

                        }

                    else

                        {

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

                            x = true;

                        }

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

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

                }

        }

    catch (Cpt::SyncExc & exc)

        {

            // we should not get exceptions

            MSG("\nSync exception in consumer ");

        }

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

    return NULL;

}

void testMultiThreadSession(Itk::TestMgr * testMgr,

                            int            count,

                            int            mainSleepSec)

{

    IntQueue

        queue(5);

    SyncSum

        sum;

    struct ProducerParam

        producerParam = {

        &queue,

        10,  // max count

        1100,    // max sleep millisecs

        testMgr

    };

    struct ConsumerParam

        consumerParam = {

        &queue,

        &sum,

        testMgr

    };

    std::list<pthread_t>

        threadHndls;

    int

        result;

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

        count,

        count);

    MSG(LENIENT_STR);

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

        {

            pthread_t

                threadHndl;

            result = pthread_create(&threadHndl,

                                    NULL,

                                    &producer,

                                    &producerParam);

            ITK_ASSERT(testMgr,

                       result == 0,

                       "Could not create producer thread %d",

                       2*i);

            threadHndls.push_back(threadHndl);

            result = pthread_create(&threadHndl,

                                    NULL,

                                    &consumer,

                                    &consumerParam);

            ITK_ASSERT(testMgr,

                       result == 0,

                       "Could not create consumer thread %d",

                       2*i + 1);

            threadHndls.push_back(threadHndl);

        }

    result = sleep(mainSleepSec);

    MSG("forcing stop ");

    queue.stopProcessing();

    MSG("forced stop ");

    MSG("joining threads ");

    int

        jIdx = 0;

    std::list<pthread_t>::iterator

        i = threadHndls.begin(),

        end = threadHndls.end();

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

        {

            void

                * retVal;

            MSG("j:%d ",

                jIdx);

            result = pthread_join(*i,

                                  &retVal);

            ITK_EXPECT(testMgr,

                       result == 0,

                       "Failed to join %s thread %d",

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

                       jIdx);

            ++jIdx;

        }

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

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

        sum.sum());

    MSG("\n session done.");

}

void testMTEarlyStop1(Itk::TestMgr * testMgr)

{

    testMultiThreadSession(testMgr,

                           1,     // count

                           2);     // sleep

}

void testMTEarlyStop2(Itk::TestMgr * testMgr)

{

    testMultiThreadSession(testMgr,

                           2,     // count

                           2);     // sleep

}

void testMTEarlyStop5(Itk::TestMgr * testMgr)

{

    testMultiThreadSession(testMgr,

                           5,     // count

                           2);     // sleep

}

void testMTLateStop1(Itk::TestMgr * testMgr)

{

    testMultiThreadSession(testMgr,

                           1,     // count

                           12);     // sleep

}

void testMTLateStop2(Itk::TestMgr * testMgr)

{

    testMultiThreadSession(testMgr,

                           2,     // count

                           12);    // sleep

}

void testMTLateStop5(Itk::TestMgr * testMgr)

{

    testMultiThreadSession(testMgr,

                           5,     // count

                           12);     // sleep

}

Itk::TesterBase * CreateSyncQueueTests()

{

    using namespace Itk;

    SuiteTester

        * syncQueueTests = new SuiteTester("syncqueue");

#define TEST "earlyStop1"

    syncQueueTests->add(TEST,

                        testMTEarlyStop1,

                        TEST,

                        LENIENT_STR);

#undef TEST

#define TEST "earlyStop2"

    syncQueueTests->add(TEST,

                        testMTEarlyStop2,

                        TEST,

                        LENIENT_STR);

#undef TEST

#define TEST "earlyStop5"

    syncQueueTests->add(TEST,

                        testMTEarlyStop5,

                        TEST,

                        LENIENT_STR);

#undef TEST

#define TEST "lateStop1"

    syncQueueTests->add(TEST,

                        testMTLateStop1,

                        TEST,

                        LENIENT_STR);

#undef TEST

#define TEST "lateStop2"

    syncQueueTests->add(TEST,

                        testMTLateStop2,

                        TEST,

                        LENIENT_STR);

#undef TEST

#define TEST "lateStop5"

    syncQueueTests->add(TEST,

                        testMTLateStop5,

                        TEST,

                        LENIENT_STR);

#undef TEST

    // ... add more tests to suite

    return syncQueueTests;

}