GDP

#include "gdpfs_tf.h"

#include <unistd.h>

#include <stdio.h>

#include <errno.h>

#include <string.h>

#include <sys/stat.h>

#include <cstdlib>

#include <string>

#include <iostream>

#include <map>

#include "tensorflow/core/lib/core/error_codes.pb.h"

#include "tensorflow/core/lib/core/status.h"

#include "tensorflow/core/lib/strings/strcat.h"

#include "tensorflow/core/platform/env.h"

#include "tensorflow/core/platform/file_system_helper.h"

#include "tensorflow/core/platform/logging.h"

#include "gdpfs.h"

namespace tensorflow {

static EP_DBG Dbg = EP_DBG_INIT("gdpfs.tf",

                                                "GDP FileSystem: Tensorflow Extensions");

/*

**        First, we implement a few classes that conform to

**  the behavior expected by the filesystem users.

**

**  These classes are thin wrappers around the classes

**  provided by the gdpfs. Note that the WritableFile

**  and AppendableFile are implemented by the same class.

*/

/*

** RandomAccessFile implementation based on GDPFileRO

** (provided by gdpfs).

*/

class GDPRandomAccessFile : public RandomAccessFile

{

        public:

        GDPRandomAccessFile(const string& fname)

        {

                ep_dbg_cprintf(Dbg, 4, "New RandomAccessFile(%s)\n",

                                                                                        fname.c_str());

                _myname = fname;

                _file = new GDPFileRO(fname);

        }

        ~GDPRandomAccessFile() override { delete _file; }

        Status Read(uint64 offset, size_t n, StringPiece* result,

                                                        char* scratch) const override

        {

                ep_dbg_cprintf(Dbg, 24, "Reading [%lld +(%ld)] bytes\n", offset, n);

                size_t read = _file->Read(offset, n, scratch);

                *result = StringPiece(scratch, read);

                if (read == n)

                        return Status::OK();

                else

                {

                        ep_dbg_cprintf(Dbg, 11,

                                                        "Read only %ld bytes (asked for %ld)\n", read, n);

                        return Status(error::OUT_OF_RANGE,

                                                        "Read less bytes than requested");

                }

        }

        private:

        GDPFileRO* _file;

        string _myname;

};

/*

** WritableFile implementation based on GDPFileWO

** (provided by gdpfs).

*/

class GDPWritableFile : public WritableFile

{

        public:

        GDPWritableFile(const string& fname)

        {

                ep_dbg_cprintf(Dbg, 4, "New WritableFile(%s)\n",

                                                                                        fname.c_str());

                _myname = fname;

                _file = new GDPFileWO(fname);

        }

        ~GDPWritableFile() override { delete _file; }

        Status Append(const StringPiece& data) override

        {

                string s(data.data(), data.size());

                _file->Append(s);

                return Status::OK();

        }

        Status Close() override

        { 

                ep_dbg_cprintf(Dbg, 6, "Close(%s)\n", _myname.c_str());

                return Status::OK();

        }

        Status Flush() override

        {

                ep_dbg_cprintf(Dbg, 6, "Flush(%s)\n", _myname.c_str());

                return Status::OK();

        }

        Status Sync() override

        {

                ep_dbg_cprintf(Dbg, 6, "Sync(%s)\n", _myname.c_str());

                return Status::OK();

        }

        private:

        GDPFileWO* _file;

        string _myname;

};

/*

** MemoryMapped implementation based on GDPFileROMemMap

** (provided by gdpfs).

*/

class GDPReadOnlyMemoryRegion : public ReadOnlyMemoryRegion

{

        public:

        GDPReadOnlyMemoryRegion(const string& fname)

        {

                ep_dbg_cprintf(Dbg, 4, "New ReadOnlyMemoryRegion(%s)\n",

                                                                                        fname.c_str());

                _myname = fname;

                _file = new GDPFileROMemMap(fname);

        }

        ~GDPReadOnlyMemoryRegion() override { delete _file; }

        const void* data() override

        {

                ep_dbg_cprintf(Dbg, 9, "%s.data()\n", _myname.c_str());

                return (const char*) _file->data();

        }

        uint64 length() override

        {

                ep_dbg_cprintf(Dbg, 9, "%s.length()\n", _myname.c_str());

                return (uint64) _file->length();

        }

        private:

