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

---

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/Array/ReadRaw.h"
#include "Link/Mpi/MpiFuncs.h"
#include "Util/PropertySet.h"
#include "Util/IOBufferFile.h"
#include "Core/Matrix/MatTranspose.h"
#include "Core/Matrix/GetDiagonal.h"
#include "Core/Matrix/GetColumn.h"
#include "Core/Vector/VectorTem.h"
#include "Core/Table/QuidTable.h"
#include "Core/Table/Read.h"
#include "Core/Table/Write.h"
#include "Core/Feature/FeatureTable.h"

#include "Core/Matrix/IDistributedAccess.h"

namespace Impala
{
namespace Core
{
namespace Matrix
{

class DistributedAccess : public IDistributedAccess
{
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);
        }
        
        inline int PartToIndex(int part)
        {
            return Min(part * mBlockSize, mTotalSize);
        }
        
        inline 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;
        }
        
        inline int PartCount()
        {
            return mPartCount;
        }
        
        inline int TotalSize()
        {
            return mTotalSize;
        }
    private:
        const int mTotalSize;
        const int mPartCount;
        int mBlockSize;
    };

    DistributedAccess(String filenamebase, Core::Database::RawDataSet* set, 
        Core::Database::RawDataSet* set2, int startNode, int nodeCount)
        : mRows(0), mColumns(0)
    {
        mValid = false;
        mStartNode = startNode;
        mNodeCount = nodeCount;
        bool read = ReadInfoFile(filenamebase, set, set2);
        if(!read)
        {
            ILOG_ERROR("failed to read info file");
        }
        else
        {
            LoadQuids(filenamebase, set, set2);
            LoadParts(filenamebase, set, set2);
        }
    }

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

    bool Valid()
    {
        return mValid;
    }

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

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

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

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

    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)
    {
        // for all pieces that contain that column, broadcast it.
        ILOG_DEBUG_NODE("*** GetColumn " << index);
        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<mRows->PartCount() ; ++i)
           parts.push_back(GetPartNr(i, columnPart));
        int msglength = GetData(buffer, bufferlength, cmd, parts);
        //ILOG_DEBUG_NODE("GetColumn done, received " << msglength << " elements");
        return msglength;
    }

    int GetRow(int index, double* buffer, int bufferlength)
    {
        // for all pieces that contain that row, broadcast it.
        ILOG_DEBUG_NODE("*** GetRow " << index);
        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<mColumns->PartCount() ; ++i)
           parts.push_back(GetPartNr(rowPart, i));
        int msglength = GetData(buffer, bufferlength, cmd, parts);
        //ILOG_DEBUG_NODE("GetRow done, received " << msglength << " elements");
        return msglength;
    }

    int GetDiagonal(double* buffer, int bufferlength)
    {
        Util::PropertySet cmd;
        cmd.Add("cmd","diagonal");
        std::vector<int> parts;
        for(int i=0 ; i<mRows->PartCount() ; ++i)
           parts.push_back(GetPartNr(i, i));
        int msglength = GetData(buffer, bufferlength, cmd, parts);
        ILOG_DEBUG_NODE("GetDiagonal done, received " << msglength << " elements");
        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");
            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");
            ILOG_DEBUG_NODE("received cmd: " << messageString);
            Vector::VectorTem<double> v;
            Matrix::Mat* m = GetPart(part);
            if(cmd == "diagonal")
                v = Matrix::GetDiagonal(m);
            if(cmd == "column")
            {
                int column = message.GetInt("arg");
                int columnpart = mColumns->IndexToPart(column);
                column -= mColumns->PartToIndex(columnpart);
                if(column < 0 || column >= m->H())
                    ILOG_ERROR_NODE("faulty column index: " << column <<
                            ". column = " << message.GetInt("arg") << "part = " << part);
                v = Vector::VectorTem<double>(m->W(), m->CPB(0,column), true);
            }
            if(cmd == "row")
            {
                int row = message.GetInt("arg");
                int rowpart = mRows->IndexToPart(row);
                row -= mRows->PartToIndex(rowpart);
                v = Matrix::GetColumn(m, row); // the parts are mirrored, remember?
            }
            //ILOG_DEBUG_NODE("sending answer");
            Link::Mpi::SendData(v.GetData(), v.Size(), source);
            //ILOG_DEBUG_NODE("data sent");
        } while(subscribers>0);
        //ILOG_INFO("EventLoop returned");
    }

    void CopyParts(Matrix::Mat* dst)
    {
        for(int i=0 ; i<mColumns->PartCount()*mRows->PartCount() ; ++i)
        {
            Matrix::Mat* part = GetPart(i);
            if(part)
            {
                // transpose because of column acces optimization (see LoadParts)
                Matrix::Mat* t = MatTranspose(part);
                Array::SetPart(dst, t, mColumns->PartToIndex(i%mColumns->PartCount()),
                                       mRows->PartToIndex(i/mColumns->PartCount()));
                delete t;
            }
        }
    }
