///////////////////////////////////////////////////////////////////////////////
//
// @@@ 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 <iostream>

using namespace std;

#include <signal.h>
#include <fcntl.h>
#include <sys/file.h>
#include <errno.h>

#include "monlogging.h"
#include "montrace.h"
#include "monitor.h"
#include "lock.h"
#include "clusterconf.h"
#include "lnode.h"
#include "pnode.h"
#include "tmsync.h"
#include "mlio.h"
#include "reqqueue.h"
#include "nameserver.h"

extern bool NameServerEnabled;
extern int trace_level;
extern int MyPNID;
extern sigset_t SigSet;
extern CMonitor *Monitor;
extern CNodeContainer *Nodes;
extern CNode *MyNode;
extern CReqQueue ReqQueue;

CTmSyncReq::CTmSyncReq( int nid, int handle, char *data, int length, int tag, bool unsolicited )
            :Nid( nid ),
             Tag( tag ),
             Handle( handle ),
             Length( length ),
             Unsolicited( unsolicited ),
             Replicated( false ),
             Completed( false ),
             Next( NULL ),
             Prev( NULL )
{
    const char method_name[] = "CTmSyncReq::CTmSyncReq";
    TRACE_ENTRY;

    // Add eyecatcher sequence as a debugging aid
    memcpy(&eyecatcher_, "TSYN", 4);

    Data = new char [Length+1];
    memmove( Data, data, Length );
    if (trace_settings & TRACE_TMSYNC)
       trace_printf("%s@%d" " - Create " "%s"  "request handle=" "%d"  "\n", method_name, __LINE__, (Unsolicited?"unsolicited ":""), Handle);
    TRACE_EXIT;
}

CTmSyncReq::~CTmSyncReq( void )
{
    const char method_name[] = "CTmSyncReq::~CTmSyncReq";
    TRACE_ENTRY;
    delete [] Data;
    if (trace_settings & TRACE_TMSYNC)
       trace_printf("%s@%d" " - Delete " "%s"  "request (%p) handle=" "%d"  "\n", method_name, __LINE__, (Unsolicited?"unsolicited ":""), this, Handle);

    // Alter eyecatcher sequence as a debugging aid to identify deleted object
    memcpy(&eyecatcher_, "tsyn", 4);

    TRACE_EXIT;
}

void CTmSyncReq::DeLink (CTmSyncReq ** head, CTmSyncReq ** tail)
{
    const char method_name[] = "CTmSyncReq::DeLink";
    TRACE_ENTRY;
    if (*head == this)
        *head = Next;
    if (*tail == this)
        *tail = Prev;
    if (Prev)
        Prev->Next = Next;
    if (Next)
        Next->Prev = Prev;
    TRACE_EXIT;
}

CTmSyncReq *CTmSyncReq::GetNext (void)
{
    const char method_name[] = "CTmSyncReq::GetNext";
    TRACE_ENTRY;
    TRACE_EXIT;
    return Next;
}

CTmSyncReq *CTmSyncReq::Link (CTmSyncReq * entry)
{
    const char method_name[] = "CTmSyncReq::Link";
    TRACE_ENTRY;
    Next = entry;
    entry->Prev = this;

    TRACE_EXIT;
    return entry;
}

CTmSync_Container::CTmSync_Container(void)
                  :CCluster(),
                   TmSyncReplies( 0 ),
                   ReqsInBlock( 0 ),
                   TmSyncReplyCode( MPI_SUCCESS ),
                   PendingSlaveTmSyncCount( 0 ),
                   Head( NULL ),
                   Tail( NULL ),
                   HandleSeq(MyPNID*MAX_TM_HANDLES),
                   PendingSlaveTmSync( false ),
                   TotalSlaveTmSyncCount( false ),
                   AbortPendingTmSync( false )
{
    const char method_name[] = "CTmSync_Container::CTmSync_Container";
    TRACE_ENTRY;

    int rc = sem_init(&UnsolicitedWaitSem, 0, 0);
    if (rc)
    {
        int err = errno;
        char la_buf[MON_STRING_BUF_SIZE];
        sprintf(la_buf, "[%s], Can't create unnamed semaphore! - errno=%d (%s)\n", method_name, err, strerror(errno));
        mon_log_write(MON_TMSYNC_INIT_1, SQ_LOG_ERR, la_buf);

        abort();
    }

    TRACE_EXIT;
}

CTmSync_Container::~CTmSync_Container(void)
{
    CTmSyncReq *req = Head;

    const char method_name[] = "CTmSync_Container::~CTmSync_Container";
    TRACE_ENTRY;

    while (req)
    {
        req->DeLink (&Head, &Tail);
        delete req;
        req = Head;
    }

    int rc = sem_destroy( &UnsolicitedWaitSem );
    if (rc)
    {
        int err = errno;
        char la_buf[MON_STRING_BUF_SIZE];
        sprintf(la_buf, "[%s], Can't destroy unnamed semaphore! - errno=%d (%s)\n", method_name, err, strerror(errno));
        mon_log_write(MON_TMSYNC_DEST_1, SQ_LOG_ERR, la_buf);

        abort();
    }

    TRACE_EXIT;
}

void CTmSync_Container::UpdateTmSyncState( int return_code )
{
    char                la_buf[MON_STRING_BUF_SIZE];
 
    const char method_name[] = "CTmSync_Container::UpdateTmSyncState";
    TRACE_ENTRY;
    
    if ( MyNode->GetTmSyncState() != SyncState_Abort )
    {
        if (( MyNode->GetTmSyncState() == SyncState_Start    ) ||
            ( MyNode->GetTmSyncState() == SyncState_Continue ) ||
            ( MyNode->GetTmSyncState() == SyncState_Commit   )   )
        {
            if ( return_code )
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - TmSync aborted" "\n", method_name, __LINE__);
                MyNode->SetTmSyncState( SyncState_Abort );
            }
            else
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - TmSync commited" "\n", method_name, __LINE__);
                MyNode->SetTmSyncState( SyncState_Commit );
            }
        }
        else
        {
            sprintf(la_buf, "[%s], Invalid SyncState (%d)! \n", method_name, MyNode->GetTmSyncState());
            mon_log_write(MON_TMSYNC_UPDATE_STATE_1, SQ_LOG_ERR, la_buf); 
            MyNode->SetTmSyncState( SyncState_Abort );
        }
        if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
           trace_printf("%s@%d" " - Physical Node " "%d"  " TmSyncState updated (" "%d" ")" "\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncState());
    }    

    TRACE_EXIT;
}