        GDPFileROMemMap* _file;

        string _myname;

};

/**********************************************************/

/******** Implementation of GDPFileSystem below ***********/

/**********************************************************/

GDPFileSystem::GDPFileSystem()

{

        // nothing to see here.

        GDPfsInit(GDP_MODE_RA);

        ep_dbg_cprintf(Dbg, 2, "Initializing file system\n");

}

GDPFileSystem::~GDPFileSystem()

{

        ep_dbg_cprintf(Dbg, 2, "Freeing file system\n");

        // we close the directories in _open_root_dirs

        std::map<string, GDPDir*>::iterator it;

        for (it=_open_root_dirs.begin(); it!=_open_root_dirs.end(); it++)

        {

                ep_dbg_cprintf(Dbg, 15, "Freeing up open dir %s\n",

                                                                                        it->first.c_str());

                delete it->second;

        }

}

GDPDir* GDPFileSystem::get_root_dir(const string& dirname)

{

        ep_dbg_cprintf(Dbg, 14, "get_root_dir(%s)\n", dirname.c_str());

        std::map<string, GDPDir*>::iterator it;

        it = _open_root_dirs.find(dirname);

        if (it == _open_root_dirs.end())

        {

                // open the directory, and return the pointer

                // dirname is also the logname for the topmost

                // directory; this is intentional.

                ep_dbg_cprintf(Dbg, 14, "Opening log/dir %s\n", dirname.c_str());

                GDPDir *ptr = new GDPDir(dirname, GDP_MODE_RA);

                _open_root_dirs[dirname] = ptr;

                return ptr;

        }

        else

        {

                ep_dbg_cprintf(Dbg, 14, "Returning alredy open log/dir %s\n",

                                                                                                        dirname.c_str());

                return it->second;

        }

}

Status GDPFileSystem::NewRandomAccessFile(

                const string& fname,

                std::unique_ptr<RandomAccessFile>*result)

{

        ep_dbg_cprintf(Dbg, 4, "NewRandomAccessFile(%s)\n", fname.c_str());

        string rootlog, path, logname;

        GDPDir *ptr;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        if (ptr->NameExists(path))

        {

                ptr->GetEntryLogname(path, &logname);

                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] => [%s]\n",

                                                                rootlog.c_str(), path.c_str(),

                                                                logname.c_str());

                result->reset(new GDPRandomAccessFile(logname));

                return Status::OK();

        }

        else

        {

                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Does NOT Exist\n",

                                                                rootlog.c_str(), path.c_str());

                result->reset(NULL);

                return errors::NotFound("File doesn't exist\n");

        }

}

Status GDPFileSystem::NewReadOnlyMemoryRegionFromFile(

                                const string& fname,

                                std::unique_ptr<ReadOnlyMemoryRegion>* result)

{

        ep_dbg_cprintf(Dbg, 4, "NewReadOnlyMemoryRegionFromFile(%s)\n",

                                                                                                        fname.c_str());

        GDPDir *ptr;

        string rootlog, path, logname;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        if (ptr->NameExists(path))

        {

                ptr->GetEntryLogname(path, &logname);

                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] => [%s]\n",

                                                                rootlog.c_str(), path.c_str(),

                                                                logname.c_str());

                result->reset(new GDPReadOnlyMemoryRegion(logname));

                return Status::OK();

        }

        else

        {

                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Does NOT Exist\n",

                                                                rootlog.c_str(), path.c_str());

                result->reset(NULL);

                return errors::NotFound("File doesn't exist\n");

        }

}

Status GDPFileSystem::NewWritableFile(

                const string& fname,

                std::unique_ptr<WritableFile>* result)

{

        ep_dbg_cprintf(Dbg, 4, "NewWritableFile(%s)\n", fname.c_str());

        GDPDir *ptr;

        string rootlog, path, logname;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        if (ptr->NameExists(path))

        {

                ep_dbg_cprintf(Dbg, 7, "Opening existing log as writable file\n");

                ptr->GetEntryLogname(path, &logname);

        }

        else

        {

                ep_dbg_cprintf(Dbg, 7, "Creating new log as writable file\n");

                ptr->NewFile(path, &logname);

        }

        result->reset(new GDPWritableFile(logname));

        ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] => [%s]\n",

                                                                rootlog.c_str(), path.c_str(),

                                                                logname.c_str());

        return Status::OK();

}

