// @@@ START COPYRIGHT @@@
//
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.
//
// @@@ END COPYRIGHT @@@

#include <sys/types.h>
#include <stdlib.h>
#include <string.h>

#include "seabed/ms.h"
#include "seabed/thread.h"

#include "tminfo.h"
#include "seabed/trace.h"
#include "tmlogging.h"
#include "tmtimer.h"

extern void HbaseTM_initiate_cp();

// -------------------------------------------------------------
// tmTimer_initiate_cp
// -------------------------------------------------------------
void tmTimer_initiate_cp ()
{
    if (gv_tm_info.state() != TM_STATE_UP && 
        gv_tm_info.state() != TM_STATE_TX_DISABLED &&
        gv_tm_info.state() != TM_STATE_QUIESCE &&
        gv_tm_info.state() != TM_STATE_DRAIN)
    {
        TMTrace (2, ("tmTimer_inititate_cp: EXIT - TM state %d disallows cp.\n",
            gv_tm_info.state()));
        return;
    }
 
   TMTrace(2, ("tmTimer_initiate_cp : ENTRY.\n"));
   if (gv_tm_info.use_tlog()) {

       // Initiate HBase TM Control Point
       HbaseTM_initiate_cp();

        // write our records
       gv_tm_info.write_all_trans_state();

       gv_system_tx_count = 0;
   }
   TMTrace(2, ("tmTimer_initiate_cp : EXIT.\n"));
} //tmTimer_initiate_cp


// -------------------------------------------------------------
// tmTimer_stats
// Purpose : wake up at specified interval.
// -------------------------------------------------------------
void tmTimer_stats()
{
   TMTrace(2, ("tmTimer_stats : ENTRY (writing tx_stats - noop in Trafodion).\n"));
}


// -------------------------------------------------------------
// tmTimer_RMRetry
// Purpose : Retry RM opens for any failed RMs.
// -------------------------------------------------------------
void tmTimer_RMRetry()
{
    if (gv_tm_info.state() != TM_STATE_UP && 
        gv_tm_info.state() != TM_STATE_TX_DISABLED &&
        gv_tm_info.state() != TM_STATE_DRAIN)
    {
        TMTrace (2, ("TM_Info::tmTimer_RMRetry: EXIT - TM state %d is not in up or tx "
            "disabled state.\n", gv_tm_info.state()));
        return;
    }
   int32 lv_failedRMs = gv_RMs.num_rm_failed();
   TMTrace(2, ("tmTimer_RMRetry : ENTRY found %d failed RMs.\n", lv_failedRMs));
   if (lv_failedRMs > 0)
      gv_tm_info.CheckFailed_RMs();
   TMTrace(2, ("tmTimer_RMRetry : EXIT.\n"));
}


