/*!
  @file           RTETask_WorkerJob.cpp
  @author         JoergM
  @ingroup        Runtime
  @brief          Scheduling

  System Independend Worker Thread For Time Consuming Or Blocking Task Work
  Needed i.e. during initialization of HotStandby 

\if EMIT_LICENCE

    ========== licence begin  GPL
    Copyright (c) 2001-2005 SAP AG

    This program is free software; you can redistribute it and/or
    modify it under the terms of the GNU General Public License
    as published by the Free Software Foundation; either version 2
    of the License, or (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
    ========== licence end



\endif
*/
#include "RunTime/Tasking/RTETask_IWorkerJob.hpp"
#include "RunTime/System/RTESys_AtomicOperation.hpp"
#include "RunTime/MemoryManagement/RTEMem_RteAllocator.hpp"
#include "SAPDBCommon/MemoryManagement/SAPDBMem_NewDestroy.hpp"
#include "RunTime/RTE_Crash.hpp"
#include "RunTime/RTE_MessageList.hpp"
#include "heo07.h"
#define KERNEL_LZU
#include "geo50_0.h"

class WorkerQueue
{
public:
    struct WorkerQueueEntry
    {
        struct WorkerQueueEntry *pPreviousEntry;
        RTETask_IWorkerJob      &theJob;

        WorkerQueueEntry(RTETask_IWorkerJob &job)
            : pPreviousEntry(0)
            , theJob(job)       {}
    };

    WorkerQueue()
        : m_TailOfQueue(0)
    {
        tsp00_ErrTextc errText;
        teo07_ThreadErr errCode;
        sqlcreatesem( &m_QueueSemaphore, 0, errText, &errCode);
        if ( THR_OK_EO07 != errCode )
        {
            RTE_Crash(SAPDBErr_MessageList(RTE_CONTEXT, RTEERR_ALLOCATION_FAILED, "worker thread semaphore"));
        }
    }

    void Enqueue(WorkerQueueEntry &newEntry);

    void MainLoop();

    teo07_Thread * GetThreadPointer() { return &m_Thread; }
private:
    struct WorkerQueueEntry *m_TailOfQueue;
    teo07_ThreadSemaphore    m_QueueSemaphore;
    teo07_Thread             m_Thread;
};

static WorkerQueue *pWorkerQueue = 0; // Initialized by RTETask_CreateWorkerThread()

volatile SAPDB_Int4 RTETask_WorkerNeverStop = 1; // to prevent compiler from complaining main loop never stops...

extern "C"
{
    void *WorkerProc(void *arg);
    void vcontinue(RTE_TaskId id); //--- continue task (callable from outside tasking code
    void vstop(RTE_TaskId ownId);  //--- stop task until vcontinue is called
};

//--------------------------------------------------------
void RTETask_CreateWorkerThread(SAPDB_Int4 workerThreadStackSize)
{
    pWorkerQueue = new ( RTEMem_RteAllocator::Instance() ) WorkerQueue();
    if ( !pWorkerQueue )
    {
        RTE_Crash(SAPDBErr_MessageList(RTE_CONTEXT, RTEERR_ALLOCATION_FAILED, "worker thread queue"));
    }

    tsp00_ErrTextc  errText;
    teo07_ThreadErr errCode;
    sqlbeginthread( workerThreadStackSize
                   ,WorkerProc
                   ,pWorkerQueue
                   ,THR_CREATE_SUSPENDED_EO07
                   ,pWorkerQueue->GetThreadPointer()
                   ,errText
                   ,&errCode );
    if ( THR_OK_EO07 != errCode )
    {
        RTE_Crash(SAPDBErr_MessageList( RTE_CONTEXT
                                      , RTEERR_CREATE_THREAD
                                      , "Task Worker"
                                      , SAPDB_ToString(errCode)
                                      , &errText[0]));
    }

    sqlresumethread( *(pWorkerQueue->GetThreadPointer()), errText, &errCode );
    if ( THR_OK_EO07 != errCode )
    {
        RTE_Crash(SAPDBErr_MessageList( RTE_CONTEXT
                                      , RTEERR_RESUME_THREAD
                                      , "Task Worker"
                                      , SAPDB_ToString(errCode)
                                      , &errText[0]));
    }
}

//--------------------------------------------------------
void RTETask_CallWorker(RTETask_IWorkerJob &job)
{
    WorkerQueue::WorkerQueueEntry newEntry(job);
    
    // enqueue new job and trigger worker thread if not already busy
    pWorkerQueue->Enqueue(newEntry);

    // stop until worker thread has done its job
    vstop(job.GetTaskId());
}

//--------------------------------------------------------
void WorkerQueue::MainLoop()
{
    do
    {
        while ( !m_TailOfQueue )
        {
            sqlwaitsem(m_QueueSemaphore);
        }

        // first top entry of queue and pointer to pointer of top entry
        WorkerQueueEntry *pEntry = m_TailOfQueue;
        WorkerQueueEntry **ppFirstEntry = &m_TailOfQueue;
        while ( pEntry->pPreviousEntry )
        {
            ppFirstEntry = &(pEntry->pPreviousEntry);
            pEntry = pEntry->pPreviousEntry;
        }

        // clear pointer to top entry of queue (if only one element, tail of queue pointer is cleared!
        *ppFirstEntry = 0;

        pEntry->theJob.DoJob();

        vcontinue(pEntry->theJob.GetTaskId());

    } while ( RTETask_WorkerNeverStop );
}

//-----------------------------------------------------------------
void WorkerQueue::Enqueue(WorkerQueueEntry &newEntry)
{
    //---- loop until we have successfully replaced tail of queue entry
    do
    {
        newEntry.pPreviousEntry = m_TailOfQueue;
    } while ( !RTESys_CompareAndExchange( m_TailOfQueue, 
#ifdef OSF1
                                          newEntry.pPreviousEntry,
                                          &newEntry,
#else
                                          const_cast<WorkerQueueEntry * const>(newEntry.pPreviousEntry),
                                          const_cast<WorkerQueueEntry * const>(&newEntry),
#endif
                                          newEntry.pPreviousEntry ) );

    //----- signal new entry
    sqlsignalsem(m_QueueSemaphore);
}

//----------------------------------------
// C Linkage worker thread main entry
extern "C"
{
    void *WorkerProc(void *arg)
    {
#      if defined (_WIN32)
        __try
        {
#        if defined(_FASTCAP)
          CAPNameThread ("TaskWorkerProc");
#        endif
#      endif

        ((WorkerQueue*)arg)->MainLoop();

#      if defined (_WIN32)
         }
         __except( sql50k_excepthandler(GetExceptionInformation()) )
        {
          ;
        }
#      endif

        return (void *)0;
    }
}

//--------- to be removed if ven81.c and vos81k.c are rewritten in C++
extern "C" void RTETask_CWrapperForCreateWorkerThread(SAPDB_Int4 stackSize)
{
    RTETask_CreateWorkerThread(stackSize);
}
