|| || || || || ||

---

DistributedAccess.h

Go to the documentation of this file.

#ifndef Impala_Core_Matrix_DistributedAccess_h
#define Impala_Core_Matrix_DistributedAccess_h

#include "Basis/String.h"
#include "Core/Matrix/Mat.h"
#include "Core/Matrix/MatFunc.h"
#include "Core/Array/PrintData.h"
#include "Core/Array/Set.h"
#include "Core/Array/SetPart.h"
#include "Link/Mpi/MpiFuncs.h"
#include "Util/PropertySet.h"
#include "Core/Matrix/MatTranspose.h"
#include "Core/Matrix/GetDiagonal.h"
#include "Core/Matrix/GetRow.h"
#include "Core/Matrix/GetColumn.h"
#include "Core/Vector/VectorTem.h"
#include "Core/Table/QuidTable.h"
#include "Core/Table/Copy.h"

#include "Core/Matrix/IDistributedAccess.h"
#include "Core/Matrix/VirtualMatrix.h"

namespace Impala
{
namespace Core
{
namespace Matrix
{

    class DistributedAccess : public IDistributedAccess, public VirtualMatrix
{
public:

    class IndexConverter
    {
    public:
        IndexConverter(int totalSize, int partCount)
            : mTotalSize(totalSize), mPartCount(partCount)
        {
            mBlockSize = mTotalSize / mPartCount + 1;
            ILOG_DEBUG_NODE("IndexConverter created: totalSize=" << mTotalSize
                            << "; partCount=" << mPartCount << "; blockSize="
                            << mBlockSize);
        }
        
        int
        PartToIndex(int part) const
        {
            return Min(part * mBlockSize, mTotalSize);
        }
        
        int
        IndexToPart(int index)
        {
            if ((index < 0) || (index >= mTotalSize))
            {
                ILOG_ERROR_NODE("DistributedAccess::IndexConverter::IndexToPart"
                                <<" range error: "<< index <<" not in [0,"<<
                                mTotalSize <<")");
                return 0;
            }
            return index / mBlockSize;
        }
        
        int
        PartCount() const
        {
            return mPartCount;
        }
        
        int
        TotalSize() const
        {
            return mTotalSize;
        }

    private:

        const int mTotalSize;
        const int mPartCount;
        int mBlockSize;
    };


    typedef Vector::VectorTem<Real64> Vector64;

    DistributedAccess()
    {
        int cpuCount = Link::Mpi::NrProcs();
        int partCount = sqrt((double)cpuCount);
        if (partCount * partCount != cpuCount)
        {
            ILOG_ERROR("Unsupported number of nodes, number of nodes MUST be"
                       << " sqare of a natural number");
            mRows = 0;
            return;
        }

        mRows = new IndexConverter(0, partCount);
        mColumns = new IndexConverter(0, partCount);
        mStartNode = 0;
        mNodeCount = cpuCount;
        mRowQuids = new Table::QuidTable();
        mColumnQuids = new Table::QuidTable();
        mHasOwnAverages = false;
    }

    DistributedAccess(int totalRows, int totalColumns,
                      int rowParts, int columnParts,
                      int startNode, int nodeCount)
    {
        mRows = new IndexConverter(totalRows, rowParts);
        mColumns = new IndexConverter(totalColumns, columnParts);
        mStartNode = startNode;
        mNodeCount = nodeCount;
        mRowQuids = new Table::QuidTable();
        mColumnQuids = new Table::QuidTable();
        mHasOwnAverages = false;
    }

    virtual
    ~DistributedAccess()
    {
        for (int i=0 ; i<mParts.size() ; i++)
            delete mParts[i];
        mParts.clear();
        if (mRowQuids)
            delete mRowQuids;
        if (mColumnQuids)
            delete mColumnQuids;
        if (mRows)
            delete mRows;
        if (mColumns)
            delete mColumns;
    }

    bool
    Valid(bool ctorOnly, bool withParts)
    {
        if (mRows == 0)
            return false;
        if (ctorOnly)
            return true;
        if ((mColumnQuids->Size() == 0) || (mRowQuids->Size() == 0))
            return false;
        if (!withParts)
            return true;
        return (mParts.size() != 0);
    }

    bool
    Symmetric()
    {
        if (mRows == 0)
            return false;
        if (GetRows() != GetColumns())
            return false;
        // Check some quids?
        return true;
    }

    // Specialiation of VirtualMatrix
    virtual int
    NrRow()
    {
        return mRows->TotalSize();
    }

    // Specialiation of VirtualMatrix
    virtual int
    NrCol()
    {
        return mColumns->TotalSize();
    }

    int
    GetRows()
    {
        return mRows->TotalSize();
    }