// -------------------------------------------------------------
// tmTimer_RecoveryWait
// Purpose : Wait for transaction count to be 0 before continuing
// -------------------------------------------------------------
void tmTimer_RecoveryWait(CTmTimerEvent * pp_event)
{

   int32 lv_nid = pp_event->request()->u.iv_tmrecovery_internal.iv_nid;
   
   if (lv_nid == -1) //Cluster recovery
   {
       int32 lv_activeTxns = gv_tm_info.transactionPool()->get_inUseList()->size();
       int32 lv_queuedTxns = gv_tm_info.ClusterRecov()->txnStateList()->size();

       TMTrace(2, ("tmTimer_RecoveryWait : Cluster ENTRY. Indoubt queued: %d, in progress %d.\n",
               lv_queuedTxns, lv_activeTxns));

       //Check the number of in doubt transactions
       if (lv_activeTxns + lv_queuedTxns > 0 ) 
       {
          int32 lv_availableTxnObjs = 
                MIN(gv_tm_info.transactionPool()->get_maxPoolSize(), gv_tm_info.maxRecoveringTxns())  - lv_activeTxns;
          if (lv_availableTxnObjs > 0)
          {
              gv_tm_info.ClusterRecov()->queueTxnObjects();
              gv_tm_info.ClusterRecov()->resolve_in_doubt_txs(-1/*all tms*/, false/*no delay*/);
          }
          else
              gv_tm_info.addTMRecoveryWait(lv_nid, 1000 /*1 sec*/);
       }
       else 
       {
          gv_tm_info.ClusterRecov()->completeRecovery();
          delete gv_tm_info.ClusterRecov();
          gv_tm_info.ClusterRecov(NULL);
          tm_log_event(DTM_RECOVERY_COMPLETED, SQ_LOG_NOTICE, "DTM_RECOVERY_COMPLETED", 
                       -1, -1, gv_tm_info.nid());
          TMTrace(1, ("System recover : DTM%d System startup recovery completed.\n", gv_tm_info.nid()));
       }
    }
    else 
    {
        int32 lv_queuedTxns = gv_tm_info.NodeRecov(lv_nid)->txnStateList()->size();

        TMTrace(2, ("tmTimer_RecoveryWait : Node ENTRY. Node %d has indoubt queued: %d.\n",
                lv_nid, lv_queuedTxns));

        //Check the number of in doubt transactions
        if (lv_queuedTxns > 0 )
        {
           int32 lv_availableTxnObjs =
                  gv_tm_info.transactionPool()->get_maxPoolSize() - gv_tm_info.num_active_txs();
           if (lv_availableTxnObjs)
           {
               gv_tm_info.NodeRecov(lv_nid)->queueTxnObjects();
               gv_tm_info.NodeRecov(lv_nid)->resolve_in_doubt_txs(lv_nid, 1000 /*1 sec*/);
           }
           else
               gv_tm_info.addTMRecoveryWait(lv_nid, 1000 /*1 sec*/);
        }
        else 
        {
           gv_tm_info.NodeRecov(lv_nid)->update_registry_txs_to_recover(0);
        }

     }

     TMTrace(2, ("tmTimer_RecoveryWait : EXIT.\n"));
}


// -------------------------------------------------------------
// tmTimer_initializeRMs
// Purpose : Initiailze the RMs.  This sends xa_open to all
// TSEs.
// -------------------------------------------------------------
void tmTimer_initializeRMs()
{
   TMTrace(2, ("tmTimer_initializeRMs : ENTRY.\n"));
   
   gv_tm_info.init_and_recover_rms();

   TMTrace(2, ("tmTimer_initializeRMs : EXIT.\n"));
} //tmTimer_initialize_RMs


// -------------------------------------------------------------
// tmTimer_recoverSystem
// Purpose : Runs system recovery under the timer thread.
// -------------------------------------------------------------
void tmTimer_recoverSystem()
{
   TMTrace(2, ("tmTimer_recoverSystem : ENTRY.\n"));
   
   gv_tm_info.ClusterRecov()->recover_system();

   TMTrace(2, ("tmTimer_recoverSystem : EXIT.\n"));
} //tmTimer_initialize_RMs


// -------------------------------------------------------------
// tmTimer_TMRestartRetry
// Purpose : Retry TM Restart for TMs which don't start on the
// first try.  This is only present in Lead TMs, and only when
// a TM is down and we need to retry the open.
// -------------------------------------------------------------
void tmTimer_TMRestartRetry(CTmTimerEvent * pp_event)
{
    int32 lv_error = FEOK;
    int32 lv_nid = pp_event->request()->u.iv_tmrestart_internal.iv_nid;
    TMTrace(2, ("tmTimer_TMRestartRetry : ENTRY Attempting to restart "
            "$TM%d.\n", lv_nid));

    if (gv_tm_info.state() != TM_STATE_UP && 
        gv_tm_info.state() != TM_STATE_TX_DISABLED &&
        gv_tm_info.state() != TM_STATE_DRAIN)
    {
        TMTrace(1, ("TM_Info::tmTimer_TMRestartRetry: EXIT - Too late! "
                "Lead TM not in up or tx disabled state.\n"));
        return;
    }
    lv_error = gv_tm_info.recover_tm(lv_nid);

    TMTrace(2, ("tmTimer_TMRestartRetry : EXIT, error %d for nid %d.\n", 
            lv_error, lv_nid));
}


