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

---

DistributedAccessRepositoryInFile.h

Go to the documentation of this file.

#ifndef Impala_Persistency_DistributedAccessRepositoryInFile_h
#define Impala_Persistency_DistributedAccessRepositoryInFile_h

#include "Persistency/RepositoryInFileSystem.h"
#include "Persistency/KernelMatrixLocator.h"
#include "Core/Matrix/DistributedAccess.h"
#include "Core/Matrix/VirtualMatrixFactory.h"
#include "Core/Array/Set.h"
#include "Core/Table/Read.h"
#include "Core/Table/Write.h"
#include "Core/Table/Copy.h"
#include "Util/StringParser.h"

namespace Impala
{
namespace Persistency
{


class DistributedAccessRepositoryInFile
{
public:

    typedef Core::Database::RawDataSet RawDataSet;
    typedef Core::Matrix::DistributedAccess DistributedAccess;

    DistributedAccessRepositoryInFile()
    {
    }

    bool
    Exists(const KernelMatrixLocator& loc)
    {
        String filename = loc.GetFeature() + ".info";
        String dir = GetDir(loc);
        Persistency::File file = RepFS().GetFile(loc, dir, filename, false,
                                                 true);
        return file.Valid();
    }

    DistributedAccess*
    Get(const KernelMatrixLocator& loc)
    {
        DistributedAccess* da = ReadInfoFile(loc);
        if (da)
        {
            LoadFeatures(da, loc);
            LoadQuids(da, loc);
            if (loc.GetDoParts() > 0)
                LoadParts(da, loc);
        }
        return da;
    }

    void
    Add(const KernelMatrixLocator& loc, DistributedAccess* da)
    {
        if (!WriteInfoFile(da, loc))
            return;
        WriteFeatures(da, loc);
        WriteQuids(da, loc);
        if (loc.GetDoParts() > 0)
        {
            if (loc.GetDoParts() == 3)
                WritePartsAsOne(da, loc);
            else
                WriteParts(da, loc);
        }
    }

    File
    ExposeFile(const KernelMatrixLocator& loc, bool toWrite)
    {
        String dir = GetDir(loc);
        String fName = loc.GetFeature() + ".info";
        bool silent = !toWrite; // if (toWrite) force creating directory
        return RepFS().GetFile(loc, dir, fName, toWrite, silent);
    }

private:

    RepositoryInFileSystem&
    RepFS()
    {
        return RepositoryInFileSystem::GetInstance();
    }

    // This function should follow KernelMatrixRepositoryInFile::GetDir
    String
    GetDir(const KernelMatrixLocator& loc)
    {
        String dir = loc.GetTopString();
        if (!loc.GetIsIndex())
            dir = FileNameConcat(dir, loc.GetWalkType());
        dir = FileNameConcat(dir, loc.GetAnnoSet());
        dir = FileNameConcat(dir, loc.GetFeatureIndexCat());
        dir = FileNameConcat(dir, loc.GetModel());
        dir = FileNameConcat(dir, loc.GetFeature());
        int inc = loc.GetIncrement();
        if (inc != 0)
            dir = FileNameConcat(dir, "inc_" + MakeString(inc));
        return dir;
    }

    DistributedAccess*
    ReadInfoFile(const KernelMatrixLocator& loc)
    {
        String description;
        if (Link::Mpi::MyId() == loc.GetStartNode())
        {
            String filename = loc.GetFeature() + ".info";
            String dir = GetDir(loc);
            Persistency::File file = RepFS().GetFile(loc, dir, filename, false,
                                                     false);
            if (!file.Valid())
                return 0;

            Util::IOBuffer* buf = file.GetReadBuffer(true, "");
            if (!buf)
            {
                ILOG_ERROR("Could not open " << loc.GetFeature() << ".info");
                return 0;
            }
            description = buf->ReadLine();
            delete buf;
        }
        description = Link::Mpi::BroadcastString(description,
                                                 loc.GetStartNode());
        Util::PropertySet properties(description);
        DistributedAccess* da = new DistributedAccess
            (properties.GetInt("totalrows"), properties.GetInt("totalcolumns"),
             properties.GetInt("rowparts"), properties.GetInt("columnparts"),
             loc.GetStartNode(), loc.GetNodeCount());
        return da;
    }

