/*

* 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 <stdio.h>

#include <stdlib.h>

#include <wchar.h>

#include <glib.h>

#include <exception>

#include <functional>

#include "cpixsynctools.h"

#include "cpixjobqueue.h"

// Own header

#include "cpix_async.h"

// cLucene API

#include "CLucene.h"

#include "CLucene/queryParser/MultiFieldQueryParser.h"

// CPix internal

#include "iidxdb.h"

#include "idxdb.h"

#include "cpixsearch.h"

#include "cpixhits.h"

#include "initparams.h"

#include "cpixutil.h"

#include "iqrytype.h"

#include "analyzer.h"

#include "document.h"

#include "cpixthreadlocaldata.h"

#include "common/cpixlog.h"

// TODO CLucene deep cancel feature will require this header

// #include "CLucene/store/SearchCancel.h"

namespace

{

    // job in progress

    //cpix_JobId currentJobInProgress;

    enum AsyncOpKind

        {

            IDX,

            QRY

        };

    /**

     * Represents an asynchronous CPix job, polymorphic base class.

     */

    class AsyncJob

    {

    public:

        //

        // public typedefs

        //

        typedef Cpt::JobQueue<AsyncJob*> Queue;

    private:

        //

        // private members

        //

        void                 * cookie_;

        cpix_AsyncCallback   * callback_;

        cpix_Error           * err_;

        /**

         * Instantiated during async engine initialization, so there

         * is point in protecting the pointer itself, and the queue

         * instance itself is thread-safe.

         */

        static Queue         * idxQueue_;

        static Queue         * qryQueue_;

        /**

         * The job id for the next AsyncJob instance created.

         *

         * NOTE: currently only one thread accesses the AsyncJob

         * constructor (the master thread), so there is no need to

         * protect this variable with mutex.

         */

        static cpix_JobId      nextJobId_;

    protected:

        //

        // protected members

        //

        IdxDbHndl              handle_;

        volatile bool          cancelled_;

        volatile bool          calledBack_;

        // accesses to: cancelled_, calledBack_ and callback_ must be

        // synchronized via this mutex

        Cpt::Mutex             mutex_;

        cpix_JobId                  id_;

    public:

        //

        // public operations

        //

        static Queue * CreateIdxQueue(size_t queueSize)

        {

            if (idxQueue_ != NULL)

                {

                    THROW_CPIXEXC(PL_ERROR "IDX job queue already exists");

                }

            idxQueue_ = new Queue(queueSize);

            return idxQueue_;

        }

        static Queue * GetIdxQueue()

        {

            if (idxQueue_ == NULL)

                {

                    THROW_CPIXEXC(PL_ERROR "IDX job queue uninitialized");

                }

            return idxQueue_;

        }

        static Queue * CreateQryQueue(size_t queueSize)

        {

            if (qryQueue_ != NULL)

                {

                    THROW_CPIXEXC(PL_ERROR "QRY job queue already exists");

                }

            qryQueue_ = new Queue(queueSize);

            return qryQueue_;

        }

        static Queue * GetQryQueue()

        {

            if (qryQueue_ == NULL)

                {

                    THROW_CPIXEXC(PL_ERROR "QRY job queue uninitialized");

                }

            return qryQueue_;

        }

        static Queue * GetQueue(AsyncOpKind asyncOpKind)

        {

            Queue

                * rv = NULL;

            switch (asyncOpKind)

                {

                case IDX:

                    rv = GetIdxQueue();

                    break;

                case QRY:

                    rv = GetQryQueue();

                    break;

                default:

                    THROW_CPIXEXC(PL_ERROR "Unknown AsyncOpKind %d",

                                  asyncOpKind);

                }

            return rv;

        }

        static void Shutdown()

        {

            delete idxQueue_;

            idxQueue_ = NULL; 

            delete qryQueue_;

            qryQueue_ = NULL; 

        }

        /**

         * Cancels this job if necessary - decision is based on

         * cancelArg. This method is called from the master (producer)

         * thread.

         *

         * @return true if the caller should destroy this job

         * instance. If the callback has already happened, then the

         * slave thread will not deal with this instance anymore, so

         * the destruction of this instance must happen now, in the

         * canceller client context.

         */

        bool cancel(cpix_JobId jobId)

        {

            bool

                rv = false;

            if (id_ != jobId)

                {

                    THROW_CPIXEXC("Cancel messup");

                }

            { // SYNC

                Cpt::SyncRegion

                    sr(mutex_);

                if (calledBack_)

                    {

                        rv = true;

                    }

                else

                    {

                        cancelled_ = true;

                    }

            } // SYNC

            return rv;

        }

        static cpix_CancelAction Cancel(cpix_JobId   jobId,

                                        AsyncOpKind  asyncOpKind)

        {

            if (asyncOpKind != QRY)

                {

                    THROW_CPIXEXC("Only qry ops can be cancelled (%d)",

                                  jobId);

                }

            cpix_CancelAction

                rv = cpix_CANCEL_ERROR;

            std::auto_ptr<AsyncJob>

                job(GetAsyncJobFromQueue(jobId,

                                         asyncOpKind));

            if (AsyncJob::GetQueue(asyncOpKind)->cancelNonStartedJob( 

                  compose1(bind2nd(equal_to<cpix_JobId>(),

                                       jobId),

                           mem_fun(&AsyncJob::id))))

                {

                    logMsg(CPIX_LL_TRACE,

                           "Cancelled job %d removed before started.",

                           jobId);

                    return cpix_CANCEL_SHALLOW;

                }

#if 0 // TODO CLucene deep cancel

            else if (jobId == currentJobInProgress)

                {

                    setSearchCancel(true);

                    rv = cpix_CANCEL_DEEP;

                }

#endif // 0

            bool

                destroy = job->cancel(jobId);

            if (destroy)

                {

                    // job is taken out of jobqueue

                    AsyncJob::GetQueue(asyncOpKind)->jobCompleted(job.get());

                    rv = cpix_CANCEL_JOBDONE;

                }

            else

                {

                    // job is not to be destroyed yet

                    job.release();

                }

            // job is taken care of by auto_ptr if needed

            return rv;

        }

        /**

         * Performs the job represented by this instance. Called from

         * slave (consumer) thread.

         *

         * @returns true if this job should be completed and destroyed

         * by the slave thread right after this function returns.

         */

        bool doJob()

        {

            try

                {

                    err_ = doJob_();

                }

            catch (...)

                {

                    if (err_ == NULL)

                        {

                            err_ = CreateError(ET_CPIX_EXC,

                                               L"doJob_ has thrown unknown exception - see logs for exact derived class / job type");

                        }

                    logMsg(CPIX_LL_ERROR,

                           "%s::doJob_ failed with exception",

                           typeid(*this).name());

                }

            bool

                rv = false;

            // On completion of job, we must call the callback unless

            // the job was cancelled on the master thread.

            { // SYNC

                Cpt::SyncRegion

                    sr(mutex_);

                if (!cancelled_)

                    {

                        try

                            {

                                (*callback_)(cookie_,

                                             id_);

                                calledBack_ = true;

                            }

                        catch (...)

                            {

                                logMsg(CPIX_LL_ERROR,

                                       "callback(%p,%d) has thrown exception",

                                       cookie_,

                                       id_);

                            }

                    }

                else

                    {

                        rv = true;

                    }

            } // SYNC

            return rv;

        }

        /**

         * @returns the error result of the completed (non-cancelled)

         * job. May be NULL, if the operation was successful. At the

         * same time, it releases ownership of the error object, if

         * any.

         */

        cpix_Error * releaseErr()

        {

            cpix_Error

                * rv = err_;

            err_ = NULL;

            return rv;

        }

        /**

         * @returns the id of this job

         */

        cpix_JobId id() const

        {

            return id_;

        }

        //

        // lifetime management

        //

        /**

         * Constructor.

         */

        AsyncJob(IdxDbHndl            handle,

                 void               * cookie,

                 cpix_AsyncCallback * callback)

            : cookie_(cookie),

              callback_(callback),

              err_(NULL),

              handle_(handle),

              cancelled_(false),

              calledBack_(false),

              id_(nextJobId_++)

        {

            ;

        }

        /**

         * Destructor (virtual).

         */

        virtual ~AsyncJob()

        {

            // if it is not NULL, it is still owned by this job,

            // therefore released at this point

            cpix_ClearError(this);

        }

        /**

         * Acquires the lock of this job and immediately releases

         * it. Sounds silly, but it is necessary - this method is only

         * called from GetStartedAsyncJobFromQueue() method, see

         * comments there.

         */

        void lockAndUnlock()

        {

            Cpt::SyncRegion

                sr(mutex_);

        }

    protected:

        //

        // protected members

        //

        /**

         * Performs the job represented by this instance. Called from

         * the slave (consumer) thread. Implementation may take

         * advantage of the fact that cancel_ can be set to true at

         * any point (called from the master thread).

         */

        virtual cpix_Error * doJob_() = 0;

        /**

         * Tries to put the created job to the queue, reporting

         * errors. Templated: cpix_IdxDb or cpix_IdxSearcher.

         */

        template<typename WRAPPER>

        static void PutAsyncJobToQueue(std::auto_ptr<AsyncJob> & job,

                                       WRAPPER                 * wrapper,

                                       AsyncOpKind               asyncOpKind)

        {

            bool

                result = GetQueue(asyncOpKind)->put(job.get());

            if (result)

                {

                    // on success, ownership is transferred to queue

                    job.release();

                }

            else

                {

                    // failure: for clarity, explicitly destroyed

                    job.reset();

                    wrapper->err_ = CreateError(ET_CPIX_EXC,

                                                "Could not put async job to queue");

                }

        }

        static AsyncJob * GetStartedAsyncJobFromQueue(cpix_JobId  jobId,

                                                      AsyncOpKind asyncOpKind)

        {

            using namespace std;

            AsyncJob

                * rv = NULL;

            bool

                result = GetQueue(asyncOpKind)

                ->findStartedJob(compose1(bind2nd(equal_to<cpix_JobId>(),

                                                  jobId),

                                          mem_fun(&AsyncJob::id)),

                                 &rv);

            if (!result)

                {

                    THROW_CPIXEXC(PL_ERROR "AsyncJob not found");

                }

            // If the client thread (master thread) is "fast" enough,

            // it could start collecting the results of a job even

            // before the slave thread has released its lock. In that

            // case we would get an EBUSY which we want to treat as an

            // error (to catch possibly logic errors wrt

            // lifetime/threading) therefore all we need to do here is

            // to sync the master thread with the slave thread:

            // collecting results (for which this method is needed)

            // will have to wait until the slave thead has released

            // the lock on this job.

            rv->lockAndUnlock();

            return rv;

        }

        static AsyncJob * GetAsyncJobFromQueue(cpix_JobId  jobId,

                                               AsyncOpKind asyncOpKind)

        {

            using namespace std;

            AsyncJob

                * rv = NULL;

            bool

                result = GetQueue(asyncOpKind)

                ->findJob(compose1(bind2nd(equal_to<cpix_JobId>(),

                                           jobId),

                                   mem_fun(&AsyncJob::id)),

                          &rv);

            if (!result)

                {

                    THROW_CPIXEXC(PL_ERROR "AsyncJob not found");

                }

            return rv;