// CompSetOp4.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 "CompSetOp4.h"

/*
We have a solution for deltas on an int field that is very efficient.  It involves recording

    s --> t,o       (s,t) - the last operation that applied an offset
                      o   - the total offset applied by that site.
                      
    insert(s,t,o)
    {
        O[s].t = t;
        O[s].o += o;
    }
    
    getdelta(rhv)
    {
        for each s
            if (O[s].t >= rhv(s)) d[s] = O[s]
    }
    
    merge(d)
    {
        tot = 0;
        for each s in d
        {
            if (d[s].t > O[s].t)
            {
                tot += d[s].o - O[s].o;
                O[s].o = d[s].o;
            }
        }
    }

Let 

    ni: cardinality of the set of (s,t) for all insertions of that key that have ever occurred
    
    nd: cardinality of the set of (s,t) for all insertions of that key that have been 
        causally masked or dominated by some deletion.

Consider that we use this solution to record x = ni-nd. 

So for each local 

    insertion we apply delta +1 to x
    deletion we apply  delta -x to x (so it becomes zero)

So space required is:

    fid --> key -->  s  -->  (t,o)
                     x
    
Note that the total offset o applied by a given site can be negative!

Problem
-------

This solution breaks our requirements for intention preservation, because it allows sites
to inadvertently delete concurrently inserted elements.

Example with two sites.

S1:  insert                     +1
S0:  receive insert from S1
S1:  delete                     -1
S0:  delete                     -1
S1:  receive delete from S0  --> x = -1
*/

using namespace ceda;


namespace SetTest4
{

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 = true;

struct SiteOffset
{
    SiteOffset() : t(-1), o(0) {}
    
    bool operator==(const SiteOffset& rhs) const 
    { 
        return t == rhs.t && o == rhs.o; 
    }
    
    int t;  // t of last operation that applied an offset by the site
    int o;  // total offset by the site
};

inline xostream& operator<<(xostream& os, const SiteOffset& so)
{
	os << "(t=" << so.t << " o=" << so.o << ')';
	return os;
}


// 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
{
    KeyInfo() : x(0) {}
    
    bool present() const { return x > 0; }

    bool operator==(const KeyInfo& rhs) const
    {
        return ms == rhs.ms && x == rhs.x;
    }
    
    void Insert(const SiteId& s, int t, int offset)
    {
        // Note that this creates a new entry if there wasn't one already
        SiteOffset& so = ms[s];
        cxAssert(so.t < t);
        so.t = t;
        so.o += offset;
        x += offset;
        cxAssert(x >= 0);
    }

    void GetRFactor(const VectorTime& v, KeyInfo& rf) const
    {
        // We only send map entries for sites that have performed an edit outside X(v)
        for (MAP::const_iterator i = ms.begin() ; i != ms.end() ; ++i)
        {
            const SiteId& s = i->first;
            const SiteOffset& so = i->second;
            if (so.t >= v(s))
            {
                rf.ms[s] = so;
            }
        }
    }

    void Merge(const KeyInfo& rf)
    {
        int totalOffset = 0;
        for (MAP::const_iterator i = rf.ms.begin() ; i != rf.ms.end() ; ++i)
        {
            const SiteId& s = i->first;
            const SiteOffset& so_src = i->second;
            SiteOffset& so_dst = ms[s];
            if (so_src.t > so_dst.t)
            {
                totalOffset += so_src.o - so_dst.o;
                so_dst = so_src;
            }
        }
        x += totalOffset;
        cxAssert(x >= 0);
    }

    int x;
    typedef std::map<SiteId, SiteOffset> MAP;
    MAP ms;
};

xostream& operator<<(xostream& os, const KeyInfo& f)
{
    os << "(x=" << f.x << "  ms=" << f.ms << ')';
    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].Insert(s,t,+1);
        }
        else
        {
            keys[key].Insert(s,t,-keys[key].x);
        }
        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;
            ki.GetRFactor(rhv, d.keys[k]);
        }
    }

    /*
    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];
            sk.Merge(dk);
        }
        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 SetTest4


void CompSetTest4()
{
    Tracer() << "Set test 4\n";
    SetTest4::DoCompSetTest(2,10000000,5);
    Tracer() << "End test\n";
}