    void
    LoadFeatures(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        String dir = GetDir(loc);
        String filename = loc.GetFeature() + ".input.txt";
        File file = RepFS().GetFile(loc, dir, filename, false, true);
        if (!file.Valid())
            return;
        std::vector<String> lines;
        file.ReadStrings(std::back_inserter(lines));

        std::vector<double> weights;
        std::vector<String> features;
        for (int i=0 ; i<lines.size() ; i++)
        {
            Util::StringParser p(lines[i]);
            Real64 weight = p.GetDouble();
            String feature = p.GetString(' ', false);
            if ((weight == 0) || (feature.empty()))
            {
                ILOG_ERROR_HEADNODE("LoadFeatures: parse error in .input.txt");
                return;
            }
            weights.push_back(weight);
            features.push_back(feature);
        }
        if (features.size() == 0)
        {
            ILOG_ERROR_HEADNODE("LoadFeatures: no features found");
            return;
        }

        filename = loc.GetFeature() + ".averages.raw";
        file = RepFS().GetFile(loc, dir, filename, false, true);
        std::vector<double> averages;
        if (file.Valid())
        {
            file.ReadNative(std::back_inserter(averages));
            if (features.size() != averages.size())
            {
                ILOG_ERROR_HEADNODE("LoadFeatures: averages do not match");
                return;
            }
        }

        for (int i=0 ; i<features.size() ; i++)
        {
            if (averages.size() == 0)
                da->AddFeature(features[i], weights[i], -1);
            else
                da->AddFeature(features[i], weights[i], averages[i]);
        }
        if (averages.size() != 0)
            da->SetHasOwnAverages(true);
    }

    void
    LoadQuids(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        String dir = GetDir(loc);
        String filenamebase = loc.GetFeature();

        Core::Table::QuidTable* rowQuids = da->GetRowQuids();
        Core::Table::QuidTable* columnQuids = da->GetColumnQuids();
        if (Link::Mpi::MyId() == da->GetStartNode())
        {
            String filename = filenamebase + ".columns.tab";
            File file = RepFS().GetFile(loc, dir, filename, false, false);
            Read(columnQuids, file);
            filename = filenamebase + ".rows.tab";
            file = RepFS().GetFile(loc, dir, filename, false, true);
            if (file.Valid())
            {
                file = RepFS().GetFile(loc, dir, filename, false, false);
                Read(rowQuids, file);
            }
            else
            {   // assume square kernel
                Core::Table::Copy(rowQuids, columnQuids);
            }

            if (rowQuids->Size() != da->GetRows())
                ILOG_ERROR("number of quids doesn't match totalrows: "
                           << rowQuids->Size() << " vs " << da->GetRows());
            if (columnQuids->Size() != da->GetColumns())
                ILOG_ERROR("number of quids doesn't match totalcolumns: "
                           << columnQuids->Size() << " vs " << da->GetColumns());
            Util::IOBuffer* buf = new Util::IOBuffer(da->GetRows()*
                                                     sizeof(Quid) + 0x1000);
            Write(rowQuids, buf, true);
            Broadcast(buf, da->GetStartNode());
            delete buf;
            buf = new Util::IOBuffer(da->GetColumns()*sizeof(Quid)+0x1000);
            Write(columnQuids, buf, true);
            Broadcast(buf, da->GetStartNode());
            delete buf;
        }
        else
        {
            Util::IOBuffer* buf = new Util::IOBuffer();
            Broadcast(buf, da->GetStartNode());
            if (buf && buf->Valid())
            {
                Read(rowQuids, buf);
                delete buf;
            }
            buf = new Util::IOBuffer();
            Broadcast(buf, da->GetStartNode());
            if (buf && buf->Valid())
            {
                Read(columnQuids, buf);
                delete buf;
            }
            if (rowQuids->Size() != da->GetRows())
                ILOG_ERROR_NODE("number of quids doesn't match totalrows");
            if (columnQuids->Size() != da->GetColumns())
                ILOG_ERROR_NODE("number of quids doesn't match totalcolumns");
        }
    }

    void
    LoadParts(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        typedef Core::Matrix::VirtualMatrixFactory VirtualMatrixFactory;
        typedef Core::Matrix::VirtualMatrix VirtualMatrix;
        String dir = GetDir(loc);
        VirtualMatrixFactory& vmf = VirtualMatrixFactory::GetInstance();
        for (int r=0 ; r<da->GetRowPartCount() ; r++)
        {
            for (int c=0 ; c<da->GetColumnPartCount() ; c++)
            {
                int i = da->GetPartNr(r,c);
                if (Link::Mpi::MyId() == da->GetOwnerOfPart(i))
                {
                    String filename = loc.GetFeature() + ".precomputed.part-R" +
                               MakeString(r) + "-C" + MakeString(c) + ".raw";
                    File file = RepFS().GetFile(loc, dir, filename, false,
                                                false);
                    VirtualMatrix* vm;
                    if (loc.GetDoParts() == 1)
                        vm = vmf.ConstructMemory(file, loc.GetEmployReal32());
                    else if (loc.GetDoParts() == 2)
                        vm = vmf.ConstructIOBufferReader(file);
                    else
                        ILOG_ERROR("[LoadParts] Do not understand GetDoParts");
                    da->AddPart(vm);
                }
            }
        }
    }

