// CompSetOp3.cpp
//
// Author David Barrett-Lennard
// (C)opyright Cedanet Pty Ltd 2009

#include "stdafx.h"
#include "Ceda/Core/cxUtils/xvector.h"
#include "Ceda/Core/cxUtils/HPTime.h"
#include "Ceda/Core/cxUtils/PseudoRandom.h"
#include "Ceda/Core/cxUtils/Assert.h"
#include "Ceda/Core/cxUtils/Tracer.h"
#include "Ceda/Core/cxUtils/ListToOStream.h"
#include "Opid.h"
#include "VectorTime.h"
#include "CompSetOp3.h"

/*

For a given FieldId for a field of type set<Key>, and for a given value of Key we record the 
following information:

    i:  the set of (s,t) for all insertions of that key that have ever occurred
    
    d:  the set of (s,t) for all insertions of that key that have been causally masked or 
        dominated by some deletion.
    
The key is present if i\d != {}. i.e. there exist insertions of the key that haven't been causally
deleted.
    
Then O1+O2 is defined as follows:

    (O1+O2).i =  O1.i U O2.i
    (O1+O2).d =  O1.d U O2.d
     
which clearly converges.

We can instead record i,d using vector times:

    i:  the smallest vector time containing the set of (s,t) for all insertions of that key 
        that have ever occurred 
        
    d:  the smallest vector time containing the set of (s,t) for all insertions of that key 
        that have been masked or dominated by some deletion.
        
The key is not present if i <= d. (i <= d means i is a subset of d)

    (O1+O2).i =  O1.i union O2.i
    (O1+O2).d =  O1.d union O2.d
    

Sending of deltas
-----------------

Unfortunately we aren't recording (s,t) of deletes, so therefore it is not possible to use rhv 
(an underestimate of what's present on the remote site) to determine whether i,d need to be
sent for a given key.

Consider that for a given (s,i,d) entry we also record (s',t') of the delete that cause the
last update to d.


Map structure 
-------------

    fid --> key --> s --> (i,d,s',t')

*/

using namespace ceda;


namespace SetTest3
{

const int NUMKEYS = 1;

// The type of a elements in a set
typedef int Key;

// Site Identifier for a site.  There must be a total ordering defined.
typedef int SiteId;


static Key PickKey()
{
    return gfnGetUniformDistInteger(0,NUMKEYS);
}

static bool enableTrace = false;

// Information about a single key value that may be recorded in a set<Key> field.
// 'p' is a flag for whether the key value is currently present.
// 'v' records the operations that have contributed to the value of 'p'.
struct KeyInfo
{
    // Key is present if it is not the case that i is a subset of d.
    bool present() const { return !(i <= d); }

    bool operator==(const KeyInfo& rhs) const
    {
        return i == rhs.i && d == rhs.d;
    }

    VectorTime i;
    VectorTime d;
};

xostream& operator<<(xostream& os, const KeyInfo& f)
{
    os << "(i=" << f.i << "  d=" << f.d << ')';
    return os;
}


// A site sends a delta to another site in order to bring it up to date.
// The sender sends its hv as well as the keys that may need to be inserted/deleted on the 
// receiving site.
struct Delta
{
    VectorTime hv;
    std::map<Key,KeyInfo> keys;
};

xostream& operator<<(xostream& os, const Delta& d)
{
    os << "hv=" << d.hv 
       << "  keys=" << d.keys; 
    return os;
}


///////////////////////////////////////////////////////////////////////////////////////////////////
// Site

struct Site
{
    Site()
    {
        t = 0;
        s = -1;
        
        // Initially no keys are present
        for (int i=0 ; i < NUMKEYS ; ++i)
        {
            keys[i] = KeyInfo();
        }
    }
    
    bool Converged(const Site& rhs) const
    {
        return keys == rhs.keys;
    }

    // Generate local insert/delete operation of 'key'
    void GenerateLocalOp(Key key, bool insert)
    {
        ++t;
        if (insert)
        {
            keys[key].i.Add(s,t);
        }
        else
        {
            // Deletes should always be a subset of what has been inserted
            cxAssert(keys[key].d <= keys[key].i);
            
            keys[key].d = keys[key].i;
        }
        hv.Add(s,t);
    }
    