void CTmSync_Container::CommitTmDataBlock( int return_code )
{
    SyncState           state;
    char                la_buf[MON_STRING_BUF_SIZE];
 
    const char method_name[] = "CTmSync_Container::CommitTmDataBlock";
    TRACE_ENTRY;
    
    UpdateTmSyncState( return_code );

    // Loop here until the TM Sync has completed
    while (1)
    {
        if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
           trace_printf("%s@%d" " - Getting all nodes TmSyncState\n", method_name, __LINE__);

        ExchangeTmSyncState( false );
        state = Nodes->GetTmState( SyncState_Commit );
        if (( state == SyncState_Abort ) ||
            ( state == SyncState_Null  )   )
        {
            if ( state == SyncState_Null )
            {
                sprintf(la_buf, "[%s], Early termination! \n", method_name);
                mon_log_write(MON_TMSYNC_COMMITTMDATA_1, SQ_LOG_ERR, la_buf);
            }
            if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
               trace_printf("%s@%d" " - TmSyncAbort Sent" "\n", method_name, __LINE__);
            EndTmSync( MsgType_TmSyncAbort );
            if ( AbortPendingTmSync )
            {
                Monitor->TmSyncAbortPending();
            }
            break;
        }
        else if ( state == SyncState_Commit )
        {
            if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
               trace_printf("%s@%d" " - TmSyncCommit Sent" "\n", method_name, __LINE__);
            EndTmSync( MsgType_TmSyncCommit );
            break;
        }
        //usleep (5000);
    }    
    // End the TM sync processing cycle for my node.
    MyNode->SetTmSyncState( SyncState_Null );
    MyNode->SetTmSyncNid( -1 );
    ExchangeTmSyncState( false );
    if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
       trace_printf("%s@%d" " - Physical Node " "%d"  " TmSyncState updated (" "%d" ")" "\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncState());
    
    TRACE_EXIT;
}

int CTmSync_Container::CoordinateTmDataBlock ( struct sync_def *sync )
{
    const char method_name[] = "CTmSync_Container::CoordinateTmDataBlock";
    TRACE_ENTRY;
    if ( MyNode->GetState() == State_Down )
    {
        // For Virtual nodes: if we are down ... 
        // just return and continue processing like normal
        return MPI_SUCCESS;
    }

    if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
    {
        CNode *node = Nodes->GetNode( sync->pnid );
        if ( node )
        {
            trace_printf("%s@%d" " - Node %s (pnid=%d) TmSync initiated (sync nid=%d, sync state=%d)\n", method_name, __LINE__, node->GetName(), sync->pnid, node->GetTmSyncNid(), node->GetTmSyncState());
        }
    }
    if ( sync->pnid == MyPNID )
    {
//        if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
//        {
//           trace_printf("%s@%d" " - Node %s (pnid=%d) TmSync initiated (sync nid=%d, sync state=%d)\n", method_name, __LINE__, MyNode->GetName(), MyPNID, MyNode->GetTmSyncNid(), MyNode->GetTmSyncState());
//        }
        if ( MyNode->GetTmSyncNid() == -1)
        {
            // Our physical node requested the TM sync
            if ( Nodes->GetTmState( SyncState_Null ) != SyncState_Null)
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - Tm Sync already pending" "\n", method_name, __LINE__);
                return MPI_ERR_PENDING;
            }
            else
            {
                MyNode->SetTmSyncState( SyncState_Start );
                MyNode->SetTmSyncNid( sync->syncnid );
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                {
                   trace_printf("%s@%d" " - Master TmSync started" "\n", method_name, __LINE__);
                   trace_printf("%s@%d" " - Physical Node %d TmSyncState updated (nid=%d, state=%d)\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncNid(), MyNode->GetTmSyncState());
                }
                syncCycle_.lock();
                exchangeTmSyncData( sync, false );
                syncCycle_.unlock();
                ExchangeTmSyncState( false );
                if (( Monitor->tmSyncPNid_ == MyPNID                           ) &&
                    ( Nodes->GetTmState( SyncState_Start ) == SyncState_Start )   )
                {
                    // send unsolicited messages to other TMs in
                    // local physical node and wait for them to reply
                    if ( MyNode->GetLNodesCount() > 1 )
                    {
                        if ( PendingSlaveTmSync )
                        {
                            SendUnsolicitedMessages();
                            while (1)
                            {
                                if ( GetTmSyncReplies() == GetTotalSlaveTmSyncCount() )
                                {
                                    break;
                                }
                                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                                    trace_printf("%s@%d" " - Master waiting for Local Unsolicited TmSync reply, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );
                                    UnsolicitedCompleteWait();
                            }    
                        }
                    }
                         
                    // send reply to our TM for the sync request
                    if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                       trace_printf("%s@%d" " - Master TmSync reply" "\n", method_name, __LINE__);
                    CommitTmDataBlock(MPI_SUCCESS);
                    return MPI_SUCCESS;
                }
                else
                {
                    if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                       trace_printf("%s@%d" " - Tm Sync failed to start, tmSyncPNid_=%d, MyPNID=%d, " "TmSyncState=%d, expecting=%d\n", method_name, __LINE__, tmSyncPNid_, MyPNID, Nodes->GetTmState( SyncState_Start ), SyncState_Start);
                    if (MyNode->GetTmSyncState() == SyncState_Start)
                    {
                        MyNode->SetTmSyncState( SyncState_Null );
                        if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                           trace_printf("%s@%d" " - Physical Node " "%d"  " TmSyncState updated (" "%d" ")" "\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncState());
                    }       
                    return MPI_ERR_PENDING;
                }
            }
        }
        else
        {
            // Another logical node in my physical node requested a TM sync
            if ( MyNode->GetTmSyncState() == SyncState_Start )
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                {
                   trace_printf("%s@%d" " - Slave TmSync started on local Physical Node "  "%d" "\n", method_name, __LINE__, sync->pnid);
                   trace_printf("%s@%d" " - Physical Node " "%d"  " TmSyncState updated (" "%d" ")" "\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncState());
                }
                UnPackSyncData(sync);
            }
        }
    }
    else
    {
        // some other physical node requested a TM sync.
        if ( MyNode->GetTmSyncState() == SyncState_Null )
        {
            // Send sync data to our node's TM
            MyNode->SetTmSyncState( SyncState_Continue );
            if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
            {
               trace_printf("%s@%d" " - Slave TmSync started on Physical Node "  "%d" "\n", method_name, __LINE__, sync->pnid);
               trace_printf("%s@%d" " - Physical Node " "%d"  " TmSyncState updated (" "%d" ")" "\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncState());
            }
            UnPackSyncData(sync);
            ExchangeTmSyncState( true );
        }
        else
        {
            MyNode->SetTmSyncState( SyncState_Abort );
            if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
            {
               trace_printf("%s@%d" " - Collision with another TM trying to sync" "\n", method_name, __LINE__);
               trace_printf("%s@%d" " - Physical Node " "%d"  " TmSyncState updated (" "%d" ")" "\n", method_name, __LINE__, MyPNID, MyNode->GetTmSyncState());
            }
        }    
    }

    TRACE_EXIT;
    return 0;
}

void CTmSync_Container::EndTmSync( MSGTYPE type )
{
    CProcess           *mytm;
    CTmSyncReq         *req = Head;
    CTmSyncReq         *next;
    CLNode             *lnode;
    struct message_def *msg = NULL;
    struct message_def *notice;
    int                 count = 0;
    int                 orig_count = 0;
    char                la_buf[MON_STRING_BUF_SIZE];

    const char method_name[] = "CTmSync_Container::EndTmSync";
    TRACE_ENTRY;

    // send a commit or abort notice to my node's TM
    msg = new struct message_def;
    msg->type = type;
    msg->noreply = true;
    msg->u.request.type = ReqType_Notice;
    while (req)
    {
        next = req->GetNext();
        if ( !req->Unsolicited )
        {
            // my node
            if (trace_settings & TRACE_TMSYNC)
            {
                trace_printf("%s@%d - Original request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
            }
            msg->u.request.u.tm_sync_notice.orig_tag[orig_count] = req->Tag;
            msg->u.request.u.tm_sync_notice.orig_handle[orig_count] = req->Handle;
            orig_count++;
        }
        if (( req->Replicated                 ) &&
            ( req->Completed                  )   )
        {
            if (trace_settings & TRACE_TMSYNC)
            {
                trace_printf("%s@%d - Request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
            }
            if ( tmSyncPNid_ == MyPNID )
            {
                if ( MyNode->GetLNodesCount() > 1 )
                {
                    if ( req->Unsolicited )
                    {
                        msg->u.request.u.tm_sync_notice.nid[count] = req->Nid;
                        msg->u.request.u.tm_sync_notice.handle[count] = req->Handle;
                        count++;
                        if (trace_settings & TRACE_TMSYNC)
                        {
                            trace_printf("%s@%d - Count Unsolicited request (%p) nid=%d, handle=%d, tag=%d, count=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, count);
                        }
                    }
                }
                else
                {
                    msg->u.request.u.tm_sync_notice.nid[count] = req->Nid;
                    msg->u.request.u.tm_sync_notice.handle[count] = req->Handle;
                    count++;
                    if (trace_settings & TRACE_TMSYNC)
                    {
                        trace_printf("%s@%d - Count request (%p) nid=%d, handle=%d, tag=%d, count=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, count);
                    }
                }
            }
            else
            {
                msg->u.request.u.tm_sync_notice.nid[count] = req->Nid;
                msg->u.request.u.tm_sync_notice.handle[count] = req->Handle;
                count++;
                if (trace_settings & TRACE_TMSYNC)
                {
                    trace_printf("%s@%d - Count request (%p) nid=%d, handle=%d, tag=%d, count=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, count);
                }
            }
            req->DeLink(&Head,&Tail);
            delete req;
        }
        req = next;
    }
    msg->u.request.u.tm_sync_notice.count = count;
    msg->u.request.u.tm_sync_notice.orig_count = orig_count;

    lnode = MyNode->GetFirstLNode();
    for ( ; lnode  ; lnode = lnode->GetNextP() )
    {
        if ( lnode->GetState() == State_Up )
        {
            // the logical node is up and 
            // is not the requesting logical node
            mytm = lnode->GetProcessLByType( ProcessType_DTM );
            if ( !mytm )
            {
                sprintf(la_buf, "[%s], Can't find TM in node=%d\n", method_name, lnode->GetNid());
                mon_log_write(MON_TMSYNC_END, SQ_LOG_ERR, la_buf); 
    
                ReqQueue.enqueueDownReq(MyPNID);
            }
            else
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                {
                    trace_printf("%s@%d - Sending Notice to TM=%s (%d,%d)\n", method_name, __LINE__, mytm->GetName(), mytm->GetNid(), mytm->GetPid());
                    trace_printf("        %s %d handles\n", (type == MsgType_TmSyncCommit?" - Commit ":" - Aborted "), count);
                }
                notice = new struct message_def;
                memmove( notice, msg, sizeof(message_def) );

                SQ_theLocalIOToClient->putOnNoticeQueue( mytm->GetPid()
                                                       , mytm->GetVerifier()
                                                       , notice
                                                       , NULL);
            }
        }
    }

    
    if ( msg )
    {
        delete msg;
    }

    // Initialize Unsolicited message counters
    TmSyncReplies = 0;
    TmSyncReplyCode = MPI_SUCCESS;
    ReqsInBlock = 0;
    PendingSlaveTmSyncCount = 0;
    TotalSlaveTmSyncCount = 0;
    TRACE_EXIT;
}

void CTmSync_Container::EndPendingTmSync( struct sync_def *sync )
{
    CProcess           *mytm;
    CTmSyncReq         *req = Head;
    CTmSyncReq         *next;
    CLNode             *lnode;
    struct message_def *msg = NULL;
    struct message_def *notice;
    int                 count = 0;
    int                 orig_count = 0;
    char                la_buf[MON_STRING_BUF_SIZE];

    const char method_name[] = "CTmSync_Container::EndPendingTmSync";
    TRACE_ENTRY;

    // send a abort notice to my node's TM
    msg = new struct message_def;
    msg->type = MsgType_TmSyncAbort;
    msg->noreply = true;
    msg->u.request.type = ReqType_Notice;
    while (req)
    {
        next = req->GetNext();
        if ( !req->Unsolicited )
        {
            // my node
            if (trace_settings & TRACE_TMSYNC)
            {
                trace_printf("%s@%d - Original request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
            }
            msg->u.request.u.tm_sync_notice.orig_tag[orig_count] = req->Tag;
            msg->u.request.u.tm_sync_notice.orig_handle[orig_count] = req->Handle;
            orig_count++;
        }
        if (( sync->syncnid == req->Nid ) &&
            ( req->Replicated           ) &&
            ( !req->Unsolicited         ) &&
            ( req->Completed            )   )
        {
            if (trace_settings & TRACE_TMSYNC)
            {
                trace_printf("%s@%d - Request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
            }
            msg->u.request.u.tm_sync_notice.nid[count] = req->Nid;
            msg->u.request.u.tm_sync_notice.handle[count] = req->Handle;
            count++;
            if (trace_settings & TRACE_TMSYNC)
            {
                trace_printf("%s@%d - Count request (%p) nid=%d, handle=%d, tag=%d, count=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, count);
            }
            req->DeLink(&Head,&Tail);
            delete req;
        }
        req = next;
    }
    
    if ( count )
    {
        msg->u.request.u.tm_sync_notice.count = count;
        msg->u.request.u.tm_sync_notice.orig_count = orig_count;

        lnode = MyNode->GetFirstLNode();
        for ( ; lnode  ; lnode = lnode->GetNextP() )
        {
            if ( lnode->GetNid() == sync->syncnid &&
                 lnode->GetState() == State_Up )
            {
                // the logical node is up and 
                // is not the requesting logical node
                mytm = lnode->GetProcessLByType( ProcessType_DTM );
                if ( !mytm )
                {
                    sprintf(la_buf, "[%s], Can't find TM in node=%d\n", method_name, lnode->GetNid());
                    mon_log_write(MON_TMSYNC_END_PENDING, SQ_LOG_ERR, la_buf); 
    
                    ReqQueue.enqueueDownReq(MyPNID);
                }
                else
                {
                    if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                    {
                        trace_printf("%s@%d - Sending Notice to TM=%s (%d,%d)\n", method_name, __LINE__, mytm->GetName(), mytm->GetNid(), mytm->GetPid());
                        trace_printf("      - Aborted %d handles\n", count);
                    }
                    notice = new struct message_def;
                    memmove( notice, msg, sizeof(message_def) );
                    SQ_theLocalIOToClient->putOnNoticeQueue( mytm->GetPid()
                                                           , mytm->GetVerifier()
                                                           , notice
                                                           , NULL);
                }
            }
        }
    }

    
    if ( msg )
    {
        delete msg;
    }

    TRACE_EXIT;
}

void CTmSync_Container::ProcessTmSyncReply( struct message_def * msg )
{
    const char method_name[] = "CTmSync_Container::ProcessTmSyncReply";
    TRACE_ENTRY;

    CTmSyncReq *tmsync_req;

    if (trace_settings & (TRACE_REQUEST | TRACE_TMSYNC))
        trace_printf("%s@%d - Unsolicited TmSync Reply\n",
                     method_name, __LINE__);
    tmsync_req = FindTmSyncReq( msg->u.reply.u.unsolicited_tm_sync.handle );
    if (tmsync_req)
    {
        if (trace_settings & (TRACE_REQUEST | TRACE_TMSYNC))
            trace_printf("%s@%d - Unsolicited TmSync reply, handle=%d\n",
                         method_name, __LINE__, tmsync_req->Handle);
        if (msg->u.reply.u.unsolicited_tm_sync.return_code == MPI_SUCCESS)
        {
            TmSyncReplyCode |= msg->u.reply.u.unsolicited_tm_sync.return_code;
            tmsync_req->Completed = true;
            UnsolicitedComplete( msg );
            if ( tmSyncPNid_ == MyPNID )
            {
                if (trace_settings & (TRACE_REQUEST | TRACE_TMSYNC))
                    trace_printf("%s@%d - Local Unsolicited TmSync reply, handle="
                                 "%d\n", method_name, __LINE__,
                                 tmsync_req->Handle);
                if ( GetTmSyncReplies() == GetTotalSlaveTmSyncCount() )
                {
                    UpdateTmSyncState( TmSyncReplyCode );
                    UnsolicitedCompleteDone();
                }
            }
            else
            {
                if ( GetTmSyncReplies() == GetTotalSlaveTmSyncCount() )
                {
                    CommitTmDataBlock(TmSyncReplyCode);
                }
            }
        }
        else
        { // The Seabed callback has not been registered, try again
            if (trace_settings & (TRACE_REQUEST | TRACE_TMSYNC))
                trace_printf("%s@%d - Retrying Local Unsolicited TmSync, handle="
                             "%d\n", method_name, __LINE__,
                             tmsync_req->Handle);
            PendingSlaveTmSyncCount--;
            tmsync_req->Completed = false;
            SendUnsolicitedMessages();
        }
    }
    else
    {
        if (trace_settings & (TRACE_REQUEST | TRACE_TMSYNC))
            trace_printf("%s@%d" " - Can't find TmSync request, handle="  "%d" "\n", method_name, __LINE__, msg->u.reply.u.unsolicited_tm_sync.handle);
    }
    msg->noreply = true;

    if (trace_settings & TRACE_TMSYNC)
       trace_printf("%s@%d" " - Unsolicited TmSync notices, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );

    TRACE_EXIT;
}

void CTmSync_Container::ExchangeTmSyncState( bool bumpSync )
{
    struct sync_def sync;

    const char method_name[] = "CTmSync_Container::ExchangeTmSyncState";
    TRACE_ENTRY;

    sync.type = SyncType_TmSyncState;
    sync.pnid = MyPNID;
    sync.syncnid = MyNode->GetTmSyncNid();
    sync.state = MyNode->GetTmSyncState();
    sync.count = 0;
    sync.length = 0;
    syncCycle_.lock();
    exchangeTmSyncData( &sync, bumpSync );
    syncCycle_.unlock();

    TRACE_EXIT;
}

CTmSyncReq *CTmSync_Container::FindTmSyncReq( int handle )
{
    CTmSyncReq *req = Head;

    const char method_name[] = "CTmSync_Container::FindTmSyncReq";
    TRACE_ENTRY;
    while (req)
    {
        if( req->Handle == handle && req->Unsolicited)
        {
            if (trace_settings & TRACE_TMSYNC)
            {
                trace_printf("%s@%d - request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
            }
            break;
        }
        req = req->GetNext();
    }
    TRACE_EXIT;

    return req;
}

int CTmSync_Container::GetHandle( void )
{
    const char method_name[] = "CTmSync_Container::GetHandle";
    TRACE_ENTRY;
    if ( HandleSeq >= (MyPNID+1)*MAX_TM_HANDLES ) 
    {
        HandleSeq = MyPNID*MAX_TM_HANDLES;
    }
    TRACE_EXIT;
    return ++HandleSeq;
}

struct sync_def *CTmSync_Container::PackSyncData( void )
{
    char *ptr;
    CTmSyncReq *req = Head;
    struct sync_def *sync;

    const char method_name[] = "CTmSync_Container::PackSyncData";
    TRACE_ENTRY;

    sync = new struct sync_def;
    sync->type = SyncType_TmData;
    sync->pnid = MyPNID;
    sync->syncnid = -1;
    sync->state = SyncState_Start;
    sync->count = 0;
    sync->length = 0;
    ptr = sync->data;
    while ( req )
    {
        if (( !req->Unsolicited                                         ) &&
            ( !req->Replicated                                          ) &&
            (  sync->length+req->Length+(sizeof(int)*3) < MAX_SYNC_DATA ) && 
            (  sync->count                              < MAX_TM_SYNCS  )   )
        {
            if ( sync->syncnid == -1 )
            {
                // The first pending request's logical node wins
                sync->syncnid = req->Nid;
            }
            if ( sync->syncnid == req->Nid )
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - Packing TmSync request, nid=%d, handle="  "%d" "\n", method_name, __LINE__, req->Nid, req->Handle);
                // load request
                sync->count++;
                *((int*)ptr) = req->Nid;
                ptr += sizeof(int);
                *((int*)ptr) = req->Handle;
                ptr += sizeof(int);
                *((int*)ptr) = req->Length;
                ptr += sizeof(int);
                memmove( ptr, req->Data, req->Length );
                ptr += req->Length;
                req->Replicated = true;
                sync->length += ((sizeof(int)*3)+req->Length);
            }
            else
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - NOT Packing TmSync request, nid=%d, handle="  "%d" "\n", method_name, __LINE__, req->Nid, req->Handle);
            }
        }
        req = req->GetNext();
    }
    TRACE_EXIT;

    return sync;
}

void CTmSync_Container::ReQueue_TmSync( bool master )
{
    int collisionNid = -1;
    MyNode->SetTmSyncNid( -1 );
    CTmSyncReq *req = Head;
    CTmSyncReq *next;

    const char method_name[] = "CTmSync_Container::ReQueue_TmSync";
    TRACE_ENTRY; 
    while (req)
    {
        if (trace_settings & TRACE_TMSYNC)
        {
            trace_printf("%s@%d - request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
        }
        next = req->GetNext();
        if ( req->Replicated )
        {
            if ( master && !req->Unsolicited )
            {
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - Collision, resetting handle="  "%d" "\n", method_name, __LINE__, req->Handle);
                req->Replicated = false;
                collisionNid = req->Nid;
            }
            if ( master && req->Unsolicited && req->Nid == collisionNid ) 
            { 
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - Collision, deleting unsolicited handle="  "%d" "\n", method_name, __LINE__, req->Handle);
                req->DeLink(&Head,&Tail);
                delete req;
            }
            if ( !master && req->Unsolicited ) 
            { 
                if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
                   trace_printf("%s@%d" " - Collision, deleting unsolicited handle="  "%d" "\n", method_name, __LINE__, req->Handle);
                req->DeLink(&Head,&Tail);
                delete req;
            }
        }
        req = next;
    }
    TRACE_EXIT; 
}

CTmSyncReq *CTmSync_Container::Q_TmSync(int nid, int handle, char *data, int len, int tag, bool unsolicited)
{
    CTmSyncReq *req = NULL;
    char        la_buf[MON_STRING_BUF_SIZE];
    
    const char method_name[] = "CTmSync_Container::Q_TmSync";
    TRACE_ENTRY;

    if ( len > MAX_SYNC_DATA )
    {
        sprintf(la_buf, "[%s], Tm Sync length greater than max, len=%d. \n", method_name, len);
        mon_log_write(MON_TMSYNC_Q_TMSYNC, SQ_LOG_ERR, la_buf);
    }
    else
    {
        req = new CTmSyncReq(nid, handle, data, len, tag, unsolicited);
        if (Head == NULL)
        {
            Head = Tail = req;
        }
        else
        {
            Tail = Tail->Link (req);
        }
    }

    TRACE_EXIT;
    return req;
}

void CTmSync_Container::SendUnsolicitedMessages (void)
{
    int                 numTMs = 0;
    CTmSyncReq         *req = Head;
    CProcess           *mytm;
    CProcess           *tm = NULL;
    CLNode             *lnode;
    struct message_def *msg = NULL;
    struct message_def *notice;
    char                la_buf[MON_STRING_BUF_SIZE];

    const char method_name[] = "CTmSync_Container::SendUnsolicitedMessages";
    TRACE_ENTRY;

    while (req)
    {
        if (trace_settings & TRACE_TMSYNC)
        {
            trace_printf("%s@%d - request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
        }
        if ( req->Unsolicited && !req->Completed )
        { 
            if (!tm || req->Nid != tm->GetNid())
            { 
                // Get the TM that initiated the sync request
                tm = LNode[req->Nid]->GetProcessLByType( ProcessType_DTM );
            }
            if (!tm && NameServerEnabled)
            {
                if (trace_settings & (TRACE_INIT | TRACE_RECOVERY | TRACE_REQUEST | TRACE_SYNC | TRACE_TMSYNC))
                {
                    trace_printf( "%s@%d - Getting process from Name Server, nid=%d, type=ProcessType_DTM\n"
                                , method_name, __LINE__, req->Nid );
                }
            
                tm = Nodes->GetProcessLByTypeNs( req->Nid, ProcessType_DTM );
            }
            if ( tm )
            {
                // send all TmSync requests data to the local TM processes
                msg = new struct message_def;
                msg->type = MsgType_UnsolicitedMessage;
                msg->noreply = false;
                msg->u.request.type = ReqType_TmSync;
                msg->u.request.u.unsolicited_tm_sync.nid = tm->GetNid();
                msg->u.request.u.unsolicited_tm_sync.pid = tm->GetPid();
                msg->u.request.u.unsolicited_tm_sync.handle = req->Handle;
                memmove( msg->u.request.u.unsolicited_tm_sync.data, req->Data, req->Length );
                msg->u.request.u.unsolicited_tm_sync.length = req->Length;

                // count the number of candidate TMs
                if ( numTMs == 0 )
                {
                    lnode = MyNode->GetFirstLNode();
                    for ( ; lnode  ; lnode = lnode->GetNextP() )
                    {
                        if ( lnode->GetNid() != req->Nid )
                        {
                            numTMs++;
                        }
                    }
                    // calculate the total number of TmSync request 
                    TotalSlaveTmSyncCount = ReqsInBlock * numTMs;
                }

                if (trace_settings & TRACE_TMSYNC)
                   trace_printf("%s@%d" " - Unsolicited TmSync notices, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );
                   
                lnode = MyNode->GetFirstLNode();
                for ( ; lnode  ; lnode = lnode->GetNextP() )
                {
                    if ( lnode->GetState() == State_Up && 
                         lnode->GetNid()   != req->Nid      )
                    {
                        // the logical node is up and 
                        // is not the requesting logical node
                        mytm = lnode->GetProcessLByType( ProcessType_DTM );
                        if ( !mytm )
                        {
                            sprintf(la_buf, "[%s], Can't find TM in node=%d\n", method_name, lnode->GetNid());
                            mon_log_write(MON_TMSYNC_SEND_UNSOLICITED, SQ_LOG_ERR, la_buf); 
                
                            ReqQueue.enqueueDownReq(MyPNID);
                        }
                        else
                        {
                            if (trace_settings & TRACE_TMSYNC)
                            {
                                trace_printf("%s@%d - Sending Unsolicited TmSync to TM=%s (%d,%d)\n", method_name, __LINE__, mytm->GetName(), mytm->GetNid(), mytm->GetPid());
                                trace_printf("        from TM=%s (%d,%d)\n", tm->GetName(), tm->GetNid(), tm->GetPid());
                                trace_printf("        tag=%d\n", req->Tag);
                                trace_printf("        handle=%d\n", req->Handle);
                            }
                            notice = new struct message_def;
                            memmove( notice, msg, sizeof(message_def) );
                            SQ_theLocalIOToClient->putOnNoticeQueue( mytm->GetPid()
                                                                   , mytm->GetVerifier()
                                                                   , notice
                                                                   , NULL);
                            mytm->IncrUnsolTmSyncCount();
                            PendingSlaveTmSyncCount++;
                        }
                    }
                }
                if ( msg )
                {
                    delete msg;
                    msg = NULL;
                }
                if (NameServerEnabled)
                {
                    if (!MyNode->IsMyNode( tm->GetNid() ))
                    {
                        if (trace_settings & (TRACE_INIT | TRACE_RECOVERY | TRACE_REQUEST | TRACE_SYNC | TRACE_TMSYNC))
                        {
                            trace_printf( "%s@%d - Deleting clone process %s, (%d,%d:%d)\n"
                                        , method_name, __LINE__
                                        , tm->GetName()
                                        , tm->GetNid()
                                        , tm->GetPid()
                                        , tm->GetVerifier() );
                        }
                        Nodes->DeleteCloneProcess( tm );
                    }
                
                }
            }
            else
            {
                sprintf(la_buf, "[%s], Can't find requesting TM for nid= %d.\n", method_name, req->Nid);
                mon_log_write(MON_TMSYNC_UNPACKSYNCDATA, SQ_LOG_ERR, la_buf); 
                CNode *node = LNode[req->Nid]->GetNode();
                ReqQueue.enqueueDownReq( node->GetPNid() );
            }
        }
        req = req->GetNext();
    }

    if (trace_settings & TRACE_TMSYNC)
       trace_printf("%s@%d" " - Unsolicited TmSync notices, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );
       
    PendingSlaveTmSync = false;

    TRACE_EXIT;
}

void CTmSync_Container::TmSync( void )
{
    int              rc;
    struct sync_def *block;

    const char method_name[] = "CTmSync_Container::TmSync";
    TRACE_ENTRY;

    block = PackSyncData();
    if ( block->count )
    {
        if (trace_settings & TRACE_TMSYNC)
           trace_printf("%s@%d" " - Processing TmSync request" "\n", method_name, __LINE__);
        rc = CoordinateTmDataBlock( block );
        if ( rc != MPI_SUCCESS )
        {
            ReQueue_TmSync (true);
            if (trace_settings & TRACE_TMSYNC)
               trace_printf("%s@%d" " - Collision, no requests processed" "\n", method_name, __LINE__);
        }
    }

    if (trace_settings & TRACE_TMSYNC)
       trace_printf("%s@%d" " - PendingTmSync=%d, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, PendingSlaveTmSync, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );

    delete block;

    TRACE_EXIT;
}

void CTmSync_Container::TmSyncAbortPending( void )
{
    bool             notDone = true;
    struct sync_def *block = NULL;

    const char method_name[] = "CTmSync_Container::TmSyncAbortPending";
    TRACE_ENTRY;

    do
    {
        block = PackSyncData();
        if ( block->count )
        {
            if (trace_settings & TRACE_TMSYNC)
               trace_printf("%s@%d" " - Aborting pending TmSync\n", method_name, __LINE__);
            EndPendingTmSync( block );
        }
        else
        {
            notDone = false;
        }

        if (trace_settings & TRACE_TMSYNC)
           trace_printf("%s@%d" " - PendingTmSync=%d, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, PendingSlaveTmSync, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );

        delete block;
    }
    while( notDone );

    AbortPendingTmSync = false;

    TRACE_EXIT;
}

bool CTmSync_Container::TmSyncPending( void )
{
    bool rc = false;
    bool ready = true;
    int pnid = 0;
    CNode *node;
    CTmSyncReq *req;

    const char method_name[] = "CTmSync_Container::TmSyncPending";
    TRACE_ENTRY;
    if ( MyNode->GetState() == State_Down )
    {
        // For Virtual nodes: if we are down ... 
        // just return and continue processing like normal
        return false;
    }
    if (trace_settings & TRACE_TMSYNC)
       trace_printf("%s@%d" " - PendingTmSync=%d, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, PendingSlaveTmSync, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );

    if (( MyNode->GetTmSyncState() == SyncState_Abort ) &&
        ( tmSyncPNid_ != MyPNID ) &&
        ( GetTmSyncReplies() == GetTotalSlaveTmSyncCount() )   )
    {
        CommitTmDataBlock( MPI_ERR_UNKNOWN );
    }
    else
    {
        if ( ! MyNode->IsSpareNode() && MyNode->GetPhase() != Phase_Ready )
        {
            MyNode->CheckActivationPhase();
        }
        if ( Nodes->GetTmState( SyncState_Suspended ) == SyncState_Suspended )
        {
            if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
               trace_printf("%s@%d" " - TmSync suspended" "\n", method_name, __LINE__);
            return false;
        }
        if (trace_settings & TRACE_TMSYNC)
           trace_printf("%s@%d" " - PendingTmSync=%d, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, PendingSlaveTmSync, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );
        if ( PendingSlaveTmSync )
        {
            if (trace_settings & TRACE_TMSYNC)
               trace_printf("%s@%d" " - Pending TmSync, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );
            SendUnsolicitedMessages();
            return false;
        }
        if ( TmSyncReplies != GetTotalSlaveTmSyncCount() )
        {
            if (trace_settings & TRACE_TMSYNC)
               trace_printf("%s@%d" " - TmSync waiting for TmSync replies, total=%d, replies=%d, pending=%d\n", method_name, __LINE__, GetTotalSlaveTmSyncCount(), GetTmSyncReplies(), GetPendingSlaveTmSyncCount() );
            return false;
        }
    }

    // if no one is trying to tmsync and we have something ... then go for it
    node = Nodes->GetNode(pnid);
    while ( node )
    {
        if (( node->GetState() == State_Up && ! node->IsSpareNode() ) &&
            ( node->GetTmSyncState() != SyncState_Null )   )
        {
            if (trace_settings & TRACE_TMSYNC)
               trace_printf("%s@%d" " - TmSync needed, but not ready. TmSyncState("  "%d" ")" "\n", method_name, __LINE__, node->GetTmSyncState());
            ready = false;
            break;
        }
        node = Nodes->GetNode(++pnid);
    }

    if ( ready )
    {
        req = Head;
        while ( req )
        {
            if ( !req->Replicated )
            {
                if (trace_settings & TRACE_TMSYNC)
                   trace_printf("%s@%d" " - TmSync needed" "\n", method_name, __LINE__);
                rc = true;
                break;
            }
            req = req->GetNext();
        }
    }

    TRACE_EXIT;
    return rc;
}

void CTmSync_Container::UnPackSyncData(struct sync_def *sync)
{
    char               *ptr;
    int                 nid;
    int                 handle;
    int                 length;
    CTmSyncReq         *req;

    const char method_name[] = "CTmSync_Container::UnPackSyncData";
    TRACE_ENTRY;

    if (trace_settings & (TRACE_SYNC | TRACE_TMSYNC))
       trace_printf("%s@%d" " - UnPacking TmSync request(s)" "\n", method_name, __LINE__);

    // Initialize Unsolicited message counters
    TmSyncReplies = 0;
    TotalSlaveTmSyncCount = 0;
    TmSyncReplyCode = MPI_SUCCESS;
    ReqsInBlock = sync->count;

    // send all TmSync request data to TM process
    ptr = sync->data;
    while (sync->count)
    {
        // load request
        nid = *((int*)ptr);
        ptr += sizeof(int);
        handle = *((int*)ptr);
        ptr += sizeof(int);
        length = *((int*)ptr);
        ptr += sizeof(int);

        // save another nodes TmSync request in our nodes' request queue,
        // but mark it as having been replicated
        req = Q_TmSync(nid, handle, ptr, length, -1, true);
        req->Replicated = true;
        PendingSlaveTmSync = true;

        ptr += length;
        sync->count--;
    }
    TRACE_EXIT;
}

void CTmSync_Container::CancelUnsolicitedMessages( CProcess *tmProcess )
{
    CTmSyncReq *req = Head;
    const char method_name[] = "CTmSync_Container::CancelUnsolicitedMessages";
    TRACE_ENTRY;

    while (req)
    {
        if (trace_settings & TRACE_TMSYNC)
        {
            trace_printf("%s@%d - request (%p) nid=%d, handle=%d, tag=%d, unsol=%d, comp=%d\n", method_name, __LINE__, req, req->Nid, req->Handle, req->Tag, req->Unsolicited, req->Completed);
        }
        if ( req->Unsolicited && !req->Completed )
        {
            // Count it as a completed reply
            IncrTmSyncReplies();
            tmProcess->DecrUnsolTmSyncCount();
            if ( trace_settings & TRACE_TMSYNC )
            {
                trace_printf("%s@%d - Canceling Unsolicited TmSync to TM=%s (%d,%d)\n", method_name, __LINE__, tmProcess->GetName(), tmProcess->GetNid(), tmProcess->GetPid());
                trace_printf("        tag=%d\n", req->Tag);
                trace_printf("        handle=%d\n", req->Handle);
                trace_printf("%s@%d - TM=%s (%d,%d), TmSync count=%d  \n", method_name, __LINE__, tmProcess->GetName(), tmProcess->GetNid(), tmProcess->GetPid(), tmProcess->GetUnsolTmSyncCount());
            }

            int rc = sem_post( &UnsolicitedWaitSem );
            if ( rc && errno != EINTR)
            {
                int err = errno;
                char la_buf[MON_STRING_BUF_SIZE];
                sprintf(la_buf, "[%s], Can't post on unnamed semaphore! - errno=%d (%s)\n", method_name, err, strerror(errno));
                mon_log_write(MON_TMSYNC_CANCEL_UNSOL_MESSAGE_1, SQ_LOG_ERR, la_buf);
            }
        
            break;
        }
        req = req->GetNext();
    }

    TRACE_EXIT;
}

void CTmSync_Container::UnsolicitedComplete( struct message_def *msg )
{
    const char method_name[] = "CTmSync_Container::UnsolicitedComplete";
    TRACE_ENTRY;

    CLNode *lnode = MyNode->GetLNode( msg->u.reply.u.unsolicited_tm_sync.nid );
    assert( lnode );

    CProcess *mytm = lnode->GetProcessLByType( ProcessType_DTM );
    if ( !mytm )
    {
        char la_buf[MON_STRING_BUF_SIZE];
        sprintf(la_buf, "[%s], Can't find TM in node nid=%d\n", method_name, lnode->GetNid());
        mon_log_write(MON_TMSYNC_UNSOL_COMPLETE_1, SQ_LOG_ERR, la_buf); 

        ReqQueue.enqueueDownReq(MyPNID);
    }
    else
    {
        IncrTmSyncReplies();
        mytm->DecrUnsolTmSyncCount();
        if ( trace_settings & TRACE_TMSYNC )
        {
            trace_printf("%s@%d - TmSync reply from TM=%s (%d,%d), TmSync count=%d  \n", method_name, __LINE__, mytm->GetName(), mytm->GetNid(), mytm->GetPid(), mytm->GetUnsolTmSyncCount());
        }
    }

    TRACE_EXIT;
}

void CTmSync_Container::UnsolicitedCompleteDone( void )
{
    const char method_name[] = "CTmSync_Container::UnsolicitedCompleteDone";
    TRACE_ENTRY;

    int rc = sem_post( &UnsolicitedWaitSem );
    if ( rc && errno != EINTR)
    {
        int err = errno;
        char la_buf[MON_STRING_BUF_SIZE];
        sprintf(la_buf, "[%s], Can't post on unnamed semaphore! - errno=%d (%s)\n", method_name, err, strerror(errno));
        mon_log_write(MON_TMSYNC_UNSOL_COMPLETE_2, SQ_LOG_ERR, la_buf);
    }

    TRACE_EXIT;
}

void CTmSync_Container::UnsolicitedCompleteWait( void )
{
    const char method_name[] = "CTmSync_Container::UnsolicitedCompleteWait";
    TRACE_ENTRY;

    int rc;
    bool waitSomeMore = true;

    do
    {
        rc = sem_wait( &UnsolicitedWaitSem );
        if ( rc )
        {
            if ( errno != EINTR )
            {
                int err = errno;
                char la_buf[MON_STRING_BUF_SIZE];
                sprintf(la_buf, "[%s], Can't wait on unnamed semaphore! - errno=%d (%s)\n", method_name, err, strerror(errno));
                mon_log_write(MON_TMSYNC_UNSOL_COMPLETE_WAIT_1, SQ_LOG_ERR, la_buf);
                waitSomeMore = false;
            }
        }
        else
        {
            waitSomeMore = false;
        }
    }
    while( waitSomeMore );

    TRACE_EXIT;
}