private:
    bool ReadInfoFile(String filenamebase, Core::Database::RawDataSet* set, 
        Core::Database::RawDataSet* set2)
    {
        String description;
        if(Link::Mpi::MyId() == 0)
        {
            Util::Database* db;
            String filename = filenamebase + ".info";
            if (set2)
            {
                filename = set2->GetFilePathPrecomputedKernels(filename,
                                                               set->GetSetName(),
                                                               false, false);
                db = set2->GetDatabase();
            }
            else
            {
                filename = set->GetFilePathPrecomputedKernels(filename,
                                                              "", false, false);
                db = set->GetDatabase();
            }
            Util::IOBuffer* buf = db->GetIOBuffer(filename, true, true, "");
            if (!buf)
            {
                ILOG_ERROR("could not open " << filenamebase << ".info");
                return false;
            }
            description = buf->ReadLine();
            delete buf;
        }
        description = Link::Mpi::BroadcastString(description);
        ILOG_DEBUG("description  = " << description);
        Util::PropertySet properties(description);
        // read the properties
        mRows = new IndexConverter(properties.GetInt("totalrows"), properties.GetInt("rowparts"));
        mColumns = new IndexConverter(properties.GetInt("totalcolumns"), properties.GetInt("columnparts"));
        //ILOG_DEBUG_NODE("ReadInfo done: " << mTotalColumns << " " << mColumnParts);
        return true;
    }

    void LoadParts(String filenamebase, Core::Database::RawDataSet* set, 
        Core::Database::RawDataSet* set2)
    {
        ILOG_DEBUG_NODE("starting LoadParts");
        int id = Link::Mpi::MyId();
        for(int r=0 ; r<mRows->PartCount() ; r++)
        {
            for(int c=0 ; c<mColumns->PartCount() ; c++)
            {
                int i = GetPartNr(r,c);
                if(GetOwnerOfPart(i) == id)
                {
                    Matrix::Mat* m = 0;
                    String filename = filenamebase + ".precomputed.part-R" +
                               MakeString(r) + "-C" + MakeString(c) + ".raw";
                    if (set2)
                    {
                        filename = set2->GetFilePathPrecomputedKernels
                            (filename, set->GetSetName(), false, false);
                        ReadRaw(m, filename, set2->GetDatabase());
                    }
                    else
                    {
                        filename = set->GetFilePathPrecomputedKernels
                            (filename, "", false, false);
                        ReadRaw(m, filename, set->GetDatabase());
                    }
                    if(!m)
                    {
                        ILOG_ERROR_NODE("did not load matrix");
                        exit(0); 
                        // exiting one node makes the whole thing crash,
                        // this is intentionally
                    }
                    //ILOG_DEBUG_NODE("loaded matrix r" << r << "c" << c);
                    mParts.push_back(MatTranspose(m));
                    delete m;
                }
            }
        }
        mValid = true;
    }

    void LoadQuids(String filenamebase, Core::Database::RawDataSet* set, 
        Core::Database::RawDataSet* set2)
    {
        //ILOG_DEBUG_NODE("starting LoadQuids");
        mRowQuids = new Table::QuidTable();
        mColumnQuids = new Table::QuidTable();
        int id = Link::Mpi::MyId();
        if(id == 0)
        {
            Util::Database* db = (set2) ? set2->GetDatabase()
                                        : set->GetDatabase();
            for(int r=0 ; r<mRows->PartCount() ; ++r)
            {
                ILOG_DEBUG_NODE("loading row quids part " << r);
                Table::QuidTable* temp = new Table::QuidTable();
                String filename = filenamebase + ".rowindices" + MakeString(r)
                    + "of" + MakeString(mRows->PartCount()) + ".quids";
                if (set2)
                {
                    filename = set2->GetFilePathPrecomputedKernels
                        (filename, set->GetSetName(), false, false);
                }
                else
                {
                    filename = set->GetFilePathPrecomputedKernels
                        (filename, "", false, false);
                }
                Read(temp, filename, db);
                mRowQuids->Append(temp);
                delete temp;
            }
            for (int c=0 ; c<mColumns->PartCount() ; ++c)
            {
                ILOG_DEBUG_NODE("loading column quids part " << c);
                Table::QuidTable* temp = new Table::QuidTable();
                String filename = filenamebase + ".columnindices" + MakeString(c)
                    + "of" + MakeString(mColumns->PartCount()) + ".quids";
                if (set2)
                {
                    filename = set2->GetFilePathPrecomputedKernels
                        (filename, set->GetSetName(), false, false);
                }
                else
                {
                    filename = set->GetFilePathPrecomputedKernels
                        (filename, "", false, false);
                }
                Read(temp, filename, db);
                mColumnQuids->Append(temp);
                delete temp;
            }
            if (mRowQuids->Size() != mRows->TotalSize())
                ILOG_ERROR_NODE("number of row quids doesn't match totalrows "
                                << mRows->TotalSize() << " != "
                                << mRowQuids->Size());
            if (mColumnQuids->Size() != mColumns->TotalSize())
                ILOG_ERROR_NODE("number of column quids doesn't match total "
                                << mColumns->TotalSize() << " != "
                                << mColumnQuids->Size());
            Util::IOBuffer* buf = new Util::IOBuffer(mRows->TotalSize()*
                                                     sizeof(Quid) + 0x1000);
            Write(mRowQuids, buf, true);
            //ILOG_DEBUG_NODE("broadcasting...");
            Broadcast(buf);
            //ILOG_DEBUG_NODE("done");
            delete buf;
            buf = new Util::IOBuffer(mColumns->TotalSize()*sizeof(Quid) + 0x1000);
            Write(mColumnQuids, buf, true);
            Broadcast(buf);
            delete buf;
        }
        else
        {
            Util::IOBuffer* buf = new Util::IOBuffer();
            //ILOG_DEBUG_NODE("broadcasting...");
            Broadcast(buf);
            //ILOG_DEBUG_NODE("done");
            if (buf && buf->Valid())
            {
                Read(mRowQuids, buf);
                //ILOG_DEBUG_NODE("rows received");
                delete buf;
            }
            buf = new Util::IOBuffer();
            Broadcast(buf);
            if (buf && buf->Valid())
            {
                Read(mColumnQuids, buf);
                //ILOG_DEBUG_NODE("columns received");
                delete buf;
            }
            if(mRowQuids->Size() != mRows->TotalSize())
                ILOG_ERROR_NODE("number of row quids doesn't match totalrows");
            if(mColumnQuids->Size() != mColumns->TotalSize())
                ILOG_ERROR_NODE("number of column quids doesn't match totalcolumns");
        }
    }

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

    int GetOwnerOfPart(int partNr)
    {
        int totalParts = mRows->PartCount() * mColumns->PartCount();
        return mStartNode + ((partNr * mNodeCount) / totalParts);
    }

    Matrix::Mat* GetPart(int partNr)
    {
        if(GetOwnerOfPart(partNr) != Link::Mpi::MyId())
            return 0;
        int index = 0;
        for(int i=partNr-1 ; i>=0 ; --i)
        {
            if(GetOwnerOfPart(i) != Link::Mpi::MyId())
                break;
            ++index;
        }
        return mParts[index];
    }

    int GetData(double* buffer, int bufferlength, Util::PropertySet& cmd,
                const std::vector<int>& parts)
    {
        //ILOG_DEBUG_NODE("GetData: bufferlength = " << bufferlength << ", received = " << 0);
        int received=0;
        for(int i=0 ; i<parts.size(); ++i)
        {
            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;
            //ILOG_DEBUG_NODE("GetData: bufferlength = " << bufferlength << ", received = " << received);
        }
        return received;
    }

    bool mValid;
    IndexConverter* mRows;
    IndexConverter* mColumns;
    int mStartNode, mNodeCount;
    std::vector<Matrix::Mat*> mParts;
    Table::QuidTable* mColumnQuids;
    Table::QuidTable* mRowQuids;

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT(DistributedAccess, Impala.Core.Matrix);


Feature::FeatureTable* LoadDistributedMatrix(DistributedAccess& access)
{
    Table::QuidTable* quids = access.GetRowQuids();
    int rows = access.GetRows();
    int columns = access.GetColumns();
    Matrix::Mat* m = new Matrix::Mat(rows, columns, 0, 0);
    access.CopyParts(m);

    Feature::FeatureDefinition fdef("precomputed");
    Feature::FeatureTable* f = new Feature::FeatureTable(fdef, rows, columns);
    f->GetColumn2()->SetStorage(m);
    Copy(f->GetColumn1(), quids->GetColumn1(), rows, 0, 0);
    f->SetSize(rows);
    return f;
}


Feature::FeatureTable* LoadDistributedMatrix(String name, 
                                             Core::Database::RawDataSet* set,
                                             Core::Database::RawDataSet* set2)
{
    ILOG_VAR(Core.Matrix.LoadDistributedMatrix);
    DistributedAccess access(name, set, set2, Link::Mpi::MyId(), 1);
    if(!access.Valid())
    {
        ILOG_ERROR("could not open distributed matrix \"" << name << "\"");
        return 0;
    }
    return LoadDistributedMatrix(access);
}

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

#endif

---