    void
    SetRows(int nr)
    {
        int partCount = mRows->PartCount();
        delete mRows;
        mRows = new IndexConverter(nr, partCount);
    }

    int
    GetColumns()
    {
        return mColumns->TotalSize();
    }

    void
    SetColumns(int nr)
    {
        int partCount = mColumns->PartCount();
        delete mColumns;
        mColumns = new IndexConverter(nr, partCount);
    }

    int
    GetRowPartCount() const
    {
        return mRows->PartCount();
    }

    int
    GetColumnPartCount() const
    {
        return mColumns->PartCount();
    }

    int
    GetTotalPartCount() const
    {
        return GetRowPartCount() * GetColumnPartCount();
    }

    int
    GetStartNode() const
    {
        return mStartNode;
    }

    Table::QuidTable*
    GetRowQuids()
    {
        return mRowQuids;
    }

    Table::QuidTable*
    GetColumnQuids()
    {
        return mColumnQuids;
    }

    void
    CopyQuidsFrom(DistributedAccess* arg)
    {
        Copy(mRowQuids, arg->mRowQuids);
        Copy(mColumnQuids, arg->mColumnQuids);
    }

    int
    GetRowStartOfPart(int rowPartIdx) const
    {
        return mRows->PartToIndex(rowPartIdx);
    }

    int
    GetRowEndOfPart(int rowPartIdx)
    {
        return mRows->PartToIndex(rowPartIdx + 1);
    }

    int
    GetColumnStartOfPart(int columnPartIdx)
    {
        return mColumns->PartToIndex(columnPartIdx);
    }

    int
    GetColumnEndOfPart(int columnPartIdx)
    {
        return mColumns->PartToIndex(columnPartIdx + 1);
    }

    int
    GetPartNr(int rowpart, int columnpart)
    {
        if (rowpart >= GetRowPartCount() || rowpart < 0 ||
            columnpart >= GetColumnPartCount() || columnpart < 0)
        {
           ILOG_ERROR_NODE("GetPartNr: invalid args: r=" << rowpart << 
                           " c=" << columnpart);
        }
        int partNr = rowpart * GetColumnPartCount() + columnpart;
        return partNr;
    }

    int
    GetOwnerOfPart(int partNr)
    {
        int totalParts = GetRowPartCount() * GetColumnPartCount();
        return mStartNode + ((partNr * mNodeCount) / totalParts);
    }

    int
    GetRowPartOfPart(int partNr)
    {
        return partNr / GetColumnPartCount();
    }

    int
    GetColumnPartOfPart(int partNr)
    {
        return partNr % GetColumnPartCount();
    }

    VirtualMatrix*
    GetPart(int partNr)
    {
        if (GetOwnerOfPart(partNr) != Link::Mpi::MyId())
        {
            ILOG_ERROR_NODE("GetPart: I am not the owner of part " << partNr);
            return 0;
        }
        int index = 0;
        for (int i=partNr-1 ; i>=0 ; --i)
        {
            if (GetOwnerOfPart(i) != Link::Mpi::MyId())
                break;
            index++;
        }
        if ((index < 0) || (index >= mParts.size()))
        {
            ILOG_ERROR_NODE("GetPart: cannot find part " << partNr <<
                            " (index = " << index << ")");
            return 0;
        }
        return mParts[index];
    }

    Matrix::Mat*
    StealPart()
    {
        if (mParts.size() != 1)
        {
            ILOG_ERROR("[StealPart] Only works with one part");
            return 0;
        }
        Matrix::Mat* m = mParts[0]->GetStorage();
        if (!m)
        {
            ILOG_ERROR("[StealPart] Couldn't steal from VirtualMatrix");
            return 0;
        }
        mParts.clear();
        return m;
    }

    void
    AddPart(VirtualMatrix* part)
    {
        mParts.push_back(part);
    }

    void
    AddFeature(CString feature, double weight, double average)
    {
        mFeatures.push_back(feature);
        mWeights.push_back(weight);
        mAverages.push_back(average);
    }

    int
    GetNrFeatures() const
    {
        return mFeatures.size();
    }

    String
    GetFeature(int i) const
    {
        return mFeatures[i];
    }

    double
    GetWeight(int i) const
    {
        return mWeights[i];
    }

    double
    GetTotalWeight() const
    {
        int total = 0;
        for (int i=0 ; i<GetNrFeatures() ; i++)
            total += GetWeight(i);
        return total;
    }

    double
    GetAverage(int i) const
    {
        return mAverages[i];
    }

    void
    SetAverage(int i, double value)
    {
        if ((i >= 0) && (i < mAverages.size()))
            mAverages[i] = value;
        else
            ILOG_ERROR("SetAverage: index " << i << " out of range");
    }

