// Session.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2005

#include "StdAfx.h"
#include "Ceda/Core/cxUtils/Tracer.h"
#include "Ceda/Core/cxUtils/PseudoRandom.h"
#include "Site.h"
#include "Session.h"

using namespace ceda;


/*

A site stores a set of sessions (for each other site it communicates with).  There are two session
objects for a single pair of sites because each site store its own session object.

A Session object has a well defined sender site and receiver site.  There is a suffix U, and rhv.

We are interested in an accurate simulation.  Therefore messages may be in transit for an 
arbitarily long amount of time.  So sending an operation is a distinct event from receiving the
operation.

Important events

*   A site tries to connect to another site.
    It sends the hv in this message.

*   A site accepts a connection from another site.  It responds with its hv.

*   A session is destroyed.  This is done atomically.

*   An operation is sent from a site.  A copy of the operation is queued in a receiever queue
    in the session of the receiver site.

*   An operation is received.  I.e. a queued operation is popped and processed.

*   A site is added to the system

*   A site is removed from the system

*   A local operation is generated, executed and appended to the local HB

*   A session is used to send all remaining operations in order to check for convergence.

*/

///////////////////////////////////////////////////////////////////////////////////////////////////
// Session

//ssize_t g_sessions = 0;

Session::Session() :
    m_status(NOT_CONNECTED),
    m_localSite(NULL),
    m_remoteSession(NULL),
    m_initialisedSuffix(false)
{
    //++g_sessions;
    //Tracer() << "Creating session\n";
}

Session::~Session()
{
    //--g_sessions; 
    //Tracer() << "Destroying session " << g_sessions << '\n';
    Clear();
}

void Session::Init(Site* localSite, Session* remoteSession)
{
    cxAssert(m_status == NOT_CONNECTED);

    m_localSite = localSite;
    m_remoteSession = remoteSession;
}

ssize_t Session::GetSiteId() const 
{ 
    cxAssert(m_localSite);
    return m_localSite->m_s; 
}

ssize_t Session::GetRemoteSiteId() const
{
    cxAssert(m_remoteSession);
    return m_remoteSession->GetSiteId();
}

void Session::InitRhv(const VectorTime& rhv)
{
    cxAssert(m_suffix.empty());
    m_rhv = rhv;
}

void Session::OnAppendOp(Session* session, const Operation& o)
{
    // Local operations have session = NULL
    // We treat incoming operations from a different session like local operations. These needs to 
    // be appended to U.  It is important to clone the operation or else transformations on U will 
    // affect the operations in the HB.
    if (m_initialisedSuffix && session != this)
    {
        Operation* copy = new Operation(o);
        m_suffix.push_back(copy);    
    }
}

void Session::InitiateConnection()
{
    cxAssert(!m_initialisedSuffix);
    cxAssert(m_status == NOT_CONNECTED);
    cxAssert(m_receiverQueue.empty());

    cxAssert(m_remoteSession);
    cxAssert(m_remoteSession->m_status == NOT_CONNECTED);
    cxAssert(m_remoteSession->m_receiverQueue.empty());

    TraceIndenter ids(12 * GetSiteId());

    if (gp_debug) Tracer() << "\nS" << GetSiteId()
                           << ": Initiating connection to S" << GetRemoteSiteId()
                           << " sending hv = " << m_localSite->m_hb.m_hv
                           << '\n';

    m_remoteSession->InitRhv(m_localSite->m_hb.m_hv);
    m_remoteSession->m_status = RECEIVED_HV;
    m_status = SENT_HV;
}

void Session::ReciprocateConnection()
{
    cxAssert(!m_initialisedSuffix);
    cxAssert(m_status == RECEIVED_HV);
    cxAssert(m_remoteSession);
    cxAssert(m_remoteSession->m_status == SENT_HV);

    TraceIndenter ids(12 * GetSiteId());

    if (gp_debug) Tracer() << "\nS" << GetSiteId()
                           << ": Reciprocating connection to S" << GetRemoteSiteId()
                           << " sending hv = " << m_localSite->m_hb.m_hv
                           << '\n';

    m_remoteSession->InitRhv(m_localSite->m_hb.m_hv);
    m_remoteSession->m_status = CONNECTED;
    m_status = CONNECTED;
}