Status GDPFileSystem::NewAppendableFile(

                                const string& fname,

                                std::unique_ptr<WritableFile>* result)

{

        ep_dbg_cprintf(Dbg, 4, "NewAppendableFile(%s)\n", fname.c_str());

        return NewWritableFile(fname, result);

}

Status GDPFileSystem::FileExists(const string& fname)

{

        ep_dbg_cprintf(Dbg, 4, "FileExists(%s)\n", fname.c_str());

        string rootlog, path;

        GDPDir *ptr;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        if (ptr->NameExists(path))

        {

                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Exists\n",

                                                                rootlog.c_str(), path.c_str());

                return Status::OK();

        }

        else

        {

                ep_dbg_cprintf(Dbg, 7, "[%s]/[%s] Does NOT Exist\n",

                                                                rootlog.c_str(), path.c_str());

                return errors::NotFound("Not found\n");

        }

}

Status GDPFileSystem::GetChildren(

                                const string& dir,

                                std::vector<string>* result)

{

        ep_dbg_cprintf(Dbg, 4, "GetChildren(%s)\n", dir.c_str());

        string rootlog, path;

        GDPDir *ptr;

        std::vector<string> tmp;

        std::vector<string>::iterator it;

        string s;

        assert (ParsePath(dir, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        ptr->GetChildren(path, &tmp);

        // Make sure we translate the names back to full path

        for (it=tmp.begin(); it!=tmp.end(); it++)

        {

                s.assign("gdp://");

                s.append(rootlog);

                s.append("/");

                s.append(*it);

                result->push_back(s);

        }

        return Status::OK();

        /*

        // XXX This should be removed after testing GDPDir's GetChildren

        for (it=ptr->dentries_.begin(); it!= ptr->dentries_.end(); it++)

        {

                BaseDirName(it->first, &dirname, &basename);

                fulldirname.assign("gdp://");

                fulldirname.append(rootlog);

                fulldirname.append("/");

                fulldirname.append(dirname);

                ep_dbg_cprintf(Dbg, 15, "comparing %s with %s\n",

                                                                fulldirname.c_str(), dir.c_str());

                if (fulldirname.compare(dir) == 0)

                {

                        ep_dbg_cprintf(Dbg, 12, "child => %s\n", basename.c_str());

                        result->push_back(basename);

                }

        }

        return Status::OK();

        */

}

Status GDPFileSystem::Stat(const string& fname, FileStatistics* stat)

{

        ep_dbg_cprintf(Dbg, 4, "Stat(%s)\n", fname.c_str());

        string rootlog, path;

        GDPDir *ptr;

        GdpStat tmp;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        ptr->Stat(path, &tmp);

        stat->is_directory = tmp.is_directory;

        stat->length = tmp.length;

        stat->mtime_nsec = tmp.mtime_nsec;

        /*

        stat->is_directory = (ptr->GetEntryType(path) == GDPfs::DIR);

        if (!stat->is_directory)

        {

                GetFileSize(fname, &size);

                stat->length = size;

                // getting last update time

                ptr->GetEntryLogname(path, &logname);

                GDPFileLowLevel f(logname, GDP_MODE_RO);

                stat->mtime_nsec = f.GetMTime();

        }

        else

        {

                stat->length = 0;

                stat->mtime_nsec = TimeNS();        // XXX: fix this

        }

        */

        ep_dbg_cprintf(Dbg, 4, "Stat(%s)=%lld, %lld\n", fname.c_str(),

                                                                        stat->length, stat->mtime_nsec);

        return Status::OK();

}

Status GDPFileSystem::GetMatchingPaths(

                                const string& pattern,

                                std::vector<string>* results)

{

        ep_dbg_cprintf(Dbg, 4, "GetMatchingPaths(%s)\n", pattern.c_str());

        ep_dbg_cprintf(Dbg, 1, "Note that this isn't fully implemented yet.\n");

        string rootlog, path;

        GDPDir *ptr;

        std::vector<string> tmp;

        std::vector<string>::iterator it;

        string s;

        assert (ParsePath(pattern, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        ptr->GetMatchingPaths(path, &tmp);

        // Make sure we translate the names back to full path

        for (it=tmp.begin(); it!= tmp.end(); it++)

        {

                s.assign("gdp://");

                s.append(rootlog);

                s.append("/");

                s.append(*it);

                results->push_back(s);

        }

        return Status::OK();

        /*

        // XXX: this should be removed once GDPDir GetMatchingPaths is tested.

        if (pattern.find('*')!=string::npos)

                return errors::Unimplemented(" '*' Not yet implemented\n");

        if (pattern.find('?')!=string::npos)

                return errors::Unimplemented(" '?' Not yet implemented\n");

        if (pattern.find('[')!=string::npos)

                return errors::Unimplemented(" '[' Not yet implemented\n");

        if (pattern.find(']')!=string::npos)

                return errors::Unimplemented(" ']' Not yet implemented\n");

        if (pattern.find("\\\\")!=string::npos)

                return errors::Unimplemented(" ds Not yet implemented\n");

        size_t pos;

        string rootlog, path;

        GDPDir *ptr;

        std::map<string, string>::iterator it;

        assert (ParsePath(pattern, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        for (it=ptr->dentries_.begin(); it!=ptr->dentries_.end(); it++)

        {

                ep_dbg_cprintf(Dbg, 30, "Matching %s against: %s\n",

                                                                        path.c_str(), it->first.c_str());

                pos = it->first.find(path);

                if (pos == 0)

                {

                        ep_dbg_cprintf(Dbg, 22, "Found match %s\n", it->first.c_str());

                        results->push_back(it->first);

                }

        }

        return Status::OK();

        */

}

Status GDPFileSystem::DeleteFile(const string& fname)

{

        ep_dbg_cprintf(Dbg, 4, "DeleteFile(%s)\n", fname.c_str());

        string rootlog, path;

        GDPDir *ptr;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        ep_dbg_cprintf(Dbg, 7, "Deleting file [%s]/[%s]\n",

                                                        rootlog.c_str(), path.c_str());

        ptr->DeleteFile(path);

        return Status::OK();

}

Status GDPFileSystem::CreateDir(const string& name)

{

        ep_dbg_cprintf(Dbg, 4, "CreateDir(%s)\n", name.c_str());

        string rootlog, path;

        GDPDir *ptr;

        assert (ParsePath(name, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        ptr->CreateDir(path);

        ep_dbg_cprintf(Dbg, 7, "New dir [%s]/[%s] => [X]\n",

                                                        rootlog.c_str(), path.c_str());

        return Status::OK();

}

Status GDPFileSystem::DeleteDir(const string& name)

{

        ep_dbg_cprintf(Dbg, 4, "DeleteDir(%s)\n", name.c_str());

        string rootlog, path;

        GDPDir *ptr;

        assert (ParsePath(name, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        ep_dbg_cprintf(Dbg, 7, "Deleting dir [%s]/[%s]\n",

                                                        rootlog.c_str(), path.c_str());

        ptr->DeleteDir(path);

        return Status::OK();

}

Status GDPFileSystem::GetFileSize(const string& fname, uint64* size)

{

        ep_dbg_cprintf(Dbg, 4, "GetFileSize(%s)\n", fname.c_str());

        string rootlog, path, logname;

        GDPDir *ptr;

        assert (ParsePath(fname, &rootlog, &path) == kSuccess);

        ptr = get_root_dir(rootlog);

        assert (ptr->GetEntryType(path) == GDPfs::FILE);

        ptr->GetEntryLogname(path, &logname);

        GDPFile f(logname, GDP_MODE_RO);

        *size = (uint64) f.GetFileSize();

        return Status::OK();

}

Status GDPFileSystem::RenameFile(const string& src, const string& target)

{

        ep_dbg_cprintf(Dbg, 4, "RenameFile(%s=>%s)\n",

                                                                src.c_str(), target.c_str());

        string rootlog1, path1, rootlog2, path2;

        GDPDir *ptr;

        assert (ParsePath(src, &rootlog1, &path1) == kSuccess);

        assert (ParsePath(target, &rootlog2, &path2) == kSuccess);

        // we don't support copying across different root

        // directories.

        assert (rootlog1.compare(rootlog2) == 0);

        ptr = get_root_dir(rootlog1);

        ptr->RenameFile(path1, path2);

        return Status::OK();

}

REGISTER_FILE_SYSTEM("gdp", GDPFileSystem);

}        // namespace tensorflow