    // Indicates whether this has its own averages or whether they come from
    // a development set.
    bool
    GetHasOwnAverages() const
    {
        return mHasOwnAverages;
    }

    void
    SetHasOwnAverages(bool flag)
    {
        mHasOwnAverages = flag;
    }

    void
    CopyFeaturesFrom(DistributedAccess* arg)
    {
        mFeatures = arg->mFeatures;
        mWeights = arg->mWeights;
        mAverages = arg->mAverages;
        mHasOwnAverages = arg->mHasOwnAverages;
    }

    void
    Subscribe()
    {
        Util::PropertySet cmd;
        cmd.Add("cmd","subscribe");
        for (int i=mStartNode ; i<mStartNode+mNodeCount ; ++i)
            Link::Mpi::SendString(cmd.GetDescription(), i);
    }

    void
    Unsubscribe()
    {
        Util::PropertySet cmd;
        cmd.Add("cmd","unsubscribe");
        for (int i=mStartNode ; i<mStartNode+mNodeCount ; ++i)
            Link::Mpi::SendString(cmd.GetDescription(), i);
    }

    int
    GetColumn(int index, double* buffer, int bufferlength)
    {
        if (Symmetric())
            return GetRow(index, buffer, bufferlength);

        Util::PropertySet cmd;
        cmd.Add("cmd", "column");
        cmd.Add("arg", index);
        int columnPart = mColumns->IndexToPart(index);
        std::vector<int> parts;
        for (int i=0 ; i<GetRowPartCount() ; i++)
           parts.push_back(GetPartNr(i, columnPart));
        int msglength = GetData(buffer, bufferlength, cmd, parts);
        return msglength;
    }

    int
    GetRow(int index, double* buffer, int bufferlength)
    {
        Util::PropertySet cmd;
        cmd.Add("cmd", "row");
        cmd.Add("arg", index);
        int rowPart = mRows->IndexToPart(index);
        std::vector<int> parts;
        for (int i=0 ; i<GetColumnPartCount() ; i++)
           parts.push_back(GetPartNr(rowPart, i));
        int msglength = GetData(buffer, bufferlength, cmd, parts);
        return msglength;
    }

    int
    GetDiagonal(double* buffer, int bufferlength)
    {
        Util::PropertySet cmd;
        cmd.Add("cmd", "diagonal");
        std::vector<int> parts;
        for (int i=0 ; i<GetRowPartCount() ; ++i)
           parts.push_back(GetPartNr(i, i));
        int msglength = GetData(buffer, bufferlength, cmd, parts);
        return msglength;
    }

    void
    StartEventLoop()
    {
        int subscribers = 0;
        do
        {
            int source;
            String messageString = Link::Mpi::ReceiveString(source);
            Util::PropertySet message(messageString);
            String cmd = message.GetString("cmd");
            ILOG_DEBUG_NODE("received cmd: " << messageString);
            if (cmd == "subscribe")
            {
                subscribers++;
                ILOG_DEBUG_NODE("subscription added: id=" << source <<
                                " new #=" << subscribers);
                continue;
            }
            if (cmd == "unsubscribe")
            {
                subscribers--;
                ILOG_DEBUG_NODE("subscription removed: #=" << subscribers);
                continue;
            }
            if ((cmd != "diagonal") && (cmd != "row") && (cmd != "column"))
                ILOG_ERROR("unknown command: " << cmd);
            int part = message.GetInt("part");
            int index = message.GetInt("arg");
            Vector64 v = GetDataFromPart(cmd, part, index);
            Link::Mpi::SendData(v.GetData(), v.Size(), source);
        }
        while (subscribers > 0);
    }