void Session::Clear()
{
    m_initialisedSuffix = false;
    m_status = NOT_CONNECTED;

    {
        for (std::deque<QUEUE_ENTRY>::iterator i = m_receiverQueue.begin() ; i != m_receiverQueue.end() ; ++i)
        {
            //Tracer() << "Delete op " << i->second << '\n';
            delete i->second;
        }    
        m_receiverQueue.clear();
    }

    {
        for (std::deque<Operation*>::iterator i = m_suffix.begin() ; i != m_suffix.end() ; ++i)
        {
            //Tracer() << "Delete suffix op " << *i << '\n';
            delete *i;
        }
        m_suffix.clear();
    }

    m_rhv.Clear();
}

void Session::BreakConnection()
{
    cxAssert(m_status == CONNECTED);
    cxAssert(m_remoteSession);
    cxAssert(m_remoteSession->m_status == CONNECTED);

    TraceIndenter ids(12 * GetSiteId());

    if (gp_debug) Tracer() << "\nBreaking connection from S" << GetSiteId()
                           << " to S" << GetRemoteSiteId() << '\n';

    Clear();
    m_remoteSession->Clear();
}

bool Session::SendOperation()
{
    cxAssert(m_status == CONNECTED);
    cxAssert(m_remoteSession);

    // Find next operation to send, if any.  This is based on rhv which is an underestimate of what
    // operations are already present on the receiver.

    VectorTime v;
    Operation* o = m_localSite->m_hb.GetNextOperationToSend(m_rhv, v);

    if (o)
    {
        TraceIndenter ids(12 * GetSiteId());

        if (gp_debug) Tracer() << "\nS" << GetSiteId()
                               << ": Sending operation " << *o << " v= " << v 
                               << " to S" << GetRemoteSiteId()
                               << '\n';
        TraceIndenter indent(4);

        // Update rhv to account for sending o.  This ensures that we won't try to send this operation again!
        m_rhv.Add(o->m_opid.id, o->m_opid.t + 1);

        // Add the operation to the receiver's queue.
        m_remoteSession->m_receiverQueue.push_back(QUEUE_ENTRY(v,o));
        return true;
    }
    else
    {
        // No more operations to send
        return false;
    }
}

void DualIT(Operation &o, std::deque<Operation*>& L)
{
    for (ssize_t i=0 ; i < L.size() ; ++i)
    {
        TracedDualIT(o, *L[i]);
    }
}

void TraceOpList(xostream& os, const std::deque<Operation*>& L)
{
    os << '[';
    for (ssize_t i=0 ; i < L.size() ; ++i)
    {
        if (i > 0) os << ',';
        os << *L[i];
    }
    os << ']';
}

bool Session::ProcessNextReceivedOp()
{
    cxAssert(m_status == CONNECTED);
    cxAssert(m_localSite);
    
    if (m_receiverQueue.empty())
    {
        return false;
    }
    else
    {
        QUEUE_ENTRY e = m_receiverQueue.front();
        m_receiverQueue.pop_front();
        const VectorTime& v = e.first;
        Operation* o = e.second;
        cxAssert(o);

        TraceIndenter ids(12 * GetSiteId());

        if (gp_debug) Tracer() << "\nS" << GetSiteId()
                               << ": Received remote operation " << *o << " v= " << v
                               << " from S" << GetRemoteSiteId()
                               << '\n';
        TraceIndenter indent(4);

        HistoryBuf& hb = m_localSite->m_hb;

        if (ExtentContainsOp(hb.m_hv, *o))
        {
            delete o;
        }
        else
        {
            if (m_initialisedSuffix)
            {            
                CalculateSuffix(v,m_suffix);
            }
            else
            {
                cxAssert(m_suffix.empty());

                m_localSite->m_hb.CalculateSuffix(v, m_suffix);
                m_initialisedSuffix = true;
            }

            DualIT(*o,m_suffix);
            m_localSite->ExecuteAndAppendOp(this, o);
        }
        return true;
    }
}