/*

* 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 <vector>

#include "itk.h"

#include "cpixjobqueue.h"

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

class Job

{

    int        id_;

    int        rounds_;

    bool       cancelled_;

public:

    Job(int id,

        int rounds)

        : id_(id),

          rounds_(rounds),

          cancelled_(false)

    {

        ;

    }

    int id() const

    {

        return id_;

    }

    void cancel(int id) // TODO OBS id

    {

        if (id != id_)

            {

                ITK_PANIC("Messed up cancel id");

            }

        cancelled_ = true;

        MSG("JOB %d - cancelled\n",

            id_);

    }

    void doJob()

    {

        MSG("Starting JOB %d - %d rounds\n",

            id_,

            rounds_);

        for (; rounds_ > 0 && !cancelled_; --rounds_)

            {

                sleep(1);

                MSG("JOB %d : %d rounds to go\n",

                    id_,

                    rounds_);

            }

        MSG("JOB %d - %s\n",

            id_,

            cancelled_ ? "CANCELLED" : "COMPLETED");

    }

};

typedef Cpt::JobQueue<Job*> TestJobQueue;

class JobVector : public std::vector<Job*>, public Cpt::SyncEntity

{

public:

    ~JobVector()

    {

        std::vector<Job*>::iterator

            i = begin(),

            e = end();

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

            {

                delete *i;

            }

    }

};

struct ThreadParam

{

    TestJobQueue    * q_;

    JobVector       * v_;

    Itk::TestMgr    * testMgr_;

};

static void * producer(void * param)

{

    using namespace std;

    struct ThreadParam

        * p = reinterpret_cast<ThreadParam *>(param);

    MSG("Producer thread started\n");

    try

        {

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

                {

                    std::auto_ptr<Job>

                        job(new Job(i,

                                    i));

                    { // SYNC

                        Cpt::SyncRegion

                            sr(*p->v_);

                        p->v_->push_back(job.get());

                        p->q_->put(job.get());

                    } // SYNC

                    job.release();

                    MSG("Producer thread created job #%d\n",

                        i);

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

                    sleep(1);

                }

            sleep(3);

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

                {

                    { // SYNC

                        Cpt::SyncRegion

                            sr(*p->v_);

                        int

                            id = 2*i;

                        Job

                            * job = NULL;

                        bool

                            found = p->q_->findJob(compose1(bind2nd(equal_to<int>(),

                                                                    id),

                                                            mem_fun(&Job::id)),

                                                   &job);

                        if (found)

                            {

                                job->cancel(id);

                                MSG("Producer thread cancelled job #%d\n",

                                    id);

                            }

                        else

                            {

                                MSG("Started job was not found");

                            }

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

                    } // SYNC

                }

            sleep(6);

            p->q_->stopProcessing();

        }

    catch (...)

        {

            MSG("Producer thread: cancelled\n");

        }

    MSG("Producer thread stopped\n");

    return NULL;

}

static void * consumer(void * param)

{

    struct ThreadParam

        * p = reinterpret_cast<ThreadParam *>(param);

    MSG("Consumer thread started\n");

    try

        {

            bool

                result = true;

            while (result)

                {

                    Job

                        * job = NULL;

                    MSG("Consumer thread getting ...\n");

                    result = p->q_->get(&job);

                    MSG("Consumer thread has got (%d) ...\n",

                        result);

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

                    if (result)

                        {

                            job->doJob();

                            p->q_->jobCompleted(job);

                        }

                }

        }

    catch (...)

        {

            MSG("Consumer thread: cancelled\n");

        }

    MSG("Consumer thread stopped\n");

    return NULL;

}

void testCancel(Itk::TestMgr * testMgr)

{

    TestJobQueue

        jobQueue(10);

    JobVector

        jobVector;

    struct ThreadParam threadParam = 

        {

            &jobQueue,

            &jobVector,

            testMgr

        };

    std::list<pthread_t>

        threadHndls;

    int

        result;

    MSG("Creating 1 producer and 1 consumer\n");

    pthread_t

        threadHndl;

    result = pthread_create(&threadHndl,

                            NULL,

                            &producer,

                            &threadParam);

    ITK_ASSERT(testMgr,

               result == 0,

               "Could not create producer thread");

    threadHndls.push_back(threadHndl);

    result = pthread_create(&threadHndl,

                            NULL,

                            &consumer,

                            &threadParam);

    ITK_ASSERT(testMgr,

               result == 0,

               "Could not create consumer thread");

    threadHndls.push_back(threadHndl);

    MSG("joining threads\n");

    int

        jIdx = 0;

    std::list<pthread_t>::iterator

        i = threadHndls.begin(),

        end = threadHndls.end();

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

        {

            void

                * retVal;

            MSG("j:%d\n",

                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("Joined all workers\n");

    MSG("session done.\n");

}

Itk::TesterBase * CreateJobQueueTests()

{

    using namespace Itk;

    SuiteTester

        * jobQueueTests = new SuiteTester("jobqueue");

#define TEST "cancel"

    jobQueueTests->add(TEST,

                       testCancel,

                       TEST,

                       SuiteTester::REDIRECT_ONLY);

#undef TEST

    // ... add more tests to suite

    return jobQueueTests;

}