//----------------------------------------------------------------------------
// timerThread_main
// Purpose : Main for timer thread
//----------------------------------------------------------------------------
void * timerThread_main(void *arg)
{
   CTmTimerEvent   *lp_event;
   CTmTimer        *lp_timerTh;
   bool             lv_exit = false;
   bool                lv_deleteEvent = false;

   arg = arg;
   CTmTxMessage *lp_msg;

   TMTrace(2, ("timerThread_main : ENTRY.\n"));

   while (gv_tm_info.tmTimer() == NULL || gv_tm_info.tmTimer()->state() != TmTimer_Up || gv_tm_info.tmAuditObj() == NULL)
   {
      SB_Thread::Sthr::usleep(10);
   }
   // Now we can set a pointer to the CTmTimer object because it exits
   lp_timerTh = gv_tm_info.tmTimer();

   TMTrace(2, ("timerThread_main : Thread %s(%p) State Up.\n",
      lp_timerTh->get_name(), (void *) lp_timerTh));

   while (!lv_exit)
   {
      lp_event = (CTmTimerEvent *) lp_timerTh->eventQ_pop();

      if (lp_event)
      {
         switch (lp_event->command())
         {
         case TmTimerCmd_Queue:
         {
            // Queue request to timer list
            TMTrace(3, ("timerThread_main : TmTimerCmd_Queue: Queue timer event. "
               "Txn ID %d, Wakeup interval %d, repeat %d, Request %d\n",
               ((lp_event->transaction())?lp_event->transaction()->seqnum():0),
               lp_event->wakeupInterval(), lp_event->wakeupRepeat(),
               lp_event->requestType()));
            lp_event->command(TmTimerCmd_Timer); //Change to timer event
            lp_timerTh->timerList()->add(lp_event);
            break;
         } //case TmTimerCmd_Queue Incoming request

         case TmTimerCmd_Timer:
         {
            // A timer popped, queue request to the transactions event queue
            TMTrace(3, ("timerThread_main : TmTimerCmd_Timer: Timer pop. "
               "Txn ID %d, Wakeup interval %d, repeat %d, Request %d, msgid %d\n",
               ((lp_event->transaction())?lp_event->transaction()->seqnum():0),
               lp_event->wakeupInterval(), lp_event->wakeupRepeat(),
               lp_event->requestType(), lp_event->msg()->msgid()));
            lp_timerTh->lock();

            switch (lp_event->requestType())
            {
            case TM_MSG_TXINTERNAL_CONTROLPOINT:
               // Control Point (must be Lead TM).
               lp_timerTh->last_cp_written(SB_Thread::Sthr::time());
               tmTimer_initiate_cp();
               break;
            case TM_MSG_TXINTERNAL_STATS:
               tmTimer_stats();
               break;
            case TM_MSG_TXINTERNAL_RMRETRY:
               break;
            case TM_MSG_TXINTERNAL_TMRESTART_RETRY:
               // If this is the first and only event or the last of several - then service it.  If 
               // it is not the last one (of duplicate events), then ignore it. 
               if (gv_tm_info.get_restartTimerEvent(lp_event->request()->u.iv_tmrestart_internal.iv_nid)
                        == lp_event)
               {
                   gv_tm_info.reset_restartTimerEvent (lp_event->request()->u.iv_tmrestart_internal.iv_nid);
                   tmTimer_TMRestartRetry(lp_event);
               }
               lv_deleteEvent = true;
               break;
            case TM_MSG_TXINTERNAL_SHUTDOWNP1_WAIT:
               lp_msg = (CTmTxMessage *) lp_event;
               gv_tm_info.ShutdownPhase1Wait(lp_msg);
               break;
            case TM_MSG_TYPE_ATTACHRM:
               lp_msg = new CTmTxMessage(lp_event->request());
               gv_tm_info.attachRm(lp_msg);
               lv_deleteEvent = true;
               delete lp_msg;
               break;
            case TM_MSG_TYPE_ENABLETRANS:
               lp_msg = new CTmTxMessage(lp_event->request());
               gv_tm_info.enableTrans(lp_msg);
               lv_deleteEvent = true;
               delete lp_msg;
               break;
            case TM_MSG_TYPE_DISABLETRANS:
               lp_msg = new CTmTxMessage(lp_event->request());
               gv_tm_info.disableTrans(lp_msg);
               lv_deleteEvent = true;
               delete lp_msg;
               break;
            case TM_MSG_TXINTERNAL_RECOVERY_WAIT:
               tmTimer_RecoveryWait(lp_event);
               lv_deleteEvent = true;
               break;
            case TM_MSG_TXINTERNAL_INITIALIZE_RMS:
               tmTimer_initializeRMs();
               lv_deleteEvent = true;
               break;
            case TM_MSG_TXINTERNAL_SYSTEM_RECOVERY:
               tmTimer_recoverSystem();
               lv_deleteEvent = true;
               break;
            default:
            {
                if (lp_event->transaction())
                {
                   // Notify transaction object
                   CTmTxMessage *lp_msg = new CTmTxMessage(lp_event->request());
                   lp_event->transaction()->queueToTransaction(lp_event->transaction()->transid(), 
                                                               lp_msg);
                   lv_deleteEvent = true;
                }
                else if (lp_event->thread())
                   // Notify non-transactional thread
                   lp_event->thread()->eventQ_push((CTmEvent *) lp_event);
                else
                {
                   TMTrace(1, ("timerThread_main : Timer Thread %p. No transaction or "
                            "thread pointer for event %p, Txn ptr %p, cmd %d, reqType %d, Wakeup interval %d, repeat %d.\n",
                            (void *) lp_timerTh, (void *) lp_event, (void *) lp_event->transaction(), lp_event->command(), 
                            lp_event->requestType(), lp_event->wakeupInterval(), lp_event->wakeupRepeat()));
                   tm_log_event (DTM_TMTIMER_BAD_EVENT2, SQ_LOG_WARNING, "DTM_TMTIMER_BAD_EVENT2",
                       lp_event->request()->iv_msg_hdr.miv_err.error, -1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,
                       lp_event->requestType(), lp_event->request()->iv_msg_hdr.dialect_type,
                       -1, NULL, -1, lp_event->wakeupInterval(), lp_event->wakeupRepeat());
                   // abort(); Don't want to abort, just discard event.
                   lv_deleteEvent = true;
                }
            } //default
            } //switch 

            if (lp_event->wakeupRepeat() != 0 && lv_deleteEvent == false)
            {
               lp_event->dec_wakeupRepeat();
               lp_timerTh->timerList()->add(lp_event);
            }
            lp_timerTh->unlock();
            break;
         } //case TmTimerCmd_Timer timer pop

         case TmTimerCmd_Stop:
         {
            lp_timerTh->lock();
            lv_exit = true;
            TMTrace(1, ("timerThread_main : Stop thread received.\n"));
            lv_deleteEvent = true;
            lp_timerTh->state(TmTimer_Down);
            // Unlock on exit
            break;
         } //case TmTimerCmd_Stop the timer thread

         case TmTimerCmd_Cancelled:
         {
            TMTrace(1, ("timerThread_main : Cancelled timer event %p popped.\n",
               (void *) lp_event));
            lv_deleteEvent = true;
            break;
         } //case TmTimerCmd_Cancelled

         default:
         {
            // EMS DTM_TXTHREAD_BAD_EVENT
            TMTrace(1, ("timerThread_main: Timer Thread (0x%p) main received an "
                "unexpected event %p, reqType %d, cmd %d, terminating.\n", 
                (void *) lp_timerTh, (void *) lp_event, lp_event->requestType(), lp_event->command()));
            tm_log_event(DTM_TMTIMER_UNEXP_EVENT, SQ_LOG_CRIT, "DTM_TMTIMER_UNEXP_EVENT",
                         -1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,-1,lp_event->requestType(),lp_event->command());
            abort();
         }
         } //switch
         if (lv_deleteEvent)
         {
            TMTrace(2, ("timerThread_main : thread_deleted p_event (0x%p), cmd %d, type %d.\n", 
                    (void *) lp_event, lp_event->command(), lp_event->requestType()));
            delete lp_event;
            lv_deleteEvent = false;
         }
      } // if lp_event
   } //while

   TMTrace(2, ("timerThread_main : EXIT.\n"));

   lp_timerTh->unlock();
   return NULL;
} //timerThread_main