    bool
    WriteInfoFile(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        if (Link::Mpi::MyId() != da->GetStartNode())
            return true;

        String filename = loc.GetFeature() + ".info";
        String dir = GetDir(loc);
        Persistency::File file = RepFS().GetFile(loc, dir, filename, true,
                                                 false);
        if (!file.Valid())
            return false;

        ILOG_INFO("Saving info in " << filename);
        Util::IOBuffer* buf = file.GetWriteBuffer();
        if (!buf)
        {
            ILOG_ERROR("Could not open " << loc.GetFeature() << ".info");
            return false;
        }
        Util::PropertySet ps;
        ps.Add("totalrows", da->GetRows());
        ps.Add("totalcolumns", da->GetColumns());
        if (loc.GetDoParts() == 3)
        {
            ps.Add("rowparts", 1);
            ps.Add("columnparts", 1);
        }
        else
        {
            ps.Add("rowparts", da->GetRowPartCount());
            ps.Add("columnparts", da->GetColumnPartCount());
        }
        ps.Print(buf);
        delete buf;
        return true;
    }

    void
    WriteFeatures(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        if (Link::Mpi::MyId() != da->GetStartNode())
            return;

        String dir = GetDir(loc);
        String filename = loc.GetFeature() + ".input.txt";
        File file = RepFS().GetFile(loc, dir, filename, true, false);
        Util::IOBuffer* buf = file.GetWriteBuffer();
        if (buf)
        {
            for (int i=0 ; i<da->GetNrFeatures() ; i++)
            {
                String s = MakeString(da->GetWeight(i)) + " " +
                    da->GetFeature(i);
                buf->Puts(s);
            }
            delete buf;
        }
        else
        {
            ILOG_ERROR("Could not write input.txt");
        }

        if (da->GetHasOwnAverages())
        {
            String filename = loc.GetFeature() + ".averages.raw";
            File file = RepFS().GetFile(loc, dir, filename, true, false);
            Util::IOBuffer* buf = file.GetWriteBuffer();
            if (buf)
            {
                for (int i=0 ; i<da->GetNrFeatures() ; i++)
                {
                    double val = da->GetAverage(i);
                    buf->Write(&val, sizeof(double));
                }
                delete buf;
            }
            else
            {
                ILOG_ERROR("Could not write averages.raw");
            }
        }
    }

    void
    WriteQuids(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        if (Link::Mpi::MyId() != da->GetStartNode())
            return;

        String dir = GetDir(loc);
        String filename = loc.GetFeature() + ".columns.tab";
        File file = RepFS().GetFile(loc, dir, filename, true, false);
        ILOG_INFO("Saving columns in " << file.GetPath());
        Write(da->GetColumnQuids(), file);

        filename = loc.GetFeature() + ".rows.tab";
        file = RepFS().GetFile(loc, dir, filename, true, false);
        ILOG_INFO("Saving rows in " << file.GetPath());
        Write(da->GetRowQuids(), file);
    }

    void
    WriteParts(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        typedef Core::Matrix::VirtualMatrixFactory VirtualMatrixFactory;
        String dir = GetDir(loc);
        VirtualMatrixFactory& vmf = VirtualMatrixFactory::GetInstance();
        for (int r=0 ; r<da->GetRowPartCount() ; r++)
        {
            for (int c=0 ; c<da->GetColumnPartCount() ; c++)
            {
                int i = da->GetPartNr(r,c);
                if (Link::Mpi::MyId() == da->GetOwnerOfPart(i))
                {
                    Core::Matrix::VirtualMatrix* part = da->GetPart(i);
                    String filename = loc.GetFeature() + ".precomputed.part-R"
                        + MakeString(r) + "-C" + MakeString(c) + ".raw" ;
                    File file = RepFS().GetFile(loc, dir, filename, true,
                                                false);
                    ILOG_INFO_NODE("Saving part in " << file.GetPath());
                    Core::Matrix::VirtualMatrix* dst = 
                        vmf.ConstructIOBufferWriter(part->NrRow(), part->NrCol(),
                                                    file, loc.GetWriteReal32(),
                                                    false);
                    dst->CopyFrom(part);
                    delete dst;
                }
            }
        }
    }

    void
    WritePartsAsOne(DistributedAccess* da, const KernelMatrixLocator& loc)
    {
        typedef Core::Matrix::VirtualMatrixFactory VirtualMatrixFactory;
        String dir = GetDir(loc);
        String filename = loc.GetFeature() + ".precomputed.part-R0-C0.raw" ;
        File file = RepFS().GetFile(loc, dir, filename, true, false);
        ILOG_INFO_NODE("Saving part in " << file.GetPath());
        VirtualMatrixFactory& vmf = VirtualMatrixFactory::GetInstance();
        Core::Matrix::VirtualMatrix* dst = 
            vmf.ConstructIOBufferWriter(da->NrRow(), da->NrCol(), file,
                                        loc.GetWriteReal32(), false);
        dst->CopyFrom(da);
        delete dst;
    }

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT(DistributedAccessRepositoryInFile, Impala.Persistency);

} // namespace Persistency
} // namespace Impala

#endif

---