    // Retrieve a delta to be sent based on rhv which is some underestimate of what 
    // operations are already present on the remote site.
    void GetDeltaToSend(const VectorTime& rhv, Delta& d) const
    {
        d.hv = hv;
        for (std::map<Key,KeyInfo>::const_iterator i = keys.begin() ; i != keys.end() ; ++i)
        {
            Key k = i->first;
            const KeyInfo& ki = i->second;
            
            //Tracer() << "Get delta : rhv = " << rhv << "  ki.v = " << ki.v << '\n';
            //if (!(ki.v <= rhv))
            {
                d.keys[k] = ki;
            }
        }
    }

    /*
    Apply the given delta to this site.
    
    A necessary condition for applying a given insert/delete specified in the delta is that 
    the remote operation hasn't already been incorporated at this receiving site.  
    i.e. 

        dk.t >= hv[sk.s]

    This ensures we will only apply something new.  Assuming this is the case, then a 
    sufficient condition for applying the operation is that the local dominating operation 
    was already present at the time the delta was gathered by the sending site (because 
    that implies that the operation in the delta dominates the local operation). i.e.

        sk.t < d.hv[sk.s]
        
    If this is not the case then it follows that the local operation and the remote 
    operation are concurrent.  In that case the merge uses the following table
    
                      delete      insert
            -------------------------------          
            delete    delete      insert
    
            insert    insert      insert
    
    */
    void ApplyReceivedDelta(const Delta& d)
    {
        for (std::map<Key,KeyInfo>::const_iterator i = d.keys.begin() ; i != d.keys.end() ; ++i)
        {
            Key k = i->first;
            const KeyInfo& dk = i->second;
            KeyInfo& sk = keys[k];
            
            //Tracer() << "\n\ndelta is new : " << (dk.t >= hv(sk.s)) << "\n";
            //Tracer() << "dk.t = " << dk.t << '\n';
            //Tracer() << "hv = " << hv << '\n';
            //Tracer() << "hv(sk.s) = " << hv(sk.s) << '\n';
            
            //Tracer() << "local op was present when delta calculated on remote site : " << (sk.t < d.hv(sk.s)) << "\n";
            
            //if (dk.t >= hv(sk.s)) Tracer() << "\n\ndelta not seen before\n"; else Tracer() <<
            //if (sk.t < d.hv(sk.s)) Tracer() << "local op was present when delta calculated on remote site\n";

            /*
            if (dk.v <= sk.v) ;
            else
            {
                if (sk.v <= dk.v)
                {
                    sk = dk;
                }
                else
                {
                    sk.p = sk.p || dk.p;
                    sk.v.UnionWith(dk.v);
                }
            }
            */
            
            sk.i.UnionWith(dk.i);
            sk.d.UnionWith(dk.d);
        }
        hv.UnionWith(d.hv);
    }

    SiteId s;          // Local site id
    int t;             // Next t for locally generated operations
    VectorTime hv;     // Represents the set of operations that have already been applied
    std::map<Key,KeyInfo> keys;  // Records the value of a set<T> field plus information about changes
};

xostream& operator<<(xostream& os, const Site& site)
{
    os << "s=" << site.s 
       << "  t=" << site.t
       << "  hv=" << site.hv
       << "  keys=" << site.keys; 
    return os;
}


///////////////////////////////////////////////////////////////////////////////////////////////////
// CompSetTest

struct CompSetTest
{
    CompSetTest(int numSites);
    int PickSite();
    void DoLocalOp();
    void TrySendOp(int s1, int s2);
    void PickDifferentSites(int& s1, int& s2);
    void SendOp();
    void TestConvergenceOfTwoSites();