    void
    Dump(int cornerWidth = 4, int cornerHeight = 4)
    {
        std::cout << "Dumping DistributedAccess" << std::endl;
        std::cout << "rows x columns = " << GetRows() << " x " << GetColumns()
                  << ", parts = " << GetRowPartCount() << " x "
                  << GetColumnPartCount() << std::endl;
        for (int i=0 ; i<mFeatures.size() ; i++)
        {
            std::cout << "feature " << i << " = " << mFeatures[i]
                      << " weight=" << mWeights[i] << " average="
                      << mAverages[i] << std::endl;
        }
        int id = Link::Mpi::MyId();
        std::cout << "my id = " << id << std::endl;
        if (mParts.size() == 0)
        {
            std::cout << "no parts loaded" << std::endl;
            return;
        }

        for (int r=0 ; r<GetRowPartCount() ; r++)
        {
            for (int c=0 ; c<GetColumnPartCount() ; c++)
            {
                int i = GetPartNr(r,c);
                std::cout << "r=" << r << ", c=" << c << ", part=" << i
                          << ", owner=" << GetOwnerOfPart(i);
                VirtualMatrix* mat = GetPart(i);
                if (mat)
                    std::cout << ", part rows x cols = " << mat->NrRow()
                              << " x " << mat->NrCol() << std::endl;
                    else
                        std::cout << ", part=0" << std::endl;
            }
        }

        int rowSize = GetColumns();
        std::cout << "rowSize = " << rowSize << std::endl;
        double* rowBuf = new double[rowSize];
        for (int y=0 ; y<GetRows() ; y++)
        {
            if ((y < cornerHeight) || (y >= (GetRows()-cornerHeight)))
            {
                int res = GetRow(y, rowBuf, rowSize);
                if (res != rowSize)
                {
                    ILOG_ERROR("[Dump2]: row=" << y << ", received " << res <<
                               " instead of " << rowSize);
                }
                std::cout << "row = " << y << " :- ";
                for (int x=0 ; x<rowSize ; x++)
                {
                    if ((x < cornerWidth) || (x >= (rowSize-cornerWidth)))
                    {
                        std::cout << rowBuf[x] << ", ";
                    }
                    else
                    {
                        if (x == cornerWidth)
                            std::cout << "..., ";
                    }
                }
                std::cout << std::endl;
            }
            else
            {
                if (y == cornerHeight)
                    std::cout << "................." << std::endl;
            }
        }
        delete rowBuf;
        std::cout << "Dump done" << std::endl;
    }

private:

    // Specialiation of VirtualMatrix
    virtual bool
    HasGetRowImplReal64()
    {
        return true;
    }

    // Specialiation of VirtualMatrix
    virtual int
    GetRowImpl(int rowNr, Real64* buffer, int bufferSize)
    {
        return GetRow(rowNr, buffer, bufferSize);
    }

    // Specialiation of VirtualMatrix
    virtual int
    GetDiagonalImpl(Real64* buffer, int bufferSize)
    {
        return GetDiagonal(buffer, bufferSize);
    }

    int
    GetData(double* buffer, int bufferlength, Util::PropertySet& cmd,
            const std::vector<int>& parts)
    {
        int received=0;
        for (int i=0 ; i<parts.size(); i++)
        {
            if (Link::Mpi::MpiUsed())
            {
                int owner = GetOwnerOfPart(parts[i]);
                cmd.Add("part", parts[i]);
                Link::Mpi::SendString(cmd.GetDescription(), owner);
                double* b = buffer + received;
                int l = Link::Mpi::ReceiveData(b, bufferlength-received, owner);
                received += l;
            }
            else
            {
                Vector64 v = GetDataFromPart(cmd.GetString("cmd"), parts[i],
                                             cmd.GetInt("arg"));
                double* b = buffer + received;
                if (v.Size() > bufferlength-received)
                    ILOG_ERROR("[GetData] buffer too small");
                for (int j=0 ; j<v.Size() ; j++)
                    b[j] = v.Elem(j);
                received += v.Size();
            }
        }
        return received;
    }

    Vector64
    GetDataFromPart(CString cmd, int part, int index)
    {
        Vector64 v;
        if (cmd == "diagonal")
            v = GetDiagonalFromPart(part);
        if (cmd == "column")
            v = GetColumnFromPart(part, index);
        if (cmd == "row")
            v = GetRowFromPart(part, index);
        return v;
    }

    Vector64
    GetColumnFromPart(int part, int index)
    {
        Vector64 v;
        int colPart = mColumns->IndexToPart(index);
        index -= mColumns->PartToIndex(colPart);
        ILOG_ERROR("[GetColumnFromPart] Not supported");
        return v;
    }

    Vector64
    GetRowFromPart(int part, int index)
    {
        Vector64 v;
        int rowPart = mRows->IndexToPart(index);
        index -= mRows->PartToIndex(rowPart);
        if (mParts.size() != 0)
            v = GetPart(part)->GetRow(index);
        return v;
    }

    Vector64
    GetDiagonalFromPart(int part)
    {
        Vector64 v;
        if (mParts.size() != 0)
            v = GetPart(part)->GetDiagonal();
        return v;
    }

    IndexConverter*             mRows;
    IndexConverter*             mColumns;
    int                         mStartNode;
    int                         mNodeCount;
    Table::QuidTable*           mRowQuids;
    Table::QuidTable*           mColumnQuids;
    std::vector<VirtualMatrix*> mParts;
    std::vector<String>         mFeatures;
    std::vector<double>         mWeights;
    std::vector<double>         mAverages;
    bool                        mHasOwnAverages;

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT(DistributedAccess, Impala.Core.Matrix);


} // namespace Matrix
} // namespace Core
} // namespace Impala

#endif

---