    xvector<Site> m_sites;
};

xostream& operator<<(xostream& os, const CompSetTest& v)
{
    for (int i=0 ; i < v.m_sites.size() ; ++i)
    {
        os << v.m_sites[i] << '\n';
    }
    return os;
}


CompSetTest::CompSetTest(int numSites) :
    m_sites(numSites)
{
    // Initialise siteids
    for (int i=0 ; i < numSites ; ++i)
    {
        m_sites[i].s = i;
    }
}

int CompSetTest::PickSite()
{
    return gfnGetUniformDistInteger(0,m_sites.size());
}

void CompSetTest::DoLocalOp()
{
    int si = PickSite();
    Key key = PickKey();
    Site& site = m_sites[si];
    bool insert = !site.keys[key].present();      // Insert iff not present
    
    site.GenerateLocalOp(key,insert);
    if (enableTrace)
    {
        Tracer() << "\nGenerated local operation on S" << si << '\n';
        IndentTrace indent(4);
        Tracer() << *this;
    }
}

void CompSetTest::TrySendOp(int s1, int s2)
{
    // receiver = site1
    Site& receiver = m_sites[s1];

    // sender = site2
    Site& sender = m_sites[s2];
    
    Delta d;
    sender.GetDeltaToSend(receiver.hv,d);
    receiver.ApplyReceivedDelta(d);
    
    if (enableTrace)
    {
        Tracer() << "\nSending delta " << d << " from S" << s2 << " to S" << s1 << '\n';
        IndentTrace indent(4);
        Tracer() << *this;
    }
}

void CompSetTest::PickDifferentSites(int& s1, int& s2)
{
    do
    {
        s1 = PickSite();    // sender
        s2 = PickSite();    // receiver
    } while (s1 == s2);
}

void CompSetTest::SendOp()
{
    int s1,s2;
    PickDifferentSites(s1,s2);
    TrySendOp(s1,s2);
}

void CompSetTest::TestConvergenceOfTwoSites()
{
    int s1,s2;
    PickDifferentSites(s1,s2);
    
    if (enableTrace)
    {
        Tracer() << "\nVerifying sites S" << s1 << " and S" << s2 << " converge\n";
    }
    IndentTrace indent(4);
    TrySendOp(s1,s2);
    TrySendOp(s2,s1);
    
    if (enableTrace)
    {
        Tracer() << "m_sites[s1].hv = " << m_sites[s1].hv << '\n';
        Tracer() << "m_sites[s2].hv = " << m_sites[s2].hv << '\n';
    }
    cxAssert(m_sites[s1].hv == m_sites[s2].hv);
    
    //Tracer() << "m_sites[s1].m_field = " << m_sites[s1].m_field << '\n';
    //Tracer() << "m_sites[s2].m_field = " << m_sites[s2].m_field << '\n';
    //cxAssert(m_sites[s1].keys == m_sites[s2].keys);
    
    cxAssert(m_sites[s1].Converged(m_sites[s2]));
    
    if (enableTrace)
    {
        Tracer() << "Converged to " << m_sites[s1].keys << '\n';
    }
}


///////////////////////////////////////////////////////////////////////////////////////////////////
// DoCompSetTest

void DoCompSetTest(int numSites, int count1, int count2)
{
    // Need to be able to pick two different sites!
    cxAssert(numSites >= 2);
    
    for (int c1 = 0 ; c1 < count1 ; ++c1)
    {
        if (c1 % 10000 == 0) 
        {
            xcout << "Time: " << HPTime::GetCurrentTime() << "  c1 = " << c1 << endl;
        }
        
        if (enableTrace) gfnClearTheTraceFile();
        
        CompSetTest at(numSites);
        
        if (enableTrace)
        {
            Tracer() << "\n------------------------------------------------------\n" << at << '\n';
        }
        
        for (int c2=0 ; c2 < count2 ; ++c2)
        {
            int c = gfnGetUniformDistInteger(0,10);
            
            if (c < 4)
            {
                at.DoLocalOp();
            }
            else if (c < 8)
            {
                at.SendOp();
            }
            else
            {
                at.TestConvergenceOfTwoSites();
            }
        }
    }
}


} // namespace SetTest3


void CompSetTest3()
{
    Tracer() << "Set test 3\n";
    
    // This test ran successfully : 8 July 2009
    SetTest3::DoCompSetTest(3,1000000,30);

    Tracer() << "End test\n";
}